notifications

package
v0.3.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 31, 2026 License: MIT Imports: 43 Imported by: 0

Documentation

Overview

Package notifications composes typed recipients, payloads and delivery channels around persistent per-channel status. It borrows existing providers, database pools, mailers, WebSocket publishers and the jobs/outbox system.

Index

Constants

View Source 
const (
	MaxBulkRecipients = 10000
	MaxBulkBatch      = 500
)

MaxBulkRecipients bounds one EnqueueMany call; MaxBulkBatch bounds the notifications stored per transaction.

View Source 
const (
	MigrationOrigin     migrate.Origin  = "foundry.notifications"
	CreateNotifications migrate.ID      = "000001_create_notifications"
	AddInboxPruneIndex  migrate.ID      = "000002_add_inbox_prune_index"
	Introduced          migrate.Version = "v0.1.0"
)
View Source 
const MaxChannels = 16
View Source 
const MaxDeliveryPage = 100

MaxDeliveryPage bounds one operator listing.

View Source 
const MaxPayloadBytes = 768 << 10
View Source 
const MaxPruneBatch = 10000

MaxPruneBatch bounds the inbox records one PruneInbox transaction deletes.

View Source 
const MaxRouteAddressBytes = 512

MaxRouteAddressBytes bounds Route.Address.

Variables

View Source 
var ErrDeliveryAnchorMoved = safeerror.Sentinel(fault.Conflict, "delivery page anchor no longer exists in this state; restart the listing")

ErrDeliveryAnchorMoved reports that the after delivery no longer exists or left the listed state, so the page position is lost. Restart the listing from the beginning (a zero after) instead of silently repeating rows.

Functions

func Migrations 

func Migrations() []migrate.Definition

Migrations are explicit ordinary migrations. Use the same schema as Config; constructing a manager never connects, migrates, resets or drops existing data.

func Module 

func Module(name foundation.ProviderID, key foundation.Key[*Manager], requires []foundation.ProviderID, construct func(foundation.Resolver) (*Manager, error)) foundation.Module

Module owns the manager. Requires must name providers owning all borrowed resources, so active notifications drain before database/transports close.

Types

type Binding 

type Binding[M model.Identifiable, K, P any] struct {
	// contains filtered or unexported fields
}

func Bind 

func Bind[M model.Identifiable, K, P any](definition Definition[P], recipient Recipient[M, K], channels ...Channel[M, P]) Binding[M, K, P]

Bind is the typed registry boundary. A channel renderer for another recipient model or notification payload cannot be passed here.

func BindOnDemand 

func BindOnDemand[P any](definition Definition[P], channels ...Channel[Route, P]) Binding[Route, string, P]

BindOnDemand binds definition to on-demand routes. Channels receive the Route as their model: use Email with a renderer addressing route.Email, or a Custom transport using route.Address. Database (inbox) channels are rejected. Capture with route.Reference(); storage, retries, delivery jobs and operator resolution work as for model recipients.

func (Binding[M, K, P]) Capture 

func (b Binding[M, K, P]) Capture(ctx context.Context, recipient model.Reference[M, K], input P, id ID[M]) (PendingNotification[M, P], error)

Capture performs no lookup or I/O. A zero ID creates one; an explicit ID is useful across caller retries. Reusing an ID with changed input fails on store.

func (Binding[M, K, P]) EnqueueMany 

func (b Binding[M, K, P]) EnqueueMany(ctx context.Context, manager *Manager, job DeliveryJob, producer *jobs.Outbox, recipients []model.Reference[M, K], input P, batch int) (BulkResult[M], error)

EnqueueMany notifies many recipients with the same input. Each recipient gets its own captured notification and delivery job; each batch of up to batch (default 100, at most MaxBulkBatch) notifications is stored together with its delivery-job outbox rows in one transaction of its own, never the caller's. Delivery then runs through the job worker. Capture validates every recipient before the first batch is stored. IDs are generated; use EnqueueManyWithIDs to supply them for retries.

