queue

type Broker interface {
	// Publish appends payload to the queue and returns the message the
	// broker minted for it.
	//
	// It takes BYTES and not a [MessageValue] on purpose: the identifier and
	// the enqueue instant are the broker's to assign, and a struct with
	// fields the caller must leave blank is a struct that will eventually be
	// filled in.
	//
	// When it returns nil, a durable implementation has flushed the message
	// to its device: a crash one instruction later does not lose it. That
	// flush is most of what the call costs — see the domain's BENCH.md. An
	// implementation that joins the transaction ctx carries (ADR 0151) has
	// instead made the message part of that transaction: it is durable, and
	// receivable, once the transaction commits, and it never existed if the
	// transaction rolls back.
	//
	// A payload larger than [PolicyValue.MaxMessageBytes] is
	// [MessageTooLarge]; an empty payload is legitimate, because a message
	// whose whole meaning is its arrival is a normal thing to send.
	Publish(ctx context.Context, payload []byte) (MessageValue, error)
	// Receive leases up to max messages and returns them. It does NOT block
	// waiting for work: an empty queue yields an empty slice and a nil
	// error, and polling is the caller's (see internal/service/data/queue.Consume,
	// which does it for you).
	//
	// Each returned message is invisible to every other consumer until its
	// [DeliveryValue.Lease] expires. A consumer that finishes acknowledges
	// with [Broker.Ack]; a consumer that fails reports it with [Broker.Nack];
	// a consumer that DIES does neither, and the lapsed lease is what makes
	// the message reappear.
	//
	// max must be positive: [InvalidBatchSize] otherwise, rather than an
	// empty slice forever from a consumer loop that spins and processes
	// nothing.
	Receive(ctx context.Context, max int) ([]DeliveryValue, error)
	// Ack removes the leased message permanently. It is the only call that
	// does.
	//
	// A receipt whose lease has already lapsed is [LeaseExpired] and removes
	// NOTHING: the message is back in the queue, or in the dead-letter
	// store, and another consumer may already hold it. Reporting that is the
	// same decision core/app/lock makes for a lapsed lease — a holder that lost
	// its claim must not be able to end somebody else's turn, and must find
	// out that it lost it.
	//
	// A receipt this broker never issued, or one it cannot parse, is
	// [UnknownReceipt].
	Ack(ctx context.Context, receipt ReceiptValue) error
	// Nack reports that processing failed and hands the message back.
	//
	// The broker decides what happens next and reports it: the message
	// becomes visible again after [PolicyValue.RetryDelay] — grown by
	// [PolicyValue.MaxRetryDelay] when the policy asks for it — or, when it
	// has now been delivered [PolicyValue.MaxDeliveries] times, it is moved to
	// the dead-letter store together with cause. See [NackValue]. A failure
	// no retry can fix is not nacked but rejected ([Rejecter]).
	//
	// cause is recorded. What is kept of it is the implementation's to
	// document, and the rule this domain follows is CLAUDE.md rule 4: the
	// wire-safe Public half and the dotted-quad code travel with the dead
	// letter, the Private half stays in the failing process's log, and the
	// message identifier is the join key between them. A nil cause is
	// accepted and recorded as an unexplained failure, because a consumer
	// that knows only "this did not work" must still be able to say so.
	//
	// A lapsed or unknown receipt is refused exactly as [Broker.Ack] refuses
	// it.
	Nack(ctx context.Context, receipt ReceiptValue, cause error) (NackValue, error)
}

