// Package audit — outbox worker (design §3.2, §3.3, §3.4).
//
// The outbox worker drains _System_AuditOutbox rows into _System_AuditLog
// field-level rows. One round per tick:
//
//  1. Acquire lease on up to BatchSize pending rows (WHERE ProcessedAt IS NULL
//     AND lease free AND PoisonClass IS NULL — the idx_unprocessed
//     index excludes any non-NULL PoisonClass, §3.3).
//  2. For each leased row: parse payload, fan out to field rows via
//     INSERT IGNORE (uk_outbox deduplicates), classify any error through
//     poison.go, then either FinalizeSuccess, MarkPoison, or BumpAttempts.
//  3. After MaxAttempts transient failures, mark the row RETRY_EXHAUSTED so
//     it stops consuming budget (§3.3 table).
//
// The worker is defined here but NOT started in backend main — wiring lands
// in a later commit; the plug point is wherever PersistenceService holds its
// *sql.DB (likely `backend/cmd/server/main.go` alongside other background
// services). A wiring commit should construct a MySQLOutboxStore (see
// outbox_store.go — lands with C7/C8) and call (*OutboxWorker).Run in a
// goroutine with a parent ctx bound to server shutdown.
//
// Field-expansion of the payload lives in expand.go (ExpandFieldRows),
// implementing design §3.6: __created__ / __deleted__ / __locked__ / Owner
// sentinels plus the PII blacklist, no-op skip, and LongText both-null
// suppression. ChangedAt on each expanded row is derived from
// row.EnqueuedAt (C6.1 invariant).
package audit

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"time"

	"github.com/rs/zerolog/log"
)

// OutboxRow is the worker's view of one pending _System_AuditOutbox row.
type OutboxRow struct {
	ID         int64
	Payload    []byte
	EnqueuedAt time.Time
	Attempts   int
}

// FieldRow is one expanded field-level row destined for _System_AuditLog.
// The store is responsible for INSERT IGNORE against uk_outbox.
type FieldRow struct {
	OutboxRowID   int64
	ObjectApiName string
	RecordId      string
	FieldName     string
	OldValue      string
	NewValue      string
	ChangedById   string
	ChangedAt     time.Time
	Action        AuditAction
	Entrypoint    AuditEntrypoint
}

// OutboxStore abstracts the DB operations the worker needs. Production uses
// a MySQL-backed implementation; tests use an in-memory fake. Keeping the
// surface narrow makes both the worker and its tests easy to reason about.
type OutboxStore interface {
	// AcquireLease claims up to batchSize pending rows for holderID with a
	// lease expiring at now+lease. Rows already leased (and not expired) or
	// with any non-NULL PoisonClass are skipped (§3.3 idx_unprocessed).
	// Returns empty slice if nothing pending.
	AcquireLease(ctx context.Context, holderID string, batchSize int, lease time.Duration, maxAttempts int) ([]OutboxRow, error)

	// InsertFieldRows writes the expanded field rows for a single outbox
	// row. Implementations MUST use INSERT IGNORE on uk_outbox to be safe
	// against at-least-once delivery (lease expiry + re-acquire).
	InsertFieldRows(ctx context.Context, outboxID int64, rows []FieldRow) error

	// FinalizeSuccess sets ProcessedAt=NOW(6), clears lease. Called after
	// all field rows inserted cleanly.
	FinalizeSuccess(ctx context.Context, outboxID int64) error

	// MarkPoison sets PoisonClass + LastError, clears lease. ProcessedAt
	// stays NULL (row is retained for compliance traceability). Poisoned
	// rows are excluded from AcquireLease for ALL classes (idx_unprocessed
	// is PoisonClass IS NULL); operators reset PoisonClass=NULL via the
	// runbook procedure to retry after fixing root cause.
	MarkPoison(ctx context.Context, outboxID int64, class PoisonClass, lastErr string) error

	// BumpAttempts increments Attempts and records LastError, clears lease.
	// Called on transient failures before the worker decides whether the
	// retry budget is exhausted.
	BumpAttempts(ctx context.Context, outboxID int64, lastErr string) error
}

// OutboxWorkerConfig holds tunables. Sensible production defaults are
// documented; tests override for speed.
type OutboxWorkerConfig struct {
	BatchSize     int           // default 100
	PollInterval  time.Duration // default 250ms
	LeaseDuration time.Duration // default 30s
	MaxAttempts   int           // default 5
	HolderID      string        // unique per worker instance (e.g. hostname+pid)

	// StopWhenIdle, when true, makes Run() exit after one poll returns
	// zero rows — used by tests to drain deterministically. Production
	// leaves this false; the worker runs until ctx is cancelled.
	StopWhenIdle bool
}

func (c OutboxWorkerConfig) withDefaults() OutboxWorkerConfig {
	if c.BatchSize <= 0 {
		c.BatchSize = 100
	}
	if c.PollInterval <= 0 {
		c.PollInterval = 250 * time.Millisecond
	}
	if c.LeaseDuration <= 0 {
		c.LeaseDuration = 30 * time.Second
	}
	if c.MaxAttempts <= 0 {
		c.MaxAttempts = 5
	}
	if c.HolderID == "" {
		c.HolderID = "outbox-worker"
	}
	return c
}

// OutboxWorker processes _System_AuditOutbox rows.
type OutboxWorker struct {
	store OutboxStore
	cfg   OutboxWorkerConfig
	clock func() time.Time
}

// NewOutboxWorker builds a worker with the given store + config.
func NewOutboxWorker(store OutboxStore, cfg OutboxWorkerConfig) *OutboxWorker {
	return &OutboxWorker{
		store: store,
		cfg:   cfg.withDefaults(),
		clock: time.Now,
	}
}