func (Binding[M, K, P]) EnqueueManyWithIDs 

func (b Binding[M, K, P]) EnqueueManyWithIDs(ctx context.Context, manager *Manager, job DeliveryJob, producer *jobs.Outbox, recipients []model.Reference[M, K], ids []ID[M], input P, batch int) (BulkResult[M], error)

EnqueueManyWithIDs is EnqueueMany with caller-supplied notification IDs, one per recipient (a zero ID generates one; nil generates all). Re-running it with the same IDs and input after an uncertain batch stores each notification at most once.

func (Binding[M, K, P]) Registration 

func (b Binding[M, K, P]) Registration() Registration

func (Binding[M, K, P]) Send 

func (b Binding[M, K, P]) Send(ctx context.Context, manager *Manager, recipient model.Reference[M, K], input P) (Report[M], error)

func (Binding[M, K, P]) Validate 

func (b Binding[M, K, P]) Validate() error

type BulkResult 

type BulkResult[M any] struct {
	Stored       []ID[M]
	Unconfirmed  []ID[M]
	NotAttempted []ID[M]
}

BulkResult reports every notification ID in recipient order, grouped by outcome. Stored batches committed. Unconfirmed is the batch whose transaction failed: its commit outcome may be unknown, so reconcile those IDs (for example with Inbox.Status or Manager.Deliveries) or retry them with EnqueueManyWithIDs, which is idempotent per ID. NotAttempted lists IDs after the failed batch that were never stored.

type Channel 

type Channel[M, P any] struct {
	// contains filtered or unexported fields
}

func Custom 

func Custom[M, P, D any](id ChannelID, payload contract.JSON[D], render Renderer[M, P, D], transport Transport[D]) Channel[M, P]

func Email 

func Email[M, P any](id ChannelID, mailer *email.Mailer, render Renderer[M, P, email.Message]) Channel[M, P]

Email renders from the current recipient on the first eligible attempt. The complete message and pinned attachment references are persisted before send.

func RealtimeChannel 

func RealtimeChannel[C any, M model.Identifiable, K comparable, P, D any](id ChannelID, realtime Realtime[C, M, K, D], publisher websocket.PublisherSource, render Renderer[M, P, D]) Channel[M, P]

RealtimeChannel preserves model/payload ownership while reusing the private room declaration. Its renderer cannot choose a different recipient room.

func (Channel[M, P]) ID 

func (c Channel[M, P]) ID() ChannelID

func (Channel[M, P]) Validate 

func (c Channel[M, P]) Validate() error

type ChannelID 

type ChannelID string

type ChannelStatus 

type ChannelStatus struct {
	Channel  ChannelID
	State    State
	Attempts uint32
}

type ClientChannelInfo 

type ClientChannelInfo struct {
	ID       ChannelID           `json:"id"`
	Kind     string              `json:"kind"`
	Payload  contract.Schema     `json:"payload"`
	Realtime *ClientRealtimeInfo `json:"realtime,omitempty"`
}

type ClientNotificationInfo 

type ClientNotificationInfo struct {
	Recipient RecipientName       `json:"recipient"`
	Name      Name                `json:"name"`
	Version   Version             `json:"version"`
	Channels  []ClientChannelInfo `json:"channels"`
}

ClientNotificationInfo exposes only rendered inbox/realtime DTOs. Send inputs, email snapshots and custom transport payloads are not browser contracts.

type ClientRealtimeInfo 

type ClientRealtimeInfo struct {
	Channel websocket.ChannelID `json:"channel"`
	Event   websocket.EventID   `json:"event"`
}

ClientRealtimeInfo links a notification to its actual registered outgoing WebSocket event. The WebSocket registry still owns room/access/replay metadata.

type Config 

type Config struct {
	Schema    string
	Clock     clock.Clock
	MaxActive int
	Timeout   time.Duration
}

func DefaultConfig 

func DefaultConfig() Config

func (Config) Validate 

