mailbox

type ConsumeFn = func(context.Context, Message) error

type ConsumeMiddleware = func(ConsumeFn) ConsumeFn

type Consumer interface {
	// Consume consumes messages from the mailbox. It is safe to call this
	// method concurrently. This method is meant to be called in a loop, the
	// caller is responsible to apply back-pressure if needed.
	//
	// It is recommended that the function:
	//   - is idempotent, it may be called multiple times for the same message
	//   - returns quickly to avoid holding the row lock for too long
	//   - does not call other methods on the consumer to avoid deadlocks
	//
	// The following errors are returned:
	//   - nil: a message was successfully consumed
	//   - ErrNoMessage: the mailbox is empty and no message was consumed
	// 	 - Any other error: an error occurred while consuming a message
	//
	// The consumer does not have any dead-letter mechanism. If the function
	// returns an error, the message is re-queued and will be processed again.
	// It is the responsibility of the caller to handle a maximum number of
	// retries and/or to move the message to a dead-letter queue, see the
	// various middlewares for more details, e.g. WithTimeoutConsume,
	// WithRetryPolicyConsume.
	Consume(context.Context) error

	// Size returns the number of messages in the mailbox.
	Size(context.Context) (size int64, err error)
}

type Mailbox interface {
	// Put adds a message to the mailbox. If the method returns an error,
	// the message is not added to the mailbox and caller is responsible
	// for retrying and/or rolling back the transaction. The transaction is
	// explicitly passed allowing the caller to control the transaction boundaries.
	Put(context.Context, *sql.Tx, Message) error
}

type MatchFn func(Message) bool

type Message struct {
	// The message identifier.
	ID string
	// The message metadata.
	Metadata map[string]string
	// The message payload.
	Payload []byte
}

type Metadata = internal.Metadata

type ObservabilityPolicy struct {
	// Metrics is the expvar map used to record metrics. If nil, no metrics
	// are recorded. Two metrics are recorded:
	//   - messages_consumed_total: the number of messages successfully consumed
	//   - messages_consumed_failed_total: the number of messages that failed to be consumed
	// The metrics keys are statically defined by MessageConsumedKey and
	// MessageConsumedFailedKey.
	Metrics *expvar.Map
	// Logger is the logger used to log events. If nil, no logging is done.
	// By default, only errors are logged. See LogSuccess for more details.
	Logger *slog.Logger
	// Attrs is the function that returns the attributes to log. If nil, no
	// attributes are extracted and logged. See slog's documentation for more
	// details on attributes.
	Attrs func(context.Context, Message) []any
	// If set, successful message consumption is logged. Errors are always
	// logged. If nil, no logging is done.
	LogSuccess bool
}

type RetryPolicy struct {
	// Backoff is the function that returns the duration to wait before
	// retrying. The argument is the number of retries and the message. If
	// the function returns 0, it retries immediately. By default, it uses
	// an exponential backoff similar to Kubernetes, e.g. 10s, 20s, 40s, ...,
	// capped at 5 minutes.
	Backoff func(int, Message) time.Duration
	// Final is invoked after all retries have been exhausted. For example,
	// it could be used to move the message to a dead-letter queue, or it could
	// be used to log the message. If the function returns an error, the
	// message is retried. By default, it swallows the error. Note that this
	// function is not recovered from.
	Final ConsumeFn
	// MaxAttempts is the maximum number of attempts before giving up. If the
	// value not strictly positive, it is set to DefaultMaxAttempts.
	MaxAttempts int
	// AttemptTimeout is the timeout for each attempt. If the value is not
	// strictly positive, it is set to DefaultAttemptTimeout.
	AttemptTimeout time.Duration
}

type Route struct {
	Match   MatchFn
	Consume ConsumeFn
}