// Package audit contains the durable, fenced audit outbox worker.
package audit

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	rand "math/rand/v2"
	"regexp"
	"sync/atomic"
	"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
}

// OutboxLease is the complete authority returned by AcquireLease. Every state
// transition replays the holder and observed Attempts from this value.
type OutboxLease struct {
	Row            OutboxRow
	HolderID       string
	LeaseExpiresAt time.Time
	OwnershipEpoch uint64
}

// FieldRow is one expanded field-level row destined for _System_AuditLog.
type FieldRow struct {
	OutboxRowID          int64
	ObjectApiName        string
	RecordId             string
	FieldName            string
	OldValue             string
	NewValue             string
	ChangedById          string
	ChangedAt            time.Time
	EventAt              time.Time
	FieldHistoryEligible *bool
	HistorySequence      *int64
	Action               AuditAction
	Entrypoint           AuditEntrypoint
	Outcome              Outcome
}

var (
	ErrOutboxLeaseLost        = errors.New("audit outbox: lease ownership lost")
	ErrOutboxCommitUnknown    = errors.New("audit outbox: commit state unknown")
	ErrOutboxNonExactReplay   = errors.New("audit outbox: duplicate is not an exact replay")
	ErrOutboxInvalidFieldRows = errors.New("audit outbox: invalid field rows")
)

// OutboxCommitState records whether fan-out is known to have had no field-row
// effect. Only NoEffect and RolledBack errors may enter AfterRollback.
type OutboxCommitState string

const (
	OutboxCommitNoEffect   OutboxCommitState = "no_effect"
	OutboxCommitRolledBack OutboxCommitState = "rolled_back"
	OutboxCommitUnknown    OutboxCommitState = "commit_unknown"
)

// OutboxOperationError carries the commit-state and deterministic disposition
// of a failed fan-out operation.
type OutboxOperationError struct {
	CommitState   OutboxCommitState
	Deterministic bool
	PoisonClass   PoisonClass
	Err           error
}

func (e *OutboxOperationError) Error() string {
	if e == nil || e.Err == nil {
		return "audit outbox operation failed"
	}
	return e.Err.Error()
}

func (e *OutboxOperationError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Err
}

func (e *OutboxOperationError) Is(target error) bool {
	return target == ErrOutboxCommitUnknown && e != nil && e.CommitState == OutboxCommitUnknown
}

// OutboxStore is the fenced persistence boundary used by the worker.
type OutboxStore interface {
	CheckReady(ctx context.Context) error
	AcquireLease(ctx context.Context, holderID string, batchSize int, lease time.Duration, maxAttempts int) ([]OutboxLease, error)
	// CommitFanout writes the complete audit effect and consumes the leased
	// outbox row in one fenced database transaction. A returned unknown commit
	// outcome must not be followed by a disposition: either both facts committed
	// or the lease remains available for a later owner to resolve.
	CommitFanout(ctx context.Context, lease OutboxLease, rows []FieldRow) error
	BumpAttemptsAfterRollback(ctx context.Context, lease OutboxLease, maxAttempts int, lastErr string) (retryExhausted bool, err error)
	MarkPoisonAfterRollback(ctx context.Context, lease OutboxLease, class PoisonClass, lastErr string) error
	RepairPoison(ctx context.Context, repair OutboxRepair) error
}

// OutboxRepair is an operator-authored compare-and-set replay request. It
// cannot affect live, successful, or unexpectedly changed work.
type OutboxRepair struct {
	OutboxRowID      int64
	ExpectedClass    PoisonClass
	ExpectedAttempts int
	RepairID         string
	Reason           string
}

// OutboxQueueState is a bounded aggregate view of the durable queue. It never
// contains payloads or tenant identifiers.
type OutboxQueueState struct {
	Depth       int64
	PoisonDepth int64
	OldestAge   time.Duration
}

// OutboxQueueObserver is optional so transitional stores and focused tests do
// not gain a second work-control surface merely to report metrics.
type OutboxQueueObserver interface {
	ObserveQueue(ctx context.Context) (OutboxQueueState, error)
}

// OutboxWorkerSnapshot contains only process-local business counters. Durable
// queue observation is owned by the process that exports operational metrics;
// it is never part of this worker's claim or effect loop.
type OutboxWorkerSnapshot struct {
	InFlight      int64
	Attempts      uint64
	Failures      uint64
	Processed     uint64
	PollChecks    uint64
	ConfirmedIdle uint64
	ClaimAttempts uint64
	EmptyClaims   uint64
	Claimed       uint64
	PollDuration  time.Duration
	ClaimDuration time.Duration
}