func (c Config) Validate() error

type DatabaseChannel 

type DatabaseChannel[M, P, D any] struct {
	// contains filtered or unexported fields
}

DatabaseChannel retains the concrete inbox representation for typed decoding.

func Database 

func Database[M, P, D any](id ChannelID, payload contract.JSON[D], render Renderer[M, P, D]) DatabaseChannel[M, P, D]

func (DatabaseChannel[M, P, D]) Channel 

func (c DatabaseChannel[M, P, D]) Channel() Channel[M, P]

func (DatabaseChannel[M, P, D]) Decode 

func (c DatabaseChannel[M, P, D]) Decode(ctx context.Context, definition Definition[P], record Record[M]) (D, error)

type Definition 

type Definition[P any] struct {
	// contains filtered or unexported fields
}

Definition is the shared versioned input contract. Bind it to one or more typed recipient declarations; use generated DTO contracts as the schema SSOT.

func Define 

func Define[P any](name Name, version Version, payload contract.JSON[P]) Definition[P]

func (Definition[P]) Name 

func (d Definition[P]) Name() Name

func (Definition[P]) Validate 

func (d Definition[P]) Validate() error

func (Definition[P]) Version 

func (d Definition[P]) Version() Version

type DeliveryContext 

type DeliveryContext struct {
	Notification NotificationID
	Delivery     DeliveryID
	Name         Name
	Version      Version
	Channel      ChannelID
}

DeliveryContext gives renderers stable metadata. Renderers run once before durable preparation (possibly concurrently before one snapshot wins). They must be pure: external side effects belong to the transport, after claiming.

type DeliveryID 

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

DeliveryID is a stable external idempotency key, derived from notification and channel identities. It is not recipient authentication or a provider receipt.

func ParseDeliveryID 

func ParseDeliveryID(text string) (DeliveryID, error)

ParseDeliveryID restores an operator-supplied delivery identity. It grants no authority; operator tooling must authorize its callers.

func (DeliveryID) Format 

func (DeliveryID) Format(s fmt.State, _ rune)

func (DeliveryID) String 

func (id DeliveryID) String() string

func (DeliveryID) Validate 

func (id DeliveryID) Validate() error

type DeliveryInfo 

type DeliveryInfo struct {
	ID           DeliveryID     `json:"-"`
	Key          string         `json:"id"`
	Notification NotificationID `json:"notification_id"`
	Channel      ChannelID      `json:"channel"`
	Kind         string         `json:"kind"`
	State        State          `json:"state"`
	Attempts     uint32         `json:"attempts"`
	UpdatedAt    time.Time      `json:"updated_at"`
}

DeliveryInfo is safe operator metadata for one channel delivery: no payload, route or rendered content.

type DeliveryJob 

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

func DefineDeliveryJob 

func DefineDeliveryJob(name jobs.Name, policy jobs.Policy) DeliveryJob

func (DeliveryJob) Declare 

func (d DeliveryJob) Declare(manager *Manager) (jobs.Declaration, error)

type DeliveryRequest 

type DeliveryRequest struct {
	ID NotificationID `json:"id"`
}

DeliveryRequest is the job's small, versioned transport DTO. All private input, rendered content and per-channel outcomes remain in notification persistence.

func (DeliveryRequest) Format 

func (DeliveryRequest) Format(s fmt.State, _ rune)

type ID 

type ID[M any] = model.ID[NotificationOf[M]]

func NewID 

func NewID[M any]() (ID[M], error)

func ParseID 

func ParseID[M any](text string) (ID[M], error)

type Inbox 

type Inbox[M model.Identifiable, K any] struct {
	// contains filtered or unexported fields
}

Inbox obtains its recipient exclusively from the verified guard in ctx. It deliberately has no ListFor/ReadFor overload taking a client-supplied identity.

func (Inbox[M, K]) Delete 

func (i Inbox[M, K]) Delete(ctx context.Context, id ID[M]) (bool, error)

