txoutbox

type Backoff func(attempt int) time.Duration

type Envelope struct {
	// ID is the primary key of the outbox row.
	ID int64
	// Topic is copied from the original Message for routing/logging.
	Topic string
	// Key is optional metadata used by senders for partitioning/idempotency.
	Key *string
	// Payload is the raw JSON message stored in the outbox.
	Payload json.RawMessage
	// RetryCount tracks how many attempts have been made (before this lease).
	RetryCount int
	// CreatedAt records when the row was inserted.
	CreatedAt time.Time
}

type Executor interface {
	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
}

type Hooks interface {
	// OnClaim fires after each Claim with the requested batch vs actual rows.
	OnClaim(ctx context.Context, batchSize int, claimed int)
	// OnSendSuccess fires for every Envelope delivered successfully.
	OnSendSuccess(ctx context.Context, env Envelope)
	// OnSendFailure fires when Sender returns an error before retry/fail handling.
	OnSendFailure(ctx context.Context, env Envelope, err error)
	// OnRetry fires when a message is rescheduled for another attempt.
	OnRetry(ctx context.Context, env Envelope, nextAttempt int, delay time.Duration)
	// OnFail fires when a message is permanently failed.
	OnFail(ctx context.Context, env Envelope, attempts int, err error)
	// OnStoreError fires when a Store call returns an error.
	OnStoreError(ctx context.Context, op string, id int64, err error)
	// OnCycle fires once per processOnce iteration with the elapsed duration.
	OnCycle(ctx context.Context, duration time.Duration)
}

type Logger interface {
	Info(ctx context.Context, format string, v ...any)
	Warn(ctx context.Context, format string, v ...any)
	Error(ctx context.Context, format string, v ...any)
}

type Message struct {
	// Topic identifies the logical event or routing destination (e.g. "order.created").
	Topic string
	// Key optionally provides a partition/idempotency key; leave empty if unused.
	Key string
	// Body is the user payload that will be marshaled to JSON.
	Body any
}

type Options struct {
	// BatchSize controls how many records the relay claims per iteration.
	BatchSize int
	// LeaseTTL defines how long a claimed message stays owned before expiring.
	LeaseTTL time.Duration
	// MaxAttempts is the number of total send tries before marking as failed.
	MaxAttempts int
	// PollInterval is the sleep duration between claim cycles when no work exists.
	PollInterval time.Duration
	// Backoff computes the retry delay based on attempt count.
	Backoff Backoff
	// Logger emits structured logs for relay activity.
	Logger Logger
	// Hooks let callers plug metrics/tracing/etc. into relay events.
	Hooks Hooks
	// WorkerID identifies this relay instance in the database.
	WorkerID string
	// Now supplies the current time; override for tests or custom time sources.
	Now func() time.Time
}

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

type Sender interface {
	Send(ctx context.Context, msg Envelope) error
}

type Store interface {
	// Add enqueues a message using the provided transaction/executor (typically *sql.Tx).
	Add(ctx context.Context, exec Executor, msg Message) error
	// Claim selects pending messages and leases them to a worker, returning envelopes to process.
	Claim(ctx context.Context, workerID string, limit int, leaseTTL time.Duration) ([]Envelope, error)
	// Send marks a message as successfully delivered.
	Send(ctx context.Context, id int64, sendAt time.Time) error
	// Retry releases the message to be retried later after incrementing the attempt counter.
	Retry(ctx context.Context, id int64, retryCount int, nextRetry time.Time) error
	// Fail flags the message as permanently failed so operators can inspect the row.
	Fail(ctx context.Context, id int64, retryCount int) error
}