type DeadLetterManager interface {
	// ReplayDeadLetter moves the dead letter id back into the queue, visible
	// at once, with its delivery count reset: its next delivery counts 1 and
	// it gets [PolicyValue.MaxDeliveries] attempts again. It keeps its ID, its
	// payload and its EnqueuedAt, so the log lines of both its lives join on
	// one identifier; the dead-letter record, its cause included, is gone once
	// the message is queued again — log it first if it matters.
	//
	// An id the store does not hold is [DeadLetterNotFound], and nothing is
	// queued. Two replays of one dead letter racing each other may both
	// queue it on a broker that cannot tell them apart, which is a duplicate
	// the at-least-once guarantee already permits; the brokers say which
	// they are.
	ReplayDeadLetter(ctx context.Context, id string) error
	// DeleteDeadLetter removes the dead letter id for good, payload, cause
	// and all. It is the only call in the domain that destroys a message
	// without its consumer's acknowledgement, and it is never made by the
	// SDK itself. An id the store does not hold is [DeadLetterNotFound].
	DeleteDeadLetter(ctx context.Context, id string) error
}

type DeadLetterReader interface {
	// DeadLetters returns up to max dead letters, oldest first. It does NOT
	// remove them: a dead letter is evidence, and a read that consumed it
	// would make the second investigator's job impossible. Removing one is a
	// decision, and [DeadLetterManager] is where it is made.
	//
	// max must be positive ([InvalidBatchSize]).
	DeadLetters(ctx context.Context, max int) ([]DeadLetterValue, error)
}

type DeadLetterValue struct {
	// FailedAt is when the last attempt failed and the message was moved.
	FailedAt time.Time
	// Reason is the SCREAMING_SNAKE identifier of the failure — the
	// errs.Error Reason of the cause, NOT_RETRYABLE for a handler's
	// [DoNotRetry] around an error that carried none, or LEASE_EXPIRED when
	// the message died by exhausting its deliveries without any consumer ever
	// reporting anything.
	Reason string
	// Cause is the wire-safe Public half of the last failure, or an empty
	// string when the consumer reported no cause at all. It never carries
	// the message payload, which is a security property this domain shares
	// with the rest of the SDK: a queue moves bytes a consumer must not
	// leak, and an error message is the classic place they leak from.
	Cause string
	// Message is the work, unmodified, with the identifier it was published
	// under.
	Message MessageValue
	// Code is the dotted-quad errs.Code of the last failure, or 0 when the
	// cause carried none. It is a uint32 rather than an errs.Code so that
	// this package's values stay free of the kernel's error type in a shape
	// a consumer may serialise; errs.Code(dl.Code) converts.
	Code uint32
	// Deliveries is how many times the message was delivered before it was
	// abandoned. It equals [PolicyValue.MaxDeliveries] for every dead letter
	// this SDK's brokers produce by running out of attempts, and is the count
	// the message had for one a handler rejected ([Rejecter]).
	Deliveries int
}

type DeliveryValue struct {
	// Lease is the claim this delivery grants and the receipt that ends it.
	Lease LeaseValue
	// Message is the work.
	Message MessageValue
	// Deliveries counts how many times this message has been handed to a
	// consumer, INCLUDING this one. It is 1 on the first delivery.
	//
	// It is a field of the port rather than a note in the documentation
	// because it is the at-least-once guarantee made unignorable: a consumer
	// cannot read a delivery without reading the number that says this may
	// not be the first time. A value above 1 means a previous consumer
	// failed, timed out, or died — the three are indistinguishable from
	// here, deliberately, since none of them changes what the handler must
	// do.
	//
	// It is also the number [PolicyValue.MaxDeliveries] is compared against:
	// a delivery whose count already equals the maximum is the last one this
	// message gets before the dead-letter store.
	Deliveries int
}

type Handler func(ctx context.Context, delivery DeliveryValue) error