Delete removes one inbox record of the current recipient. It returns false for both an absent ID and another recipient's ID. The notification and its delivery states are kept, so a retry never recreates the record.

func (Inbox[M, K]) List 

func (i Inbox[M, K]) List(ctx context.Context, page orm.PageRequest, unreadOnly bool) (orm.PageResult[Record[M]], error)

func (Inbox[M, K]) MarkAllRead 

func (i Inbox[M, K]) MarkAllRead(ctx context.Context) (int, error)

MarkAllRead marks every unread record of the current recipient read in bounded transactions, oldest first, and returns the number changed. Records that were already read keep their first read instant. A failed batch leaves earlier committed batches in place.

func (Inbox[M, K]) MarkRead 

func (i Inbox[M, K]) MarkRead(ctx context.Context, id ID[M]) (bool, error)

MarkRead returns false for both an absent ID and another recipient's ID. Marking read repeatedly preserves the first read instant. MarkUnread clears it.

func (Inbox[M, K]) MarkUnread 

func (i Inbox[M, K]) MarkUnread(ctx context.Context, id ID[M]) (bool, error)

func (Inbox[M, K]) Status 

func (i Inbox[M, K]) Status(ctx context.Context, id ID[M]) (Report[M], error)

Status includes non-inbox channels and terminal ineligibility. Like inbox operations, it checks the guard, recipient scope and stored key before lookup.

func (Inbox[M, K]) UnreadCount 

func (i Inbox[M, K]) UnreadCount(ctx context.Context) (int64, error)

type Manager 

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

Manager borrows every dependency. It owns admission and active call lifetimes; Close cancels work and Done closes only after actual callback/resource exit. Never copy a Manager. Callbacks must honor cancellation and be concurrency-safe.

func New 

func New(db *database.DB, registry *Registry, config Config) (*Manager, error)

func (*Manager) Close 

func (m *Manager) Close(ctx context.Context) error

func (*Manager) Deliver 

func (m *Manager) Deliver(ctx context.Context, id NotificationID) ([]ChannelStatus, error)

Deliver attempts every retryable channel of one stored notification now, as its delivery job does, and reports the resulting channel states.

func (*Manager) Deliveries 

func (m *Manager) Deliveries(ctx context.Context, state State, limit int, after DeliveryID) ([]DeliveryInfo, error)

Deliveries lists up to limit deliveries in state, oldest update first. Pass the last returned ID as after to continue; resolving rows between pages can move them, and a moved or deleted anchor returns ErrDeliveryAnchorMoved. Use it to find Running and Uncertain deliveries that need an operator decision.

func (*Manager) Done 

func (m *Manager) Done() <-chan struct{}

func (*Manager) Format 

func (*Manager) Format(s fmt.State, _ rune)

func (*Manager) PruneInbox 

func (m *Manager) PruneInbox(ctx context.Context, cutoff time.Time, readOnly bool, limit int) (int, error)

PruneInbox deletes up to limit inbox records created before cutoff across every recipient, oldest first; readOnly keeps unread records. It returns the number deleted; repeat until fewer than limit are deleted. The notifications and their delivery states are kept, so retries never recreate a pruned record. It is an operator operation: authorize its callers.

func (*Manager) ResolveDelivery 

func (m *Manager) ResolveDelivery(ctx context.Context, id DeliveryID, expected State, resolution Resolution) (bool, error)

ResolveDelivery applies a resolution only while the delivery is still in the expected Running or Uncertain state; changed=false means it moved meanwhile. It also invalidates the running claim, so a process that was still active cannot overwrite the decision. Resolve Running only after confirming its process is gone: a live call may still reach the provider. Resend risks a duplicate if the provider did accept; follow it with Deliver or a job.

type Name 

type Name string

type Notification 

type Notification struct{}

type NotificationID 

type NotificationID = model.ID[Notification]

NotificationID is the heterogeneous transport identity, not inbox authority.

type NotificationOf 

type NotificationOf[M any] struct {
	// contains filtered or unexported fields
}

