events

type Delivery ¶

type Delivery struct {
	ID             string
	EventID        string
	SubscriptionID string
	URL            string
	Secret         string
	Payload        []byte
	Status         DeliveryStatus
	Attempts       int
	LastError      string
	NextAttemptAt  time.Time
	CreatedAt      time.Time
	UpdatedAt      time.Time
}

Delivery is one durable outbox row: a single event bound to a single subscription, with its attempt bookkeeping. The payload is the exact signed bytes so a retry re-sends an identical body (and identical signature), which is what makes delivery idempotent by event id on the receiving end.

type DeliveryStatus ¶

type DeliveryStatus string

DeliveryStatus is the lifecycle of one outbox row — one (event, subscription) delivery attempt-set.

const (
	// StatusPending means the delivery is awaiting its first attempt or a
	// retry whose NextAttemptAt has not yet arrived.
	StatusPending DeliveryStatus = "pending"
	// StatusDelivered means the subscriber acknowledged the webhook (2xx).
	StatusDelivered DeliveryStatus = "delivered"
	// StatusFailed means the delivery exhausted its retry budget and was
	// abandoned. Failures are surfaced to the operator (audit/log), never
	// silently dropped.
	StatusFailed DeliveryStatus = "failed"
)

type Discard ¶

type Discard struct{}

Discard is the no-op Publisher used when outbound eventing is disabled. It validates nothing and stores nothing, so emitting is free.

func (Discard) Emit(context.Context, Event) error

type Event ¶

type Event struct {
	ID         string    `json:"id"`
	Type       EventType `json:"type"`
	ProjectID  string    `json:"project_id"`
	TenantID   string    `json:"tenant_id,omitempty"`
	OccurredAt time.Time `json:"occurred_at"`
	User       User      `json:"user"`
}

Event is the typed payload emitted by the service layer and delivered to subscribers. ID is a unique, caller-supplied identifier used for at-least-once idempotency: a subscriber that has already processed an ID can safely ignore a redelivery, and the delivery engine never enqueues the same (event, subscription) pair twice.

type EventType ¶

type EventType string

EventType enumerates the user-lifecycle events the service emits. The string values are stable wire identifiers: they appear in the webhook payload's "type" field and are matched against a subscription's event filter, so they must never be renumbered or repurposed (append only).

const (
	// EventUserCreated is emitted after a new user account is persisted.
	EventUserCreated EventType = "user.created"
	// EventUserUpdated is emitted after a user's mutable profile fields
	// (name, email, status transitions other than deactivation) change.
	EventUserUpdated EventType = "user.updated"
	// EventUserDeactivated is emitted after a user is deactivated (status
	// set to a non-active value, or the account deleted). This is the
	// deprovisioning signal outbound SCIM connectors act on.
	EventUserDeactivated EventType = "user.deactivated"
)

func (t EventType) Valid() bool

type FailureHook ¶

type FailureHook func(d *Delivery)

FailureHook is called once when a delivery exhausts its retry budget, so the composition root can surface the abandonment via audit/metrics rather than swallowing it. It must not block.

type IDFunc ¶

type IDFunc func() string

IDFunc generates a unique delivery-row id. Injectable so tests get deterministic ids; production uses a random generator supplied by the composition root.

type MemoryOutbox ¶

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

MemoryOutbox is an in-process OutboxStore. It is the differential reference the SQL stores are held to and backs unit tests + single-node deployments.

func NewMemoryOutbox() *MemoryOutbox

func (m *MemoryOutbox) AddSubscription(s Subscription)

func (m *MemoryOutbox) ClaimDue(_ context.Context, now time.Time, limit int) ([]*Delivery, error)

func (m *MemoryOutbox) EnqueueDelivery(_ context.Context, d *Delivery) error

func (m *MemoryOutbox) Get(id string) (Delivery, bool)

func (m *MemoryOutbox) ListActiveSubscriptions(_ context.Context, projectID string) ([]Subscription, error)

func (m *MemoryOutbox) MarkDelivered(_ context.Context, id string, at time.Time) error

func (m *MemoryOutbox) MarkFailed(_ context.Context, id string, attempts int, lastErr string) error

func (m *MemoryOutbox) Reschedule(_ context.Context, id string, attempts int, lastErr string, nextAttemptAt time.Time) error

type OutboxPublisher ¶

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

OutboxPublisher is the durable Publisher: Emit fans an event out to every matching active subscription by writing one pending outbox row per subscription, then returns. A background Worker (Run) delivers the rows. Because Emit only writes durable rows, a slow or down subscriber never adds latency to — or fails — the originating RPC.

func NewOutboxPublisher(store OutboxStore, newID IDFunc, now func() time.Time, logger *zap.Logger) *OutboxPublisher

func (p *OutboxPublisher) Emit(ctx context.Context, e Event) error

