package repository import ( "context" "encoding/json" "time" "github.com/jackc/pgx/v5/pgxpool" ) // JobAuditWriter persists async job lifecycle rows to job_audit (optional cross-process queue foundation). type JobAuditWriter struct { pool *pgxpool.Pool } func NewJobAuditWriter(pool *pgxpool.Pool) *JobAuditWriter { if pool == nil { return nil } return &JobAuditWriter{pool: pool} } // UpsertQueued inserts a queued job row (best-effort). func (w *JobAuditWriter) UpsertQueued(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, moduleID *string, meta map[string]any) { if w == nil || w.pool == nil { return } metaJSON, _ := json.Marshal(meta) var idem any if idempotencyKey != nil && *idempotencyKey != "" { idem = *idempotencyKey } var mod any if moduleID != nil && *moduleID != "" { mod = *moduleID } _, _ = w.pool.Exec(ctx, ` INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, module_id, meta_json, created_at) VALUES ($1::uuid, $2::uuid, $3, 'queued', $4, $5::uuid, $6::jsonb, now()) ON CONFLICT (tenant_id, idempotency_key) WHERE idempotency_key IS NOT NULL DO UPDATE SET status='queued', meta_json=EXCLUDED.meta_json, module_id=EXCLUDED.module_id`, jobID, tenantID, kind, idem, mod, metaJSON) } // UpsertRunning inserts or updates a running job row (best-effort). func (w *JobAuditWriter) UpsertRunning(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, meta map[string]any) { if w == nil || w.pool == nil { return } metaJSON, _ := json.Marshal(meta) var idem any if idempotencyKey != nil && *idempotencyKey != "" { idem = *idempotencyKey } _, _ = w.pool.Exec(ctx, ` INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, meta_json, created_at, started_at) VALUES ($1::uuid, $2::uuid, $3, 'running', $4, $5::jsonb, now(), now()) ON CONFLICT (id) DO UPDATE SET status='running', started_at=COALESCE(job_audit.started_at, now()), meta_json=EXCLUDED.meta_json`, jobID, tenantID, kind, idem, metaJSON) } // MarkTerminal updates job_audit terminal state (best-effort). func (w *JobAuditWriter) MarkTerminal(ctx context.Context, tenantID, jobID, status string, errMsg *string, finishedAt time.Time) { if w == nil || w.pool == nil { return } _, _ = w.pool.Exec(ctx, ` UPDATE job_audit SET status=$3, error_message=$4, finished_at=$5 WHERE id=$1::uuid AND tenant_id=$2::uuid`, jobID, tenantID, status, errMsg, finishedAt.UTC()) }