type Outcome 

type Outcome uint8

Outcome is returned by custom transports. Retry means definite nonacceptance; never use it after a timeout that may have reached the remote service. 

const (
	Accepted Outcome = iota + 1
	Retry
	Reject
	Unknown
)

type PendingNotification 

type PendingNotification[M, P any] struct {
	// contains filtered or unexported fields
}

PendingNotification freezes input and recipient identity, not routes or preferences. Keep it for deliberate transaction/send retries using the same ID. It is a runtime handle; queue through Enqueue instead of serializing it.

func (PendingNotification[M, P]) Enqueue 

func (p PendingNotification[M, P]) Enqueue(ctx context.Context, tx *database.Tx, manager *Manager, job DeliveryJob, producer *jobs.Outbox) (outbox.ID[DeliveryRequest], error)

Enqueue atomically persists the captured notification and an ordinary job outbox row inside the business transaction. Use the SAME schema for notification and outbox migrations. The savepoint prevents partial work even if callers ignore an error. The returned ID is provisional until the outer commit.

func (PendingNotification[M, P]) Format 

func (PendingNotification[M, P]) Format(s fmt.State, _ rune)

func (PendingNotification[M, P]) ID 

func (p PendingNotification[M, P]) ID() ID[M]

func (PendingNotification[M, P]) MarshalJSON 

func (PendingNotification[M, P]) MarshalJSON() ([]byte, error)

func (PendingNotification[M, P]) Send 

func (p PendingNotification[M, P]) Send(ctx context.Context, manager *Manager) (Report[M], error)

Send durably records the notification before delivery. Reusing this pending handle attempts only retryable channels. Inspect Report even when err != nil.

type Realtime 

type Realtime[C any, M model.Identifiable, K comparable, D any] struct {
	// contains filtered or unexported fields
}

Realtime owns an ownership-enforcing channel and server-only event. Register Registration with the existing WebSocket registry before constructing its hub. No arbitrary channel can be substituted into a notification delivery binding.

func DefineRealtime 

func DefineRealtime[C any, M model.Identifiable, K comparable, D any](recipient Recipient[M, K], channel websocket.ChannelID, rooms websocket.Rooms[K], event websocket.EventID, data contract.JSON[D]) Realtime[C, M, K, D]

func (Realtime[C, M, K, D]) Registration 

func (r Realtime[C, M, K, D]) Registration() websocket.Registration

func (Realtime[C, M, K, D]) WithReplay 

func (r Realtime[C, M, K, D]) WithReplay(config websocket.ReplayConfig) Realtime[C, M, K, D]

type RealtimeMessage 

type RealtimeMessage[D any] struct {
	ID      NotificationID `json:"id"`
	Name    Name           `json:"name"`
	Version Version        `json:"version"`
	Data    D              `json:"data"`
}

RealtimeMessage carries a stable notification ID across client reconnects; clients should ignore an ID they already handled. Publication is local admission/distributed attempt, not an end-user receipt.

func (RealtimeMessage[D]) Format 

func (RealtimeMessage[D]) Format(s fmt.State, _ rune)

type Recipient 

type Recipient[M model.Identifiable, K any] struct {
	// contains filtered or unexported fields
}

Recipient binds one model/key authority and inbox guard. Preferences run with a freshly loaded eligible model before EACH channel attempt, including retries. A false preference permanently skips that channel for this notification.

func DefineRecipient 

func DefineRecipient[M model.Identifiable, K any](name RecipientName, provider auth.Provider[M, K], guard auth.Guard[M], preferences func(context.Context, M, Name, ChannelID) (bool, error)) Recipient[M, K]

func (Recipient[M, K]) Format 

func (Recipient[M, K]) Format(s fmt.State, _ rune)

func (Recipient[M, K]) Inbox 

func (r Recipient[M, K]) Inbox(manager *Manager) (Inbox[M, K], error)

func (Recipient[M, K]) Name 

