From 8ebce28e34d520ae06e593372aab86f1dc11335b Mon Sep 17 00:00:00 2001
From: Denozordec <shatskiy.blog@gmail.com>
Date: Thu, 21 May 2026 10:46:13 +0700
Subject: [PATCH] perf: wire job_audit terminal persistence hook
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- TerminalHook в Registry для записи статуса job в PostgreSQL job_audit
- Подключение через BootstrapWorkers при наличии pool

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 internal/httpapi/bootstrap.go | 15 +++++++++++++++
 internal/jobs/job.go          | 23 +++++++++++++++++++++++
 internal/jobs/worker.go       |  3 +++
 3 files changed, 41 insertions(+)

diff --git a/internal/httpapi/bootstrap.go b/internal/httpapi/bootstrap.go
index ddf03c4..6359130 100644
--- a/internal/httpapi/bootstrap.go
+++ b/internal/httpapi/bootstrap.go
@@ -53,6 +53,21 @@ func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.R
 	wk := &jobs.Worker{Store: backend, HTTPClient: cdnHTTP}
 	reg := jobs.NewRegistry(wk.Process)
 	wk.Registry = reg
+	if pool != nil {
+		audit := repository.NewJobAuditWriter(pool)
+		reg.SetTerminalHook(func(j *jobs.Job) {
+			if j == nil {
+				return
+			}
+			st := j.Snapshot()
+			status, _ := st["status"].(string)
+			var errMsg *string
+			if e, ok := st["error"].(string); ok && e != "" {
+				errMsg = &e
+			}
+			audit.MarkTerminal(context.Background(), j.TenantID, j.ID, status, errMsg, time.Now().UTC())
+		})
+	}
 	observability.RegisterStoreBackend(backend)
 	return backend, reg, pool, nil
 }
diff --git a/internal/jobs/job.go b/internal/jobs/job.go
index 821b84f..0628c65 100644
--- a/internal/jobs/job.go
+++ b/internal/jobs/job.go
@@ -181,6 +181,7 @@ type Registry struct {
 	byIdempo    map[idempoKey]*Job
 	workerStart func(j *Job)
 	workerSem   chan struct{}
+	onTerminal  func(j *Job)
 }
 
 type idempoKey struct {
@@ -198,6 +199,28 @@ func NewRegistry(workerStart func(j *Job)) *Registry {
 	}
 }
 
+// SetTerminalHook registers a best-effort callback when jobs reach a terminal state.
+func (r *Registry) SetTerminalHook(fn func(j *Job)) {
+	if r == nil {
+		return
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.onTerminal = fn
+}
+
+func (r *Registry) fireTerminal(j *Job) {
+	if r == nil || j == nil {
+		return
+	}
+	r.mu.RLock()
+	fn := r.onTerminal
+	r.mu.RUnlock()
+	if fn != nil {
+		fn(j)
+	}
+}
+
 func registryMaxConcurrentJobs() int {
 	if n, err := strconv.Atoi(strings.TrimSpace(os.Getenv("EVOBGP_JOB_MAX_CONCURRENT"))); err == nil && n > 0 {
 		return n
diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go
index 3f64f30..651fa7e 100644
--- a/internal/jobs/worker.go
+++ b/internal/jobs/worker.go
@@ -82,6 +82,9 @@ func (w *Worker) httpClient() *http.Client {
 func (w *Worker) Process(j *Job) {
 	defer func() {
 		observability.RecordJobTerminal(j.Kind, j.statusLocked())
+		if w != nil && w.Registry != nil {
+			w.Registry.fireTerminal(j)
+		}
 	}()
 
 	if w == nil || w.Store == nil {