// Run blocks until ctx is cancelled (or — when StopWhenIdle=true — until the
// first empty poll). Errors on individual rows are logged inside processBatch
// and the worker keeps running; the only thing that stops the loop is ctx.
func (w *OutboxWorker) Run(ctx context.Context) {
	ticker := time.NewTicker(w.cfg.PollInterval)
	defer ticker.Stop()

	// Run one poll immediately so tests with tight timeouts don't miss a tick.
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

		processed, err := w.processBatch(ctx)
		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
			// Swallow — tx/driver errors at the batch level are expected to
			// be transient. A real server logger would go here; for now we
			// rely on per-row errors being captured via BumpAttempts /
			// MarkPoison, which are the durable signal.
			_ = err
		}
		if processed == 0 && w.cfg.StopWhenIdle {
			return
		}

		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

// processBatch runs one acquire → process-each → continue round. Returns
// the count of rows attempted (not necessarily succeeded).
func (w *OutboxWorker) processBatch(ctx context.Context) (int, error) {
	rows, err := w.store.AcquireLease(ctx, w.cfg.HolderID, w.cfg.BatchSize, w.cfg.LeaseDuration, w.cfg.MaxAttempts)
	if err != nil {
		return 0, fmt.Errorf("acquire lease: %w", err)
	}
	for i := range rows {
		w.processRow(ctx, rows[i])
	}
	return len(rows), nil
}

// processRow handles one leased outbox row end-to-end.
func (w *OutboxWorker) processRow(ctx context.Context, row OutboxRow) {
	// 1. Parse payload. Parse failures are terminal poison.
	var ev AuditEvent
	if err := json.Unmarshal(row.Payload, &ev); err != nil {
		class := ClassifyError(err)
		if class == PoisonClassNone {
			class = PoisonClassParseFail // json errors not caught by ClassifyError default to parse
		}
		if merr := w.store.MarkPoison(ctx, row.ID, class, sanitizeError(err)); merr != nil {
			log.Warn().Int64("outbox_id", row.ID).Err(merr).Msg("audit outbox: mark-poison failed (row may reprocess)")
		}
		return
	}

	// 2. Expand into field rows per design §3.6 (C7). Sentinels for
	//    CREATE / DELETE / LOCK / OWNER_CHANGE; one row per UPDATE diff
	//    with PII + no-op + LongText-both-null skips.
	fieldRows := ExpandFieldRows(ev, row, time.Now)

	// 3. Insert. INSERT IGNORE in the store makes retries safe.
	if err := w.store.InsertFieldRows(ctx, row.ID, fieldRows); err != nil {
		w.classifyAndMark(ctx, row, err)
		return
	}

	// 4. Success.
	if err := w.store.FinalizeSuccess(ctx, row.ID); err != nil {
		// Finalize failure is itself transient (probably deadlock).
		// Bump attempts and let next poll retry — INSERT IGNORE handles
		// the duplicate fan-out.
		w.classifyAndMark(ctx, row, err)
	}
}

// classifyAndMark dispatches a post-insert error to the right shelf:
// poison (terminal) or bump-attempts (+ possibly RETRY_EXHAUSTED).
func (w *OutboxWorker) classifyAndMark(ctx context.Context, row OutboxRow, err error) {
	class := ClassifyError(err)
	msg := sanitizeError(err)
	if class != PoisonClassNone {
		// Terminal: mark immediately, no Attempts bump.
		if merr := w.store.MarkPoison(ctx, row.ID, class, msg); merr != nil {
			log.Warn().Int64("outbox_id", row.ID).Err(merr).Msg("audit outbox: mark-poison failed (row may reprocess)")
		}
		return
	}
	// Transient: bump then decide on exhaustion. Note row.Attempts is the
	// pre-increment value seen at lease time; after BumpAttempts, the
	// new attempts count is row.Attempts+1.
	if berr := w.store.BumpAttempts(ctx, row.ID, msg); berr != nil {
		log.Warn().Int64("outbox_id", row.ID).Err(berr).Msg("audit outbox: bump-attempts failed (row may reprocess with stale attempts)")
	}
	if row.Attempts+1 >= w.cfg.MaxAttempts {
		if merr := w.store.MarkPoison(ctx, row.ID, PoisonClassRetryExhausted, msg); merr != nil {
			log.Warn().Int64("outbox_id", row.ID).Err(merr).Msg("audit outbox: mark-poison (retry-exhausted) failed (row may reprocess)")
		}
	}
}

// truncate caps LastError at the column width (VARCHAR(1024) per §2.4)
// leaving 4 bytes for an ellipsis marker.
const lastErrorMaxLen = 1020

func truncate(s string) string {
	if len(s) <= lastErrorMaxLen {
		return s
	}
	return s[:lastErrorMaxLen] + "..."
}

// quotedLiteral matches single- or double-quoted string literals in a MySQL
// error message (e.g. "Duplicate entry 'alice@example.com' for key 'uk_X'").
// Non-greedy so adjacent literals don't fuse into one span.
//
// TODO(C13): integrate with _System_RequestCapture PII blacklist for full
// structured scrub (field-name aware, not just quote-aware).
var quotedLiteral = regexp.MustCompile(`'[^']*'|"[^"]*"`)

// sanitizeError renders err into a LastError-safe string: strips quoted
// string literals (which MySQL commonly uses to echo user-provided values
// — e.g. "Duplicate entry 'alice@example.com' ...") and then truncates
// to the column width. Returns "" for nil.
func sanitizeError(err error) string {
	if err == nil {
		return ""
	}
	s := err.Error()
	s = quotedLiteral.ReplaceAllStringFunc(s, func(m string) string {
		if len(m) == 0 {
			return m
		}
		q := string(m[0]) // preserve the quote style for readability
		return q + "<redacted>" + q
	})
	return truncate(s)
}