// OutboxWorkerConfig holds bounded worker tunables.
type OutboxWorkerConfig struct {
	BatchSize      int
	PollInterval   time.Duration
	LeaseDuration  time.Duration
	RetryJitterMax time.Duration
	MaxAttempts    int
	HolderID       string
	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.RetryJitterMax <= 0 {
		c.RetryJitterMax = 250 * time.Millisecond
	}
	if c.MaxAttempts <= 0 {
		c.MaxAttempts = 5
	}
	return c
}

// OutboxWorker processes _System_AuditOutbox rows.
type OutboxWorker struct {
	store   OutboxStore
	cfg     OutboxWorkerConfig
	initial []OutboxLease

	inFlightN          atomic.Int64
	attemptsN          atomic.Uint64
	failuresN          atomic.Uint64
	processedN         atomic.Uint64
	pollChecksN        atomic.Uint64
	confirmedIdleN     atomic.Uint64
	claimAttemptsN     atomic.Uint64
	emptyClaimsN       atomic.Uint64
	claimedN           atomic.Uint64
	pollDurationNanos  atomic.Int64
	claimDurationNanos atomic.Int64
}

func NewOutboxWorker(store OutboxStore, cfg OutboxWorkerConfig) *OutboxWorker {
	return &OutboxWorker{store: store, cfg: cfg.withDefaults()}
}

func (w *OutboxWorker) Snapshot() OutboxWorkerSnapshot {
	if w == nil {
		return OutboxWorkerSnapshot{}
	}
	return OutboxWorkerSnapshot{
		InFlight:   w.inFlightN.Load(),
		Attempts:   w.attemptsN.Load(),
		Failures:   w.failuresN.Load(),
		Processed:  w.processedN.Load(),
		PollChecks: w.pollChecksN.Load(), ConfirmedIdle: w.confirmedIdleN.Load(),
		ClaimAttempts: w.claimAttemptsN.Load(), EmptyClaims: w.emptyClaimsN.Load(),
		Claimed: w.claimedN.Load(), PollDuration: time.Duration(w.pollDurationNanos.Load()),
		ClaimDuration: time.Duration(w.claimDurationNanos.Load()),
	}
}

func (w *OutboxWorker) acquireLeases(ctx context.Context) ([]OutboxLease, error) {
	started := time.Now()
	leases, err := w.store.AcquireLease(
		ctx, w.cfg.HolderID, w.cfg.BatchSize, w.cfg.LeaseDuration, w.cfg.MaxAttempts,
	)
	duration := time.Since(started).Nanoseconds()
	w.pollChecksN.Add(1)
	w.claimAttemptsN.Add(1)
	w.pollDurationNanos.Add(duration)
	w.claimDurationNanos.Add(duration)
	if err != nil {
		w.failuresN.Add(1)
		return nil, err
	}
	if len(leases) == 0 {
		w.confirmedIdleN.Add(1)
		w.emptyClaimsN.Add(1)
	} else {
		w.claimedN.Add(uint64(len(leases)))
	}
	return leases, nil
}

// Prepare performs the bounded real initial claim. Runtime marks the process
// ready only after this succeeds; Run consumes these leases before polling.
func (w *OutboxWorker) Prepare(ctx context.Context) error {
	if w == nil || w.store == nil {
		return errors.New("audit outbox: nil worker store")
	}
	if len(w.cfg.HolderID) < 1 || len(w.cfg.HolderID) > 64 {
		return errors.New("audit outbox: holder ID length must be 1..64")
	}
	leases, err := w.acquireLeases(ctx)
	if err != nil {
		return fmt.Errorf("audit outbox initial claim: %w", err)
	}
	w.initial = leases
	return nil
}

func (w *OutboxWorker) Run(ctx context.Context) error {
	ticker := time.NewTicker(w.cfg.PollInterval)
	defer ticker.Stop()

	if len(w.initial) > 0 {
		initial := w.initial
		w.initial = nil
		if err := w.processLeases(ctx, initial); err != nil {
			return err
		}
	}

	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}
		processed, err := w.processBatch(ctx)
		if errors.Is(err, ErrOutboxWorkOwnershipLost) {
			return err
		}
		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
			log.Warn().Err(err).Msg("audit outbox: poll failed")
		}
		if processed == 0 && w.cfg.StopWhenIdle {
			return nil
		}
		select {
		case <-ctx.Done():
			return nil
		case <-ticker.C:
		}
	}
}

func (w *OutboxWorker) processBatch(ctx context.Context) (int, error) {
	leases, err := w.acquireLeases(ctx)
	if err != nil {
		return 0, fmt.Errorf("acquire lease: %w", err)
	}
	if err := w.processLeases(ctx, leases); err != nil {
		return len(leases), err
	}
	return len(leases), nil
}
func (w *OutboxWorker) processLeases(ctx context.Context, leases []OutboxLease) error {
	for i := range leases {
		if ctx.Err() != nil {
			return nil
		}
		if err := w.processLease(ctx, leases[i]); err != nil {
			return err
		}
	}
	return nil
}

