EvoBGP/internal/repository/job_audit_claim.go

package repository

import (
	"context"
	"encoding/json"
	"time"

	"github.com/jackc/pgx/v5"
)

// ClaimedJob is a durable job_audit row claimed for in-process execution (SKIP LOCKED).
type ClaimedJob struct {
	ID             string
	TenantID       string
	Kind           string
	IdempotencyKey *string
	ModuleID       *string
	Meta           map[string]any
	CreatedAt      time.Time
}

// ReclaimStaleRunning resets orphaned running jobs older than staleAfter back to queued.
func (w *JobAuditWriter) ReclaimStaleRunning(ctx context.Context, staleAfter time.Duration) (int64, error) {
	if w == nil || w.pool == nil {
		return 0, nil
	}
	if staleAfter <= 0 {
		staleAfter = 15 * time.Minute
	}
	tag, err := w.pool.Exec(ctx, `
		UPDATE job_audit
		SET status = 'queued', started_at = NULL, error_message = NULL
		WHERE status = 'running'
		  AND started_at IS NOT NULL
		  AND started_at < now() - $1::interval`,
		staleAfter.String())
	if err != nil {
		return 0, err
	}
	return tag.RowsAffected(), nil
}

// ClaimQueued claims up to limit queued job_audit rows via FOR UPDATE SKIP LOCKED and marks them running.
// Only rows older than grace are claimed so the originating process can own fresh enqueues.
func (w *JobAuditWriter) ClaimQueued(ctx context.Context, limit int, grace time.Duration) ([]ClaimedJob, error) {
	if w == nil || w.pool == nil {
		return nil, nil
	}
	if limit <= 0 {
		limit = 8
	}
	if grace <= 0 {
		grace = 30 * time.Second
	}
	tx, err := w.pool.BeginTx(ctx, pgx.TxOptions{})
	if err != nil {
		return nil, err
	}
	defer func() { _ = tx.Rollback(ctx) }()

	rows, err := tx.Query(ctx, `
		WITH cte AS (
			SELECT id
			FROM job_audit
			WHERE status = 'queued'
			  AND created_at < now() - $2::interval
			ORDER BY created_at ASC
			FOR UPDATE SKIP LOCKED
			LIMIT $1
		)
		UPDATE job_audit j
		SET status = 'running', started_at = COALESCE(j.started_at, now())
		FROM cte
		WHERE j.id = cte.id
		RETURNING j.id::text, j.tenant_id::text, j.kind, j.idempotency_key, j.module_id::text, j.meta_json, j.created_at`,
		limit, grace.String())
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []ClaimedJob
	for rows.Next() {
		var c ClaimedJob
		var idem, mod *string
		var metaBytes []byte
		if err := rows.Scan(&c.ID, &c.TenantID, &c.Kind, &idem, &mod, &metaBytes, &c.CreatedAt); err != nil {
			return nil, err
		}
		c.IdempotencyKey = idem
		if mod != nil && *mod != "" {
			c.ModuleID = mod
		}
		meta := map[string]any{}
		if len(metaBytes) > 0 {
			_ = json.Unmarshal(metaBytes, &meta)
		}
		c.Meta = meta
		out = append(out, c)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	if err := tx.Commit(ctx); err != nil {
		return nil, err
	}
	return out, nil
}

// CountByStatus returns job_audit row counts grouped by status (best-effort metrics).
func (w *JobAuditWriter) CountByStatus(ctx context.Context) (map[string]int64, error) {
	out := map[string]int64{}
	if w == nil || w.pool == nil {
		return out, nil
	}
	rows, err := w.pool.Query(ctx, `SELECT status, count(*) FROM job_audit GROUP BY status`)
	if err != nil {
		return out, err
	}
	defer rows.Close()
	for rows.Next() {
		var status string
		var n int64
		if err := rows.Scan(&status, &n); err != nil {
			return out, err
		}
		out[status] = n
	}
	return out, rows.Err()
}