type OutboxStore ¶

type OutboxStore interface {
	// EnqueueDelivery durably stores a pending delivery. It returns
	// ErrDeliveryExists when a row for the same (EventID, SubscriptionID)
	// already exists, so the caller's fan-out is idempotent.
	EnqueueDelivery(ctx context.Context, d *Delivery) error

	// ClaimDue returns up to limit pending deliveries whose NextAttemptAt
	// is at or before now, marking each claimed so a concurrent worker
	// does not pick the same row. The engine processes the returned batch
	// and reports the outcome via MarkDelivered / Reschedule / MarkFailed.
	ClaimDue(ctx context.Context, now time.Time, limit int) ([]*Delivery, error)

	// MarkDelivered records a successful delivery (2xx ack).
	MarkDelivered(ctx context.Context, id string, at time.Time) error

	// Reschedule records a failed attempt and sets the next retry time.
	Reschedule(ctx context.Context, id string, attempts int, lastErr string, nextAttemptAt time.Time) error

	// MarkFailed records a delivery that exhausted its retry budget.
	MarkFailed(ctx context.Context, id string, attempts int, lastErr string) error

	// ListActiveSubscriptions returns every active subscription in the
	// given project so the engine can fan an event out.
	ListActiveSubscriptions(ctx context.Context, projectID string) ([]Subscription, error)
}

OutboxStore is the persistence boundary for the delivery engine. An in-memory implementation (MemoryOutbox) backs tests and single-node runs; a SQL-backed store (postgres/sqlite) implements the same contract for durable, multi-replica delivery. All methods must be safe for concurrent use.

type Publisher ¶

type Publisher interface {
	// Emit records an event for delivery. It returns an error only when
	// the event itself is invalid or the durable write fails; a delivery
	// failure to a downstream subscriber is never surfaced here (it is
	// retried by the worker). A no-op publisher always returns nil.
	Emit(ctx context.Context, e Event) error
}

Publisher is the pluggable eventing sink the service layer holds. The service calls Emit on every user-mutation path; the implementation decides what (if anything) happens. The default (Discard) does nothing.

Emit must be safe for concurrent use and must not block on network I/O: the outbox-backed implementation only writes a durable row and lets a background worker handle delivery, so a slow or unreachable subscriber never adds latency to (or fails) the originating RPC.

type RetryPolicy ¶

type RetryPolicy struct {
	MaxAttempts int
	BaseDelay   time.Duration
	MaxDelay    time.Duration
}

RetryPolicy bounds delivery retries. After MaxAttempts failed attempts a delivery is abandoned (StatusFailed) and surfaced via FailureHook. The backoff is exponential — BaseDelay * 2^(attempt-1) — capped at MaxDelay.

func DefaultRetryPolicy() RetryPolicy

type Sender ¶

type Sender interface {
	Deliver(ctx context.Context, d *Delivery) error
}

Sender delivers one signed webhook. The default httpSender POSTs the payload; tests inject a fake. Deliver returns nil on a 2xx acknowledgement and a non-nil error otherwise (so the worker retries).

func NewHTTPSender(client *http.Client) Sender

type Subscription ¶

type Subscription struct {
	ID         string
	ProjectID  string
	TenantID   string
	URL        string
	Secret     string
	EventTypes []EventType // empty ⇒ all types
	Active     bool
}

Subscription is a per-tenant webhook endpoint. The delivery engine fans an event out to every active subscription in the event's project whose EventTypes filter matches (an empty filter matches all types). Secret is the HMAC key used to sign the payload so the subscriber can verify the webhook originated from this server.

type User ¶

type User struct {
	ID            string `json:"id"`
	Email         string `json:"email"`
	Name          string `json:"name,omitempty"`
	Status        string `json:"status,omitempty"`
	EmailVerified bool   `json:"email_verified"`
}

User is the subset of a user record carried in a lifecycle event. It is deliberately small and free of secrets (no password hash, no tokens): downstream SaaS provisioning needs identity and status, nothing more.

type Worker ¶

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

Worker drains the outbox: it claims due deliveries, sends each, and records the outcome (delivered / rescheduled with backoff / failed). It is safe to run one Worker per replica against a shared durable store — ClaimDue prevents two workers from sending the same row.

func NewWorker(cfg WorkerConfig) *Worker

func (w *Worker) ProcessOnce(ctx context.Context) error

func (w *Worker) Run(ctx context.Context) error

type WorkerConfig ¶

type WorkerConfig struct {
	Store       OutboxStore
	Sender      Sender
	Policy      RetryPolicy
	Now         func() time.Time
	Logger      *zap.Logger
	Batch       int
	Interval    time.Duration
	FailureHook FailureHook
}

WorkerConfig configures a Worker. Store and Sender are required; the rest have sane defaults.