package outbox import ( "context" "encoding/json" "errors" "fmt" "log/slog" "time" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/alpyxn/varyaone/internal/backup" ) const maxAttempts = 10 const supportedSchemaVersion = 1 type Event struct { ID string Type string SchemaVersion int CompanyID string TraceID string Payload json.RawMessage Attempts int lockToken time.Time } type Handler func(context.Context, Event) error type PermanentError struct{ Err error } func (e PermanentError) Error() string { return e.Err.Error() } func (e PermanentError) Unwrap() error { return e.Err } type Worker struct { pool *pgxpool.Pool logger *slog.Logger handlers map[string]Handler auditEvents map[string]struct{} } func New(pool *pgxpool.Pool, logger *slog.Logger) *Worker { return &Worker{ pool: pool, logger: logger, handlers: map[string]Handler{}, auditEvents: knownAuditEvents(), } } func (w *Worker) Run(ctx context.Context) error { ticker := time.NewTicker(time.Second) defer ticker.Stop() for { worked, err := w.runOne(ctx) if err != nil && !errors.Is(err, context.Canceled) { w.logger.Error("outbox cycle failed", "error_class", "database") } if worked { continue } select { case <-ctx.Done(): return nil case <-ticker.C: } } } func (w *Worker) runOne(ctx context.Context) (bool, error) { // Hold the backup write barrier for the whole cycle. The worker is a writer // — it claims, acks and dead-letters rows — so a backup that pinned the // file tree without pausing it would capture a database the files no longer // match. // // The barrier is taken before claiming rather than around each statement, // so a paused worker starts no new work at all. The cost is that a backup // can wait for one in-flight event; the engine bounds that wait and reports // a timeout rather than producing an archive it cannot vouch for. barrier, err := w.pool.Acquire(ctx) if err != nil { return false, err } defer barrier.Release() if err := backup.AcquireWriteBarrier(ctx, barrier); err != nil { return false, err } defer backup.ReleaseWriteBarrier(barrier) event, found, err := w.claim(ctx) if err != nil || !found { return false, err } if event.SchemaVersion != supportedSchemaVersion { return true, w.deadLetter(ctx, event, "unknown_event") } if handler, known := w.handlers[event.Type]; known { err = handler(ctx, event) } else if _, known := w.auditEvents[event.Type]; known { return true, w.ack(ctx, event) } else { return true, w.deadLetter(ctx, event, "unknown_event") } if err == nil { return true, w.ack(ctx, event) } if isPermanent(err) || event.Attempts >= maxAttempts { return true, w.deadLetter(ctx, event, "permanent") } _, retryErr := w.pool.Exec(ctx, `UPDATE outbox_events SET locked_at=NULL,available_at=now()+($2 * interval '1 second'),last_error_class='retryable' WHERE event_id=$1 AND processed_at IS NULL AND dead_lettered_at IS NULL AND locked_at=$3`, event.ID, retryDelay(event.Attempts).Seconds(), event.lockToken) return true, retryErr } func (w *Worker) claim(ctx context.Context) (Event, bool, error) { tx, err := w.pool.BeginTx(ctx, pgx.TxOptions{}) if err != nil { return Event{}, false, err } defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }() var event Event var payload []byte err = tx.QueryRow(ctx, ` -- company_id/trace_id are nullable in the existing schema; keep old rows readable. -- If a target database predates these columns, the parent must add its migration first. SELECT event_id,type,schema_version,COALESCE(company_id::text,''),COALESCE(trace_id,''),payload,attempts+1 FROM outbox_events WHERE processed_at IS NULL AND dead_lettered_at IS NULL AND available_at<=now() AND (locked_at IS NULL OR locked_at 8 { attempt = 8 } return time.Duration(1<<(attempt-1)) * time.Second }