queue

type AdmissionPolicy string

type Config struct {
	// MaxDepth bounds the number of pending runs. Zero disables admission
	// control (unbounded intake).
	MaxDepth int
	// Policy selects reject vs. park behavior at capacity.
	Policy AdmissionPolicy
	// DefaultMaxAttempts is applied to enqueued runs that do not set one.
	DefaultMaxAttempts int
}

type Dialect string

type Dispatcher interface {
	Dispatch(ctx context.Context, e *OutboxEntry) error
}

type DispatcherFunc func(ctx context.Context, e *OutboxEntry) error

type Executor func(ctx context.Context, r *Run) Result

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

type OutboxEntry struct {
	ID             string     `json:"id"`
	SessionID      string     `json:"session_id"`
	IdempotencyKey string     `json:"idempotency_key"`
	Topic          string     `json:"topic"`
	Payload        []byte     `json:"payload,omitempty"`
	Status         string     `json:"status"` // pending, sent, failed
	Attempts       int        `json:"attempts"`
	LastError      string     `json:"last_error,omitempty"`
	CreatedAt      time.Time  `json:"created_at"`
	SentAt         *time.Time `json:"sent_at,omitempty"`
}

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

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

type Result struct {
	// Err, if non-nil, marks the attempt failed; the worker retries (with
	// backoff) until MaxAttempts, then fails the run terminally.
	Err error
	// Sleep, if > 0, durably reschedules the run to run again after the delay.
	Sleep time.Duration
	// ParkSignal, if set, parks the run until that signal is delivered.
	ParkSignal string
	// Patch, if non-nil, replaces the run's persisted payload on sleep/park so
	// the resumed attempt observes updated state.
	Patch []byte
}

type Run struct {
	ID             string     `json:"id"`
	SessionID      string     `json:"session_id"`
	GraphID        string     `json:"graph_id"`
	Kind           string     `json:"kind"`
	Payload        []byte     `json:"payload,omitempty"`
	Status         string     `json:"status"`
	Priority       int        `json:"priority"`
	Attempts       int        `json:"attempts"`
	MaxAttempts    int        `json:"max_attempts"`
	LeaseOwner     string     `json:"lease_owner,omitempty"`
	LeaseExpiresAt *time.Time `json:"lease_expires_at,omitempty"`
	AvailableAt    time.Time  `json:"available_at"`
	WaitSignal     string     `json:"wait_signal,omitempty"`
	SignalPayload  []byte     `json:"signal_payload,omitempty"`
	IdempotencyKey string     `json:"idempotency_key,omitempty"`
	LastError      string     `json:"last_error,omitempty"`
	CreatedAt      time.Time  `json:"created_at"`
	UpdatedAt      time.Time  `json:"updated_at"`
}

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

type Signal struct {
	SessionID string `json:"session_id"`
	Name      string `json:"name"`
	Payload   []byte `json:"payload,omitempty"`
}

type Store interface {
	// Migrate creates the queue schema (idempotent).
	Migrate(ctx context.Context) error

	// EnqueueRun inserts a run.
	EnqueueRun(ctx context.Context, r *Run) error

	// DequeueRun atomically claims the highest-priority available run for owner,
	// setting a lease that expires at now+lease. Returns ErrEmpty if none.
	DequeueRun(ctx context.Context, owner string, lease time.Duration, now time.Time) (*Run, error)

	// Heartbeat extends the lease on a run still owned by owner. Returns
	// ErrLeaseLost if the run is no longer leased to owner.
	Heartbeat(ctx context.Context, runID, owner string, lease time.Duration, now time.Time) error

	// CompleteRun sets a leased run to a terminal status. Returns ErrLeaseLost if
	// owner no longer holds the lease.
	CompleteRun(ctx context.Context, runID, owner, status, lastErr string, now time.Time) error

	// RescheduleRun returns a leased run to pending, available at availableAt,
	// optionally replacing its payload (durable sleep). It does NOT touch the
	// attempts counter: durable timers and parks are normal redeliveries, not
	// failed attempts, and must not burn the retry budget.
	RescheduleRun(ctx context.Context, runID, owner string, availableAt time.Time, patch []byte, now time.Time) error

	// RetryRun returns a leased run to pending, available at availableAt, AND
	// increments its attempts counter. Unlike RescheduleRun this consumes retry
	// budget: it is the path taken after an execution error.
	RetryRun(ctx context.Context, runID, owner string, availableAt time.Time, patch []byte, now time.Time) error

	// ParkRun suspends a leased run pending a signal named waitSignal. If a
	// matching signal was already delivered for the session, the run is made
	// available immediately instead of parked (no lost-signal race).
	ParkRun(ctx context.Context, runID, owner, waitSignal string, patch []byte, now time.Time) error

	// DeliverSignal wakes parked runs of the session waiting on sig.Name and
	// returns the count awakened. If none are waiting, the signal is retained.
	DeliverSignal(ctx context.Context, sig *Signal, now time.Time) (int, error)

	// ReleaseParked promotes up to limit runs parked on waitSignal to pending.
	ReleaseParked(ctx context.Context, waitSignal string, limit int, now time.Time) (int, error)

	// RecoverOrphans re-enqueues runs whose lease expired (worker died); runs
	// past MaxAttempts are failed. Returns the number of runs recovered.
	RecoverOrphans(ctx context.Context, now time.Time) (int, error)

	// CountByStatus returns the number of runs in the given status.
	CountByStatus(ctx context.Context, status string) (int, error)

	// GetRun returns a run by ID (ErrNotFound if absent).
	GetRun(ctx context.Context, id string) (*Run, error)

	// MarkIdempotent records key; fresh is true only for the first observation.
	MarkIdempotent(ctx context.Context, key string, now time.Time) (fresh bool, err error)

	// EnqueueOutbox records an external effect. Duplicate IdempotencyKeys are
	// ignored (returns nil without inserting).
	EnqueueOutbox(ctx context.Context, e *OutboxEntry) error

	// ClaimOutbox returns up to limit pending outbox entries.
	ClaimOutbox(ctx context.Context, limit int) ([]*OutboxEntry, error)

	// MarkOutboxSent marks an entry delivered.
	MarkOutboxSent(ctx context.Context, id string, now time.Time) error

	// MarkOutboxFailed records a delivery failure, incrementing attempts. When
	// the incremented attempts reach maxAttempts the entry is dead-lettered
	// (status OutboxFailed) so it is no longer claimed; otherwise it stays
	// pending for retry. A maxAttempts <= 0 disables dead-lettering.
	MarkOutboxFailed(ctx context.Context, id, errMsg string, maxAttempts int, now time.Time) error

	// Close releases resources owned by the store.
	Close() error
}

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

type WorkerConfig struct {
	// ID uniquely identifies this worker; it owns the leases it claims.
	ID string
	// Lease is the lease TTL granted on claim and extended by each heartbeat.
	Lease time.Duration
	// Heartbeat is how often the worker extends the lease while executing.
	Heartbeat time.Duration
	// PollInterval is how long to wait after an empty dequeue before retrying.
	PollInterval time.Duration
	// Backoff computes the retry delay from the attempt count. If nil, a bounded
	// exponential backoff is used.
	Backoff func(attempt int) time.Duration
}