outbox

package
v0.0.11 Latest Latest
Warning

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

Go to latest
Published: Nov 1, 2026 License: Apache-2.0 Imports: 13 Imported by: 0

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

View Source
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.

View Source
const (
	TypeEvent       = "EVENT"
	TypeAuditLog    = "AUDIT_LOG"
	TypeDispatchJob = "DISPATCH_JOB"
)

Row types: the outbox processor maps the type column to an endpoint by exact string.

View Source
const DefaultTable = "outbox_messages"

DefaultTable is the outbox table every SDK writes.

View Source
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.

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

func FormatInstant(t time.Time) string

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.

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).
	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 the job has no client.
	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
	// 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) 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) 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

type MetadataEntry struct {
	Key   string `json:"key"`
	Value string `json:"value"`
}

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 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

func AuditRow(cfg Config, event usecase.DomainEvent, command any) (Row, error)

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 (Row) Args

func (r Row) Args() []any

Args returns the INSERT arguments in InsertSQL's column order.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL