Documentation
¶
Overview ¶
Package outbox builds the rows the FlowCatalyst outbox writers store in outbox_messages: the row columns, the INSERT statement, and the payload of each row type (EVENT, AUDIT_LOG, DISPATCH_JOB). It does no I/O; the drivers are outboxpgx (pgx, the usecasepgx Sink) and outboxsql (database/sql).
The rows are the shared outbox contract: the platform's outbox processor sends each row's payload text unchanged, as one element of {"items":[...]}, to POST /api/events/batch, /api/audit-logs/batch or /api/dispatch-jobs/batch. The payloads use the batch endpoints' camelCase members and match the application framework's conformance vector (flowcatalyst-framework conformance/outbox/outbox.json; a copy is pinned in testdata/) under that vector's comparison rule, with one deliberate difference: an event's data and an audit row's operationData are written as the JSON value itself, never as a string holding encoded JSON (the other SDKs encode them; the platform unwraps those at ingest).
Index ¶
- Constants
- func AuditPayload(event usecase.DomainEvent, command any, applicationCode, clientCode string) ([]byte, error)
- func EventPayload(event usecase.DomainEvent, clientCode string) ([]byte, error)
- func FormatInstant(t time.Time) string
- func InsertStatement(table string, style Placeholder) (string, error)
- func PushInsertStatement(table string) (string, error)
- type AuditDataProvider
- type AuditOperationNamer
- type ClientCoded
- type Config
- type DispatchJob
- func (j DispatchJob) DataOnlyEnabled() DispatchJob
- func (j DispatchJob) Resolved() (DispatchJob, error)
- func (j DispatchJob) Validate() error
- func (j DispatchJob) WithClientCode(code string) DispatchJob
- func (j DispatchJob) WithConnectionID(id string) DispatchJob
- func (j DispatchJob) WithCorrelationID(id string) DispatchJob
- func (j DispatchJob) WithDescriptor(d string) DispatchJob
- func (j DispatchJob) WithDispatchPoolID(id string) DispatchJob
- func (j DispatchJob) WithEventID(id string) DispatchJob
- func (j DispatchJob) WithExternalID(id string) DispatchJob
- func (j DispatchJob) WithID(id string) DispatchJob
- func (j DispatchJob) WithIdempotencyKey(k string) DispatchJob
- func (j DispatchJob) WithJSONPayload(v any) (DispatchJob, error)
- func (j DispatchJob) WithKind(kind string) DispatchJob
- func (j DispatchJob) WithMaxRetries(n int) DispatchJob
- func (j DispatchJob) WithMessageGroup(g string) DispatchJob
- func (j DispatchJob) WithMetadata(key, value string) DispatchJob
- func (j DispatchJob) WithMode(mode string) DispatchJob
- func (j DispatchJob) WithPayload(text, contentType string) DispatchJob
- func (j DispatchJob) WithPlatformClientID(id string) DispatchJob
- func (j DispatchJob) WithQueue(q string) DispatchJob
- func (j DispatchJob) WithSequence(n int) DispatchJob
- func (j DispatchJob) WithServiceAccountID(id string) DispatchJob
- func (j DispatchJob) WithSource(source string) DispatchJob
- func (j DispatchJob) WithSubject(subject string) DispatchJob
- func (j DispatchJob) WithSubscriptionID(id string) DispatchJob
- func (j DispatchJob) WithTimeoutSeconds(n int) DispatchJob
- type MetadataEntry
- type Placeholder
- type Push
- type PushError
- type Row
Constants ¶
const ( DispatchKindEvent = "EVENT" DispatchKindTask = "TASK" DispatchModeImmediate = "IMMEDIATE" DispatchModeNextOnError = "NEXT_ON_ERROR" DispatchModeBlockOnError = "BLOCK_ON_ERROR" DispatchQueueDefault = "DEFAULT" DispatchQueueHighPriority = "HIGH_PRIORITY" )
Dispatch-job kinds, modes and queues the batch endpoint accepts.
const ( TypeEvent = "EVENT" TypeAuditLog = "AUDIT_LOG" TypeDispatchJob = "DISPATCH_JOB" )
Row types: the outbox processor maps the type column to an endpoint by exact string.
const ( // MaxPushTopicLength is the longest topic, in characters. MaxPushTopicLength = 255 // MaxPushTypeLength is the longest type, in characters. MaxPushTypeLength = 100 // DefaultPushMaxDataBytes is the default data limit: push.max-data-bytes. DefaultPushMaxDataBytes = 16384 // MaxPushDataBytes is the most the data limit may be set to. MaxPushDataBytes = 65536 )
The limits of a push (push contract section 1).
const ( PushTopicInvalid = "TOPIC_INVALID" PushTypeInvalid = "TYPE_INVALID" PushTypeReserved = "TYPE_RESERVED" PushDataInvalid = "DATA_INVALID" PushDataNotObject = "DATA_NOT_OBJECT" PushDataTooLarge = "DATA_TOO_LARGE" PushIDInvalid = "PUSH_ID_INVALID" PushClientInvalid = "CLIENT_INVALID" )
The codes of a refused push: the codes of the framework's conformance/push/topics.json, which the platform's ingest answers with as well.
const DefaultTable = "outbox_messages"
DefaultTable is the outbox table every SDK writes.
const InsertSQL = "INSERT INTO outbox_messages (\n" +
" id, type, message_group, payload, status, created_at, updated_at,\n" +
" client_id, payload_size, headers\n" +
")\n" +
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
InsertSQL is the outbox INSERT, byte for byte the framework's (conformance/sql/outbox/insert.sql), with ? placeholders. The arguments are Row.Args. retry_count and error_message are left to their column defaults.
const PlatformClientCode = "_platform"
PlatformClientCode is the client code of platform-scoped writers. Events and audit rows still carry it; dispatch jobs do not (no clientCode).
const PushNotifyChannel = "fc_outbox_push"
PushNotifyChannel is the Postgres notification channel a PUSH row's writer NOTIFYs in the same statement as the insert (PushInsertStatement), so a processor that listens on it delivers the push at once instead of at its next poll. Writers that cannot notify (MySQL, SQLite) leave the processor's poll interval as the latency.
const TypePush = "PUSH"
TypePush is the row type of a push to clients: the platform's outbox processor sends the payload, with the other PUSH rows', to the push gateway's POST /api/push/batch. The platform (and any other drainer) must know the type before an application writes one: an older drainer posts an unknown type to the bare platform URL and fails the row.
Variables ¶
This section is empty.
Functions ¶
func AuditPayload ¶
func AuditPayload(event usecase.DomainEvent, command any, applicationCode, clientCode string) ([]byte, error)
AuditPayload serialises the audit row of (event, command) as POST /api/audit-logs/batch's item: entityType and entityId from the event subject, operation (AuditOperationNamer, else the command's type name), operationData (the redacted command, or its AuditData; the value itself, left out when nil), principalId, performedAt (the event time, ISO-8601 UTC with 0, 3, 6 or 9 fraction digits), source, correlationId, applicationCode and clientCode (each left out when empty). The platform refuses a row without principalId, and requires applicationCode of an application-scoped credential.
func EventPayload ¶
func EventPayload(event usecase.DomainEvent, clientCode string) ([]byte, error)
EventPayload serialises a domain event as POST /api/events/batch's item: specVersion, type, source (left out when empty: the platform then defaults it to the type's {application}:{subdomain}), subject, data (the event's ToDataJSON value itself), correlationId, causationId, deduplicationId (the event id), messageGroup, clientCode (the event's own when it implements ClientCoded, else clientCode; left out when empty), and contextData (principalId and executionId, each when present). The event time is not sent: the platform stamps it at ingest.
func FormatInstant ¶
FormatInstant formats t in UTC as ISO-8601 with the fraction in groups of three digits, only as many as needed (2026-10-08T10:15:30Z, 2026-10-08T10:15:30.123Z, 2026-10-08T10:15:30.000123Z, ...): Java's Instant.toString, which the other writers use.
func InsertStatement ¶
func InsertStatement(table string, style Placeholder) (string, error)
InsertStatement returns InsertSQL for table (DefaultTable when empty) in the given placeholder style. table must be a plain or schema-qualified identifier.
func PushInsertStatement ¶ added in v0.0.17
PushInsertStatement returns the Postgres INSERT of a PUSH row with the notification: the shared InsertSQL's columns, written in one statement with pg_notify(PushNotifyChannel), so the processor's LISTEN wakes when the writer's transaction commits and not before. Arguments are Row.Args, as for InsertStatement with PlaceholderDollar. table must be a plain or schema-qualified identifier.
Types ¶
type AuditDataProvider ¶
type AuditDataProvider interface {
AuditData() any
}
AuditDataProvider is implemented by a command that chooses what its audit row records as operationData. AuditData's value is serialised and redacted like a command; nil leaves operationData out. Without it the whole command is recorded (redacted).
type AuditOperationNamer ¶
type AuditOperationNamer interface {
AuditOperation() string
}
AuditOperationNamer is implemented by a command whose audit row names the operation explicitly (for example "place-order"). Without it the operation is the command's Go type name.
type ClientCoded ¶
type ClientCoded interface {
ClientCode() string
}
ClientCoded is implemented by a domain event that names the client it belongs to (the client's identifier slug). It overrides the writer's configured client code for that event's rows.
type Config ¶
type Config struct {
// ClientID is stored in every row's client_id column (the outbox
// processor does not read it; it is for the application's own
// partitioning). May be empty.
ClientID string
// ClientCode is the client's identifier slug, written as clientCode in
// event and audit payloads so the platform attributes them to that
// client (a domain event implementing ClientCoded overrides it). Empty
// leaves it out: the platform then scopes by the credential (an anchor
// writes platform-scoped rows, a single-client credential its client).
// Dispatch jobs that name no client of their own default to it, except
// under PlatformClientCode.
ClientCode string
// ApplicationCode is written as each audit row's applicationCode. The
// platform requires it of an application-scoped credential.
ApplicationCode string
// Now is the clock for created_at/updated_at. Default time.Now.
Now func() time.Time
}
Config is what the writers need beyond the row's own data.
type DispatchJob ¶
type DispatchJob struct {
// ID is the job id, [0-9A-Za-z_-]{1,17}; generated (a 13-character TSID)
// when empty. It is in the payload, so a batch the processor sends again
// creates a job with the same id.
ID string
Kind string // EVENT or TASK; default TASK
Code string // {application}:{rest}, at most 255 characters
Source string // default: the code's application segment
Subject string
TargetURL string // absolute http(s) URL with a host and no user info
// Payload is the body delivered to TargetURL, with PayloadContentType.
// Default: {} as application/json.
Payload string
PayloadContentType string
DataOnly bool
ExternalID string
EventID string
CorrelationID string
// PlatformClientID is the platform's client id (never a slug). ClientCode
// is the client's identifier slug, which the platform resolves at ingest
// (an unknown code refuses the whole batch: 400 for an anchor, 403
// otherwise). Set at most one. Without either, DispatchJobRow gives the
// job the writer's Config.ClientCode (none for PlatformClientCode).
PlatformClientID string
ClientCode string
SubscriptionID string
ServiceAccountID string
DispatchPoolID string
MessageGroup string
Mode string // IMMEDIATE, NEXT_ON_ERROR or BLOCK_ON_ERROR
// Sequence, TimeoutSeconds and MaxRetries are at least 1 when set (the
// platform turns 0 into its default, so 0 cannot be sent); nil leaves
// them to the platform (99, 30 and 3).
Sequence *int
TimeoutSeconds *int
MaxRetries *int
Metadata []MetadataEntry
Queue string // DEFAULT or HIGH_PRIORITY
Descriptor string
// ConnectionID is the connection the job is delivered through: pausing it
// holds the job, and (with no subscription) its service account signs the
// delivery. Must be global or the job's client's, shared or the job's
// application's; the platform refuses the whole batch otherwise.
ConnectionID string
// IdempotencyKey makes the create idempotent: the first job with a key,
// within its application (the code's first segment) and client, wins it; a
// repeat answers that job's id and creates nothing. Written exactly as
// given; when set, not blank and at most 100 bytes of UTF-8; kept 7 days.
// An ID is not an idempotency key.
IdempotencyKey string
// contains filtered or unexported fields
}
DispatchJob is a job for POST /api/dispatch-jobs/batch, written to the outbox in the same transaction as the work that asks for it. Build it with NewDispatchJob and the With* methods, then Validate (the writers validate before writing, so a row the platform would refuse terminally is never stored). Zero values mean "not set" and are left out of the payload.
func DispatchJobPayload ¶
func DispatchJobPayload(job DispatchJob) (DispatchJob, []byte, error)
DispatchJobPayload validates the job, applies its defaults and serialises it as POST /api/dispatch-jobs/batch's item. It returns the resolved job (its id is the row id).
func NewDispatchJob ¶
func NewDispatchJob(code, targetURL string) DispatchJob
NewDispatchJob starts a job for code delivered to targetURL.
func (DispatchJob) DataOnlyEnabled ¶
func (j DispatchJob) DataOnlyEnabled() DispatchJob
func (DispatchJob) Resolved ¶
func (j DispatchJob) Resolved() (DispatchJob, error)
Resolved returns the job with its defaults applied (id generated, kind TASK, source from the code, payload {} as application/json), after validating it.
func (DispatchJob) Validate ¶
func (j DispatchJob) Validate() error
Validate applies the batch endpoint's rules (framework write-path 7a).
func (DispatchJob) WithClientCode ¶
func (j DispatchJob) WithClientCode(code string) DispatchJob
func (DispatchJob) WithConnectionID ¶ added in v0.0.14
func (j DispatchJob) WithConnectionID(id string) DispatchJob
func (DispatchJob) WithCorrelationID ¶
func (j DispatchJob) WithCorrelationID(id string) DispatchJob
func (DispatchJob) WithDescriptor ¶
func (j DispatchJob) WithDescriptor(d string) DispatchJob
func (DispatchJob) WithDispatchPoolID ¶
func (j DispatchJob) WithDispatchPoolID(id string) DispatchJob
func (DispatchJob) WithEventID ¶
func (j DispatchJob) WithEventID(id string) DispatchJob
func (DispatchJob) WithExternalID ¶
func (j DispatchJob) WithExternalID(id string) DispatchJob
func (DispatchJob) WithID ¶
func (j DispatchJob) WithID(id string) DispatchJob
func (DispatchJob) WithIdempotencyKey ¶ added in v0.0.14
func (j DispatchJob) WithIdempotencyKey(k string) DispatchJob
func (DispatchJob) WithJSONPayload ¶
func (j DispatchJob) WithJSONPayload(v any) (DispatchJob, error)
WithJSONPayload sets the delivered body to v serialised as JSON (application/json).
func (DispatchJob) WithKind ¶
func (j DispatchJob) WithKind(kind string) DispatchJob
func (DispatchJob) WithMaxRetries ¶
func (j DispatchJob) WithMaxRetries(n int) DispatchJob
func (DispatchJob) WithMessageGroup ¶
func (j DispatchJob) WithMessageGroup(g string) DispatchJob
func (DispatchJob) WithMetadata ¶
func (j DispatchJob) WithMetadata(key, value string) DispatchJob
WithMetadata appends one metadata pair.
func (DispatchJob) WithMode ¶
func (j DispatchJob) WithMode(mode string) DispatchJob
func (DispatchJob) WithPayload ¶
func (j DispatchJob) WithPayload(text, contentType string) DispatchJob
WithPayload sets the delivered body and its content type.
func (DispatchJob) WithPlatformClientID ¶
func (j DispatchJob) WithPlatformClientID(id string) DispatchJob
func (DispatchJob) WithQueue ¶
func (j DispatchJob) WithQueue(q string) DispatchJob
func (DispatchJob) WithSequence ¶
func (j DispatchJob) WithSequence(n int) DispatchJob
func (DispatchJob) WithServiceAccountID ¶
func (j DispatchJob) WithServiceAccountID(id string) DispatchJob
func (DispatchJob) WithSource ¶
func (j DispatchJob) WithSource(source string) DispatchJob
func (DispatchJob) WithSubject ¶
func (j DispatchJob) WithSubject(subject string) DispatchJob
func (DispatchJob) WithSubscriptionID ¶
func (j DispatchJob) WithSubscriptionID(id string) DispatchJob
func (DispatchJob) WithTimeoutSeconds ¶
func (j DispatchJob) WithTimeoutSeconds(n int) DispatchJob
type MetadataEntry ¶
MetadataEntry is one dispatch-job metadata pair; the batch endpoint reads metadata as an array of {"key","value"}, in order.
type Placeholder ¶
type Placeholder int
Placeholder is a driver's bind-parameter style.
const ( // PlaceholderQuestion is ? (MySQL, SQLite). PlaceholderQuestion Placeholder = iota // PlaceholderDollar is $1, $2, ... (Postgres: pgx, lib/pq). PlaceholderDollar )
type Push ¶ added in v0.0.17
type Push struct {
// ID is the row id and the payload's pushId (the platform writes a pushId once); a 13-character
// TSID is generated when empty. At most 26 characters [0-9A-Za-z_-].
ID string
// Topic is {application}:{segment}(:{segment})*, the first segment [a-z][a-z0-9-]* the
// application (which the writer's credential must be allowed to push for), the others
// [0-9A-Za-z_.-]+, at most 255 characters.
Topic string
// Type is [a-z][a-z0-9-]*(.[a-z][a-z0-9-]*)*, at most 100 characters; reset, auth-expired,
// subscribed and fw.* are reserved.
Type string
// Data is the JSON object the listeners receive, as compact JSON.
Data json.RawMessage
// ClientCode is the client the push is for, by identifier. Empty: the writer's
// Config.ClientCode (none for PlatformClientCode, which leaves the credential to decide: an
// anchor pushes for the platform client).
ClientCode string
// MessageGroup is the outbox message group. Empty: the topic, so the pushes of one topic
// reach the gateway in order (the processor sends a group's rows one at a time, in order).
// With Ungrouped it is ignored.
MessageGroup string
// Ungrouped writes the row with no message group: the processor batches such rows into
// concurrent calls, which is faster and keeps no order. For signal-only topics.
Ungrouped bool
// MaxDataBytes is the data limit (default DefaultPushMaxDataBytes, at most MaxPushDataBytes);
// set it to the application's push.max-data-bytes.
MaxDataBytes int
// contains filtered or unexported fields
}
Push is a push to the clients of an application, written to the outbox in the same transaction as the change it announces. Build it with NewPush and the With* methods, then PushRow (the writers validate before writing, so a push the platform would refuse terminally is never stored). Pushes are signals: the clients read the change again; durable work is events and dispatch jobs.
func NewPush ¶ added in v0.0.17
NewPush starts a push of type to topic carrying data: a struct or map marshalled to JSON, or JSON text (json.RawMessage, []byte). A value that is not a JSON object is refused by Validate.
func PushPayload ¶ added in v0.0.17
PushPayload validates the push, applies its defaults (a generated id) and serialises it as POST /api/push/batch's item. It returns the resolved push (its id is the row id and the pushId).
func (Push) Validate ¶ added in v0.0.17
Validate applies framework section 1 (the topic, type and data rules and their codes) and the row's own: the id and the client code. The error is a *PushError.
func (Push) WithClientCode ¶ added in v0.0.17
func (Push) WithMaxDataBytes ¶ added in v0.0.17
func (Push) WithMessageGroup ¶ added in v0.0.17
func (Push) WithUngrouped ¶ added in v0.0.17
type PushError ¶ added in v0.0.17
type PushError struct{ Code, Message string }
PushError is a push the writer refuses; Code is one of the Push* codes above.
type Row ¶
type Row struct {
ID string
Type string
MessageGroup string
Payload []byte
ClientID string
CreatedAt time.Time
}
Row is one outbox_messages row as the writers store it: status 0 (pending), created_at = updated_at = the write time in UTC, headers NULL. An empty MessageGroup or ClientID is stored as NULL.
func AuditRow ¶
AuditRow builds the AUDIT_LOG row of (event, command): a generated id (13-character TSID), no message group.
func DispatchJobRow ¶
func DispatchJobRow(cfg Config, job DispatchJob) (Row, error)
DispatchJobRow builds the DISPATCH_JOB row of a job: the row id is the job id, message_group the job's message group.
func EventRow ¶
func EventRow(cfg Config, event usecase.DomainEvent) (Row, error)
EventRow builds the EVENT row of a domain event: the row id is the event id (also the payload's deduplicationId), message_group the event's message group.
func PushRow ¶ added in v0.0.17
PushRow builds the PUSH row of a push: the row id is the push id, message_group the topic unless the push names another group or is Ungrouped. The client rule is the dispatch job's: the push's own clientCode, else the writer's configured client except PlatformClientCode.