type LeaseExtender interface {
	// Extend renews the lease named by receipt for a further by, measured
	// from now, and returns the lease that replaces it.
	//
	// It returns a NEW [LeaseValue] — a new receipt included — and the old
	// receipt is void from that moment. That is not ceremony: an
	// implementation whose lease deadline lives in a durable name can only
	// change it by replacing the name, and a receipt that silently kept
	// meaning something after the thing it named had moved would be the one
	// bug this whole domain is built to avoid.
	//
	// A lapsed lease is [LeaseExpired] and is NOT renewed, even when no
	// other consumer has taken the message yet: expiry is a deadline, not
	// "until somebody else wants it". The handler that asks and is refused
	// has learnt that its work is now a duplicate, which is exactly when it
	// needed to know.
	//
	// A non-positive by is [QueueMisconfigured]: the two readings of zero —
	// "lapse now" and "never lapse" — are opposites and both harmful. So is a
	// by whose deadline reaches past what a durable implementation can record
	// (2262, Unix nanoseconds): it is refused before anything moves, and the
	// lease the caller holds stays valid. Validate never sees this duration,
	// so [MaxDeadlineOffset] cannot bound it; the instant itself is checked.
	Extend(ctx context.Context, receipt ReceiptValue, by time.Duration) (LeaseValue, error)
}

type LeaseValue struct {
	// ExpiresAt is when the claim lapses and the message becomes eligible for
	// redelivery. It is the broker's clock.
	//
	// It is a promise about VISIBILITY and not about mutual exclusion: after
	// this instant another consumer may hold the same message while this one
	// is still running. A handler that is still working past it is producing
	// a duplicate, which the at-least-once guarantee already permits — see
	// [LeaseExtender] for the way to avoid producing one on purpose.
	ExpiresAt time.Time
	// Receipt names this lease to [Broker.Ack] and [Broker.Nack].
	Receipt ReceiptValue
}

type MessageValue struct {
	// EnqueuedAt is when the broker accepted the message. It is the ordering
	// key a FIFO implementation sorts on, and it is the broker's clock, not
	// the producer's.
	EnqueuedAt time.Time
	// ID identifies the message for the life of the queue and beyond it: a
	// dead letter carries the same value, so the log line the failing
	// consumer wrote and the record left on disk can be joined without either
	// one carrying the other's half of the story (CLAUDE.md rule 4).
	//
	// It is opaque. Its shape is the implementation's, it is unique within
	// one queue, and nothing outside the broker that minted it may parse it.
	ID string
	// Payload is the caller's bytes, unmodified and uninterpreted. This
	// domain does not encode: pick a codec, marshal, publish the bytes.
	//
	// The slice belongs to the receiver — a durable broker read it from a
	// device, and an in-memory one copies before handing it over, so a
	// handler may keep it, mutate it, or hold it past the acknowledgement
	// without corrupting a redelivery.
	Payload []byte
}

type NackValue struct {
	// Deliveries is how many times the message had been delivered when it
	// failed, counting from 1.
	Deliveries int
	// VisibleAt is when the message becomes available again. It is the zero
	// Time when DeadLettered is true.
	VisibleAt time.Time
	// DeadLettered reports that the message exhausted
	// [PolicyValue.MaxDeliveries] and was moved to the dead-letter store
	// with its cause. It will not be delivered again, unless somebody replays
	// it ([DeadLetterManager]).
	DeadLettered bool
}

