Skip to main content

internal/infra/pg/notificationstore.go

internal/infra/pg · 279 lines · 16 declarations · source

Declarations​

var ErrEndpointNotFound, ErrEndpointExists​

var (
	// ErrEndpointNotFound is an endpoint that is not this project's, or is not there at all. One
	// error for both, so an id cannot be used to learn what another project registered.
	ErrEndpointNotFound = errors.New("notification endpoint not found")
	// ErrEndpointExists is the same URL registered twice in one project. Refused rather than
	// duplicated, because two rows would deliver every formation twice to one listener.
	ErrEndpointExists = errors.New("that destination is already registered for this project")
)

Errors a caller can act on. Everything else is this system's problem and says so.

source

type NotificationStore​

type NotificationStore struct {
	pool   *pgxpool.Pool
	schema Schema
}

NotificationStore holds where a project wants to be told, and what is owed.

source

func NewNotificationStore​

func NewNotificationStore(pool *pgxpool.Pool, schema Schema) *NotificationStore

source

type Endpoint​

type Endpoint struct {
	ID         string     `json:"id"`
	URL        string     `json:"url"`
	CreatedAt  time.Time  `json:"created_at"`
	DisabledAt *time.Time `json:"disabled_at,omitempty"`
}

Endpoint is a destination as it is listed. The secret is absent: it is returned once, by Register, and never read back — an operator who has lost it rotates rather than looks.

source

type Registered​

type Registered struct {
	Endpoint
	// Secret signs every delivery to this destination. Shown once, like a credential, for the same
	// reason: what is stored is what we must sign with, and what is shown is what the receiver must
	// verify with, and there is no third party who needs it afterwards.
	Secret string `json:"secret"`
}

Registered is an endpoint and the one sight of its secret.

source

method NotificationStore.Register​

func (s *NotificationStore) Register(ctx context.Context, scope, url string) (Registered, error)

Register adds a destination to a project and mints its signing secret.

source

method NotificationStore.Endpoints​

func (s *NotificationStore) Endpoints(ctx context.Context, scope string) ([]Endpoint, error)

Endpoints lists a project's destinations, disabled ones included, because an operator asking "where does this go" is owed the ones that used to.

source

method NotificationStore.Disable​

func (s *NotificationStore) Disable(ctx context.Context, scope, id string) error

Disable stops delivery to a destination without forgetting it, so a parked delivery can still say where it was going.

source

method NotificationStore.Owed​

func (s *NotificationStore) Owed(ctx context.Context, scope string, formed, stored int64, parked int) (int, error)

Owed enqueues a delivery per live endpoint for a scope that has formed further than that endpoint has been told.

Why this reads the watermark rather than being called by formation​

A delivery written inside the formation transaction leaves two choices, both bad: a failed delivery rolls back a turn, so memory is lost to somebody else's broken endpoint; or the failure is swallowed, which is a promise made and not kept. Reading the watermark afterwards costs one query per pass and owes nothing to the outcome of the send.

The unique constraint on (endpoint, formed_through) is what makes this safe to call on every pass: formation runs continuously and most passes have no news, so the insert simply finds the row it would have written already there.

source

type Due​

type Due struct {
	ID            string
	Scope         string
	EndpointID    string
	URL           string
	Secret        []byte
	FormedThrough int64
	StoredThrough int64
	ParkedTurns   int
	Attempts      int
}

Due is one delivery to attempt, claimed so that two workers cannot take the same one.

source

method NotificationStore.Claim​

func (s *NotificationStore) Claim(ctx context.Context) (Due, bool, error)

Claim takes the oldest delivery that is due, or reports that nothing is.

`FOR UPDATE SKIP LOCKED` rather than a status column: a worker that dies holding a claim releases it when its connection goes, which is the same guarantee formation's scope lock relies on, and it needs no sweeper to notice.

One clock, and it is the database's​

Due-ness is decided by `now()` here rather than by a timestamp the caller supplies. The schedule is written by the database — `next_attempt_at` defaults to its clock and backoff is computed from it — so comparing it against an application's clock makes delivery depend on two machines agreeing. They do not: the database in this repository's own test environment runs about a tenth of a second ahead of its host, which is enough to make every freshly written delivery not yet due; a host a minute behind would deliver nothing at all and report no error. Whichever clock writes the schedule is the one that must read it.

source

method NotificationStore.Delivered​

func (s *NotificationStore) Delivered(ctx context.Context, id string, status int) error

Delivered closes a delivery.

source

method NotificationStore.Failed​

func (s *NotificationStore) Failed(ctx context.Context, id string, attempts, budget int,
	status int, reason string, backoff time.Duration) error

Failed records an attempt that did not land, and decides whether to wait or to stop.

Parking rather than retrying forever: an endpoint that has refused a delivery for hours is not coming back within this delivery's usefulness, and a queue that never empties is a queue nobody can read. Parked is visible, which is the point — it is a state to be asked about, not a log line.

source

type DeliveryRecord​

type DeliveryRecord struct {
	ID            string     `json:"id"`
	EndpointID    string     `json:"endpoint_id"`
	URL           string     `json:"url"`
	FormedThrough int64      `json:"formed_through"`
	StoredThrough int64      `json:"stored_through"`
	Attempts      int        `json:"attempts"`
	DeliveredAt   *time.Time `json:"delivered_at,omitempty"`
	ParkedAt      *time.Time `json:"parked_at,omitempty"`
	LastStatus    *int       `json:"last_status,omitempty"`
	LastError     *string    `json:"last_error,omitempty"`
	CreatedAt     time.Time  `json:"created_at"`
}

DeliveryRecord is one delivery as an operator or a customer reads it back. No content, because there was none: a scope, a watermark and what happened.

source

method NotificationStore.Deliveries​

func (s *NotificationStore) Deliveries(ctx context.Context, scope string, limit int) ([]DeliveryRecord, error)

Deliveries lists a project's recent deliveries, newest first.

source

func nullIfZero​

func nullIfZero(status int) any

source