outbox

type ClaimedEntry struct {
	kout.Entry
	Attempts int
	LeaseID  string
}

type FailureBudget struct {
	// contains filtered or unexported fields
}

type PendingDepthObserver interface {
	ObservePendingDepth(ctx context.Context, n int64)
}

type Relay struct {
	// contains filtered or unexported fields
}

type RelayConfig struct {
	// PollInterval is how often the relay polls for pending entries.
	PollInterval time.Duration
	// BatchSize is the maximum number of entries fetched per poll cycle.
	BatchSize int
	// RetentionPeriod is how long published entries are kept before cleanup.
	RetentionPeriod time.Duration
	// MaxAttempts is the maximum number of publish attempts before an entry
	// is marked as dead-lettered. Default 5.
	MaxAttempts int
	// BaseRetryDelay is the base delay for exponential backoff. Default 5s.
	// Actual delay = cappedDelay(BaseRetryDelay * 2^attempts) + jitter.
	BaseRetryDelay time.Duration
	// ClaimTTL is how long a claiming entry is held before ReclaimStale
	// recovers it back to pending. Default 60s.
	ClaimTTL time.Duration
	// MaxRetryDelay caps the exponential backoff delay to prevent
	// unbounded retry intervals at high attempt counts. Default 5m.
	MaxRetryDelay time.Duration
	// ReclaimInterval controls the independent ReclaimStale goroutine
	// frequency, decoupled from cleanup interval. Default 30s.
	ReclaimInterval time.Duration
	// ReclaimBatchSize caps a single ReclaimStale UPDATE; the relay's reclaim
	// loop drains residual on its own tick when the stale row count exceeds
	// this cap. Default 1000 (see defaultRelayReclaimBatchSize for the
	// industry-comparison rationale).
	ReclaimBatchSize int
	// DeadRetentionPeriod is how long dead-lettered entries are kept before
	// cleanup. Separate from RetentionPeriod to give operators more time
	// to investigate and manually retry failed entries. Default 30 days.
	DeadRetentionPeriod time.Duration
	// Metrics is the relay metrics collector for Prometheus integration.
	// If nil, a NoopRelayCollector is used (zero overhead).
	// ref: Temporal client.Options{MetricsHandler} — inject-at-construction pattern
	Metrics kout.RelayCollector

	// PollFailureBudget is the consecutive poll-loop failure count that trips
	// /readyz unhealthy. 0 disables the checker. Default 5.
	// ref: K8s workqueue ItemExponentialFailureRateLimiter — absolute count + Forget.
	PollFailureBudget int
	// ReclaimFailureBudget is the consecutive reclaim-loop failure count that
	// trips /readyz unhealthy. 0 disables. Default 5.
	ReclaimFailureBudget int
	// CleanupFailureBudget is the consecutive cleanup-loop failure count that
	// trips /readyz unhealthy. 0 disables. Default 5.
	CleanupFailureBudget int

	// CleanupWaitFloor is the minimum sleep between cleanup passes.
	// Exported so tests can lower it to 1ms without touching the global
	// constant. <= 0 uses the package default (5s).
	// Tests can set it directly via the RelayConfig literal.
	CleanupWaitFloor time.Duration
}

type Store interface {
	// ClaimPending atomically transitions up to batchSize rows from pending
	// to claiming status, skipping rows locked by other relay instances.
	// Each entry returned carries the LeaseID generated for this claim batch;
	// callers MUST echo LeaseID back through Mark* CAS to retain ownership.
	// Returns an empty slice + nil error when there is nothing to claim.
	ClaimPending(ctx context.Context, batchSize int) ([]ClaimedEntry, error)

	// MarkPublished optimistically transitions an entry from claiming to
	// published. The leaseID must match the value returned from ClaimPending;
	// updated=false means the lease was lost (entry reclaimed by ReclaimStale
	// or claimed by another worker) — not an error.
	MarkPublished(ctx context.Context, id, leaseID string) (updated bool, err error)

	// MarkRetry transitions a failing entry back to pending with attempts
	// incremented and the supplied nextRetryAt. Relay is responsible for
	// computing nextRetryAt (Go-side backoff). updated=false same as above.
	MarkRetry(ctx context.Context, id, leaseID string, attempts int, nextRetryAt time.Time, lastError string) (updated bool, err error)

	// MarkDead transitions a failing entry to dead (exceeded max attempts).
	// updated=false same as above.
	MarkDead(ctx context.Context, id, leaseID string, attempts int, lastError string) (updated bool, err error)

	// ReclaimStale transitions up to batchSize claiming rows whose claimed_at
	// is older than claimTTL back to pending (with attempts+1 and
	// next_retry_at = backoff) or to dead (when attempts+1 >= maxAttempts).
	// Returns count of rows recovered across both destinations. Callers MUST
	// loop until count < batchSize so a queue larger than one sweep drains
	// promptly without producing a multi-second UPDATE that blocks
	// VACUUM/replication.
	ReclaimStale(
		ctx context.Context,
		claimTTL time.Duration,
		maxAttempts int,
		baseDelay, maxDelay time.Duration,
		batchSize int,
	) (count int, err error)

	// CleanupPublished deletes a batch of published rows older than cutoff.
	// Caller is responsible for looping until deleted < batchSize.
	CleanupPublished(ctx context.Context, cutoff time.Time, batchSize int) (deleted int, err error)

	// CleanupDead deletes a batch of dead rows older than cutoff.
	CleanupDead(ctx context.Context, cutoff time.Time, batchSize int) (deleted int, err error)

	// CountPending returns the number of pending entries eligible for ClaimPending:
	// status=pending AND (next_retry_at IS NULL OR next_retry_at <= now()).
	// Rows that are pending but still in backoff (next_retry_at > now()) are
	// intentionally excluded so the count reflects actual work available to relay
	// workers — consistent with the ClaimPending eligibility predicate.
	//
	// May be approximate under high concurrency — multiple concurrent ClaimPending
	// calls can race between the COUNT and the next claim. Callers should treat
	// any error as transient and skip the metric update without panicking.
	//
	// ref: Eventuate transactional-outbox events_pending;
	// Debezium MilliSecondsBehindSource — non-blocking depth probe on reclaim cadence.
	CountPending(ctx context.Context) (int64, error)

	// OldestEligibleAt returns the oldest published_at (when status=kout.StatePublished)
	// or dead_at (when status=kout.StateDead) in the table. The relay uses this to
	// schedule data-driven cleanup wake-ups: sleep until oldest+retention instead of
	// polling on a fixed interval. Returns ok=false when no rows of the given status
	// exist.
	//
	// status MUST be kout.StatePublished or kout.StateDead. Implementations must
	// reject other values to keep the contract narrow.
	OldestEligibleAt(ctx context.Context, status kout.State) (at time.Time, ok bool, err error)
}