Documentation
¶
Overview ¶
Package store is the PostgreSQL persistence layer: the event archive and the delivery queue. The queue is plain SQL — due rows are claimed with FOR UPDATE SKIP LOCKED, so any number of workers can run concurrently without double delivery and without any broker beside Postgres.
Index ¶
- Constants
- Variables
- type Claimed
- type Delivery
- type Event
- type EventFilter
- type EventSummary
- type Store
- func (s *Store) ClaimDue(ctx context.Context, limit int, visibility time.Duration) ([]Claimed, error)
- func (s *Store) Close()
- func (s *Store) GetEvent(ctx context.Context, id uuid.UUID) (*Event, []Delivery, error)
- func (s *Store) InsertEvent(ctx context.Context, ev *Event, targets []string) error
- func (s *Store) ListEvents(ctx context.Context, f EventFilter) ([]EventSummary, error)
- func (s *Store) MarkDelivered(ctx context.Context, id uuid.UUID, attempt int, responseStatus int) error
- func (s *Store) MarkFailed(ctx context.Context, id uuid.UUID, attempt int, lastErr string, ...) error
- func (s *Store) Migrate(ctx context.Context) ([]string, error)
- func (s *Store) PendingMigrations(ctx context.Context) ([]string, error)
- func (s *Store) Ping(ctx context.Context) error
- func (s *Store) Prune(ctx context.Context, cutoff time.Time, dryRun bool) (int64, error)
- func (s *Store) QueueDepth(ctx context.Context) (int64, error)
- func (s *Store) ReplayEvent(ctx context.Context, eventID uuid.UUID, targets []string) (int64, error)
- func (s *Store) ReplayRange(ctx context.Context, source string, from, to time.Time, targets []string) (int64, error)
Constants ¶
const ( VerifyVerified = "verified" VerifyRejected = "rejected" VerifyUnverified = "unverified" )
Event statuses (verify_status column).
const ( StatusPending = "pending" StatusRetrying = "retrying" StatusInflight = "inflight" StatusDelivered = "delivered" StatusDead = "dead" )
Delivery statuses.
Variables ¶
var ErrNotFound = errors.New("not found")
ErrNotFound is returned when an event id does not exist.
Functions ¶
This section is empty.
Types ¶
type Claimed ¶
type Claimed struct {
DeliveryID uuid.UUID
EventID uuid.UUID
Target string
Attempt int // attempts before this one
Source string
Body []byte
ContentType string
}
Claimed is a delivery leased to a worker, joined with what the worker needs to perform the HTTP call.
type Delivery ¶
type Delivery struct {
ID uuid.UUID `json:"id"`
EventID uuid.UUID `json:"event_id"`
Target string `json:"target"`
Status string `json:"status"`
Kind string `json:"kind"`
Attempt int `json:"attempt"`
NextAttemptAt time.Time `json:"next_attempt_at"`
LastError string `json:"last_error,omitempty"`
ResponseStatus *int `json:"response_status,omitempty"`
DeliveredAt *time.Time `json:"delivered_at,omitempty"`
CreatedAt time.Time `json:"created_at"`
}
type Event ¶
type Event struct {
ID uuid.UUID `json:"id"`
Source string `json:"source"`
ReceivedAt time.Time `json:"received_at"`
Headers map[string][]string `json:"headers,omitempty"`
Body []byte `json:"-"`
ContentType string `json:"content_type"`
BodySize int `json:"body_size"`
VerifyStatus string `json:"verify_status"`
VerifyReason string `json:"verify_reason,omitempty"`
}
type EventFilter ¶
type EventFilter struct {
Source string
Status string
Since time.Time
Until time.Time
Limit int
Offset int
}
EventFilter narrows ListEvents. Status filters by delivery state (pending/delivered/dead) or by verify_status (rejected).
type EventSummary ¶
type EventSummary struct {
ID uuid.UUID `json:"id"`
Source string `json:"source"`
ReceivedAt time.Time `json:"received_at"`
ContentType string `json:"content_type"`
BodySize int `json:"body_size"`
VerifyStatus string `json:"verify_status"`
InProgress int `json:"in_progress"`
Delivered int `json:"delivered"`
Dead int `json:"dead"`
}
EventSummary is a list row: the event plus aggregate delivery counts.
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
func (*Store) ClaimDue ¶
func (s *Store) ClaimDue(ctx context.Context, limit int, visibility time.Duration) ([]Claimed, error)
ClaimDue leases up to limit due deliveries to the calling worker.
Due means: pending/retrying with next_attempt_at in the past, or inflight rows whose lease expired (claimed_at older than visibility) — the worker that claimed them crashed mid-delivery. FOR UPDATE SKIP LOCKED makes concurrent claimers never block or double-claim; the visibility timeout makes crashed claims re-deliverable (at-least-once, never lost).
func (*Store) GetEvent ¶
GetEvent returns the full event including body, headers and its deliveries.
func (*Store) InsertEvent ¶
InsertEvent persists the event and, unless it was rejected, fans out one pending delivery per target — all in a single transaction, so an event is never visible without its deliveries.
func (*Store) ListEvents ¶
func (s *Store) ListEvents(ctx context.Context, f EventFilter) ([]EventSummary, error)
ListEvents returns newest-first event summaries with delivery counts. The LEFT JOIN is aggregated per event id, so the join cannot multiply rows.
func (*Store) MarkDelivered ¶
func (s *Store) MarkDelivered(ctx context.Context, id uuid.UUID, attempt int, responseStatus int) error
MarkDelivered finalizes a successful delivery.
func (*Store) MarkFailed ¶
func (s *Store) MarkFailed(ctx context.Context, id uuid.UUID, attempt int, lastErr string, responseStatus *int, dead bool, nextAttemptAt time.Time) error
MarkFailed records a failed attempt. When dead is false the delivery is rescheduled for nextAttemptAt; when true it goes to the dead letter state and stays visible until replayed or pruned.
func (*Store) Migrate ¶
Migrate applies embedded migrations in filename order. Migrations are forward-only; each file runs once inside its own transaction and is recorded in schema_migrations.
func (*Store) PendingMigrations ¶
PendingMigrations lists embedded migrations not yet applied (for --dry-run).
func (*Store) Prune ¶
Prune deletes events received before cutoff (deliveries cascade). With dryRun it only counts.
func (*Store) QueueDepth ¶
QueueDepth returns how many deliveries are waiting or in flight. Used by tests and by serve's periodic queue log line.
func (*Store) ReplayEvent ¶
func (s *Store) ReplayEvent(ctx context.Context, eventID uuid.UUID, targets []string) (int64, error)
ReplayEvent enqueues fresh deliveries for one event, one per target. Rejected events can be replayed too — replay is an explicit operator decision made after inspecting the event.
func (*Store) ReplayRange ¶
func (s *Store) ReplayRange(ctx context.Context, source string, from, to time.Time, targets []string) (int64, error)
ReplayRange enqueues replay deliveries for every event of a source in [from, to]. Rejected events are skipped: they failed signature verification and never earned deliveries — replay them individually if you really mean it.
Source Files