func (r Recipient[M, K]) Name() RecipientName

func (Recipient[M, K]) Validate 

func (r Recipient[M, K]) Validate() error

type RecipientName 

type RecipientName string
const OnDemandRecipient RecipientName = "foundry.on-demand"

OnDemandRecipient names the shared recipient of on-demand bindings.

type Record 

type Record[M any] struct {
	ID        ID[M]
	Name      Name
	Version   Version
	CreatedAt temporal.DateTime
	ReadAt    value.Nullable[temporal.DateTime]
	// contains filtered or unexported fields
}

Record is a recipient-owned inbox result. Decode its private data using the declared DatabaseChannel, then construct a typed HTTP response DTO. Persistence rows and unrelated notification payloads never become dynamic response maps.

func (Record[M]) Format 

func (Record[M]) Format(s fmt.State, _ rune)

func (Record[M]) MarshalJSON 

func (Record[M]) MarshalJSON() ([]byte, error)

type Registration 

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

type Registry 

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

Registry is immutable and owns no resources. Duplicate recipient or payload names require reusing the original declaration, including across versions.

func NewRegistry 

func NewRegistry(items ...Registration) (*Registry, error)

func (*Registry) ClientDescriptions 

func (r *Registry) ClientDescriptions() ([]ClientNotificationInfo, error)

ClientDescriptions returns deterministic owned metadata from the same typed bindings used for delivery, without resolving recipients or running renderers. Bindings with no client-facing channel are omitted.

type Renderer 

type Renderer[M, P, D any] func(context.Context, M, DeliveryContext, P) (D, error)

type Report 

type Report[M any] struct {
	ID       ID[M]
	Channels []ChannelStatus
}

func (Report[M]) Format 

func (Report[M]) Format(s fmt.State, _ rune)

func (Report[M]) Retryable 

func (r Report[M]) Retryable() bool

type Resolution 

type Resolution string

Resolution is an operator's decision about a Running or Uncertain delivery after checking the provider. Delivered records confirmed acceptance; Rejected records confirmed non-acceptance without sending again; Resend returns the frozen output to Prepared so the next delivery attempt sends it again. 

const (
	ResolveDelivered Resolution = "delivered"
	ResolveRejected  Resolution = "rejected"
	ResolveResend    Resolution = "resend"
)

type Route 

type Route struct {
	Email   email.Address `json:"email,omitzero"`
	Address string        `json:"address,omitempty"`
}

Route is an on-demand notification recipient without an application model: an email address and/or an opaque address for custom channels (for example a phone number or webhook URL). It is stored with the notification as its identity, so treat it as private data. Routes have no inbox, preferences or eligibility check: the caller chose the destination explicitly.

func (Route) FoundryIdentity 

func (r Route) FoundryIdentity() (model.Identity, error)

func (Route) Reference 

func (r Route) Reference() (model.Reference[Route, string], error)

Reference is the route's typed identity for Binding.Capture and Send.

func (Route) Validate 

func (r Route) Validate() error

type State 

type State string

State is persisted independently for each channel. Running may mean an active call or a process lost after claiming; it is never automatically reclaimed. 

const (
	Pending    State = "pending"
	Prepared   State = "prepared"
	Running    State = "running"
	Delivered  State = "delivered"
	Skipped    State = "skipped"
	Ineligible State = "ineligible"
	Rejected   State = "rejected"
	Uncertain  State = "uncertain"
)

func (State) Retryable 

func (s State) Retryable() bool

type Transport 

type Transport[D any] interface {
	Deliver(context.Context, DeliveryID, D) (Outcome, error)
}

Transport is the focused custom-channel boundary. It receives a freshly decoded rendered DTO and stable delivery identity. An error is ambiguous; explicit Retry with nil error promises the service definitely did not accept.

type Version 

type Version uint32

Source Files

View all Source files

Directories

Path Synopsis
Package command supplies application-owned notification delivery operations.
 Package command supplies application-owned notification delivery operations.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.