type PolicyValue struct {
	// VisibilityTimeout is how long a [Broker.Receive] hides a message from
	// every other consumer. It is the deadline the consumer is racing, and
	// it must be longer than the slowest run of the handler — or the handler
	// must renew it through [LeaseExtender].
	//
	// It is the single knob that sets how quickly a dead consumer's work is
	// picked up: a message held by a process that has been SIGKILLed
	// reappears exactly this long after it was leased, and not before.
	//
	// A non-positive value is refused ([QueueMisconfigured]), and so is one
	// above [MaxDeadlineOffset].
	VisibilityTimeout time.Duration
	// RetryDelay is how long a NACKED message stays invisible before it
	// becomes eligible again, measured from the nack.
	//
	// It does NOT apply to a lease that simply lapsed: that message has
	// already been invisible for a whole visibility timeout, and charging it
	// a second wait would punish a crashed consumer more than a failing one.
	//
	// Zero is a working value and means "eligible as soon as the nack
	// returns". Negative is read as zero. Above [MaxDeadlineOffset] it is
	// refused ([QueueMisconfigured]).
	//
	// It is the same wait after every failure unless [PolicyValue]
	// .MaxRetryDelay makes it grow.
	RetryDelay time.Duration
	// MaxRetryDelay makes the retry delay GROW, and is where the growth
	// stops. When it is positive, a message nacked on its n-th delivery waits
	// RetryDelay × 2^(n−1), held at MaxRetryDelay — the SDK's one backoff
	// curve (kernel/backoff, ADR 0103), keyed on the delivery that failed,
	// so a downstream that is down is asked less and less often instead of at
	// a fixed cadence it is already failing to keep up with (ADR 0151).
	//
	// Zero keeps RetryDelay constant, which is what every policy did before
	// this field existed: a policy that does not name it behaves exactly as
	// it did. That is the zero's one reading — a ceiling nobody set is not a
	// request for unbounded growth, which could not be recorded anyway.
	//
	// A negative value is refused, because it is not a ceiling and the caller
	// who wrote it meant something; so is one above [MaxDeadlineOffset],
	// whose deadline a durable broker could not record; so is a ceiling with
	// no RetryDelay to grow from, which would leave every retry at zero while
	// reading like a backoff; and so is a ceiling below RetryDelay, which
	// would silently shorten every wait the caller asked for
	// ([QueueMisconfigured], naming the field).
	MaxRetryDelay time.Duration
	// MaxDeliveries is how many times one message may be handed to a
	// consumer before the broker gives up and dead-letters it. It is
	// compared against [DeliveryValue.Deliveries], which counts from 1, so a
	// value of 1 means "one attempt, then the dead-letter store" and is a
	// legitimate configuration rather than an off-by-one.
	//
	// A non-positive value is refused ([QueueMisconfigured]).
	MaxDeliveries int
	// MaxMessageBytes bounds one payload. A larger payload is refused at
	// [Broker.Publish] with [MessageTooLarge], at the call site that built
	// it, rather than accepted and discovered by a consumer that cannot read
	// it back.
	//
	// Zero means [DefaultMaxMessageBytes]. Negative is refused, because it
	// is not a bound at all and the caller who wrote it meant something.
	MaxMessageBytes int
}

type ReceiptValue string

type Rejecter interface {
	// Reject moves the leased message named by receipt to the dead-letter
	// store with cause, recorded as [Broker.Nack] records the cause of a
	// message that ran out of attempts: its Reason, its dotted-quad Code and
	// its wire-safe Public half, never the Private one. The dead letter's
	// Deliveries is the count the message had, which may be below
	// [PolicyValue.MaxDeliveries].
	//
	// A lapsed receipt is [LeaseExpired] and one this broker cannot read is
	// [UnknownReceipt], exactly as [Broker.Ack] refuses them; neither moves
	// anything.
	Reject(ctx context.Context, receipt ReceiptValue, cause error) error
}

type WakeValue struct {
	// Signal is closed by the next event that may make a message receivable
	// through this broker in this process: a Publish, or a Nack that hands a
	// message back. It is never nil, and it is never closed by a publication
	// made in another process.
	Signal <-chan struct{}
	// In is how long, measured on the broker's own clock at the moment Wake
	// was called, until a message the broker already holds becomes
	// receivable with no further event: a retry delay ending, or a lease
	// lapsing because its consumer died. It is meaningful only when
	// Scheduled is true; zero or less means now.
	//
	// It is a duration and not an instant so that a consumer whose clock is
	// not the broker's still waits the right length of time rather than
	// comparing two clocks that disagree.
	In time.Duration
	// Scheduled reports that In names such a moment. It is false when the
	// broker holds nothing that will become receivable on its own.
	Scheduled bool
}

type Waker interface {
	// Wake returns what an idle consumer should wait on. Call it BEFORE the
	// Receive that may come back empty: a publication landing between that
	// Receive and the wait closes the channel already handed out, so it is
	// not missed.
	Wake() WakeValue
}