func (w *OutboxWorker) processLease(ctx context.Context, lease OutboxLease) error {
	w.attemptsN.Add(1)
	w.inFlightN.Add(1)
	defer w.inFlightN.Add(-1)
	var event AuditEvent
	if err := json.Unmarshal(lease.Row.Payload, &event); err != nil {
		w.failuresN.Add(1)
		return w.poisonAfterNoEffect(ctx, lease, PoisonClassParseFail, err)
	}
	if err := event.Validate(); err != nil {
		w.failuresN.Add(1)
		return w.poisonAfterNoEffect(ctx, lease, PoisonClassParseFail, err)
	}

	rows := ExpandFieldRows(event, lease.Row, time.Now)
	if err := w.store.CommitFanout(ctx, lease, rows); err != nil {
		w.failuresN.Add(1)
		return w.disposeFanoutFailure(ctx, lease, err)
	}
	w.processedN.Add(1)
	return nil
}

func (w *OutboxWorker) poisonAfterNoEffect(ctx context.Context, lease OutboxLease, class PoisonClass, cause error) error {
	if ctx.Err() != nil {
		return nil
	}
	if err := w.store.MarkPoisonAfterRollback(ctx, lease, class, sanitizeError(cause)); err != nil {
		if errors.Is(err, ErrOutboxWorkOwnershipLost) {
			return err
		}
		if !errors.Is(err, ErrOutboxLeaseLost) {
			log.Warn().Int64("outbox_id", lease.Row.ID).Err(err).Msg("audit outbox: deterministic disposition failed")
		}
	}
	return nil
}

func (w *OutboxWorker) disposeFanoutFailure(ctx context.Context, lease OutboxLease, err error) error {
	if errors.Is(err, ErrOutboxWorkOwnershipLost) {
		return err
	}
	if ctx.Err() != nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrOutboxLeaseLost) {
		return nil
	}
	var operationErr *OutboxOperationError
	if !errors.As(err, &operationErr) ||
		(operationErr.CommitState != OutboxCommitNoEffect && operationErr.CommitState != OutboxCommitRolledBack) {
		// An unknown commit may contain both the effect and the acknowledgement.
		// Do not mutate the outbox; expiry and the atomic fanout resolve it.
		return nil
	}
	if operationErr.Deterministic {
		class := operationErr.PoisonClass
		if class != PoisonClassParseFail && class != PoisonClassFKViolation {
			return nil
		}
		if markErr := w.store.MarkPoisonAfterRollback(ctx, lease, class, sanitizeError(err)); markErr != nil {
			if errors.Is(markErr, ErrOutboxWorkOwnershipLost) {
				return markErr
			}
			if !errors.Is(markErr, ErrOutboxLeaseLost) {
				log.Warn().Int64("outbox_id", lease.Row.ID).Err(markErr).Msg("audit outbox: poison disposition failed")
			}
		}
		return nil
	}
	if _, bumpErr := w.store.BumpAttemptsAfterRollback(ctx, lease, w.cfg.MaxAttempts, sanitizeError(err)); bumpErr != nil {
		if errors.Is(bumpErr, ErrOutboxWorkOwnershipLost) {
			return bumpErr
		}
		if !errors.Is(bumpErr, ErrOutboxLeaseLost) {
			// A disposition failure is never retried by the old worker.
			log.Warn().Int64("outbox_id", lease.Row.ID).Err(bumpErr).Msg("audit outbox: retry disposition failed")
		}
	} else {
		w.waitRetryJitter(ctx)
	}
	return nil
}

func (w *OutboxWorker) waitRetryJitter(ctx context.Context) {
	maximum := w.cfg.RetryJitterMax
	if maximum <= 0 {
		return
	}
	delay := time.Duration(rand.Int64N(maximum.Nanoseconds() + 1))
	timer := time.NewTimer(delay)
	defer timer.Stop()
	select {
	case <-ctx.Done():
	case <-timer.C:
	}
}

const lastErrorMaxLen = 1020

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

var quotedLiteral = regexp.MustCompile(`'[^']*'|"[^"]*"`)

func sanitizeError(err error) string {
	if err == nil {
		return ""
	}
	s := quotedLiteral.ReplaceAllStringFunc(err.Error(), func(value string) string {
		if value == "" {
			return value
		}
		quote := string(value[0])
		return quote + "<redacted>" + quote
	})
	return truncate(s)
}