outboxpgx

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: 7 Imported by: 0

Documentation

Overview

Package outboxpgx implements the consumer-facing Sink for use cases running against pgx. It writes domain events, audit logs and dispatch jobs to the outbox_messages table; the fc-outbox-processor polls that table and forwards items to the FlowCatalyst platform API.

The Sink satisfies usecasepgx.Sink, so any use case that uses usecasepgx.Commit / CommitDelete / CommitAll / EmitEvent will emit events through the outbox transparently.

Index

Constants

View Source
const CreateOutboxTableSQL = `` /* 817-byte string literal not displayed */

CreateOutboxTableSQL is the schema for the outbox_messages table. Matches the columns expected by the fc-outbox-processor (also used by the TypeScript and Laravel SDKs). Safe to run multiple times.

Variables

This section is empty.

Functions

func InitSchema

func InitSchema(ctx context.Context, pool *pgxpool.Pool) error

InitSchema creates the outbox_messages table and its indexes on the given pool. Idempotent — safe to call on every boot. In production, run your real migration system instead and skip this helper.

Types

type Config

type Config struct {
	// TableName is the outbox table. Default: "outbox_messages".
	TableName string
	// ClientID is stored in every row's client_id column. May be empty. The
	// outbox processor does not read it.
	ClientID string
	// ClientCode is the client's identifier slug, sent as clientCode in
	// event and audit payloads so the platform attributes them to that
	// client. The platform refuses a code the credential cannot reach (403
	// for the whole batch); empty leaves it out, and the platform then scopes
	// by the credential (an anchor writes platform-scoped events, a
	// single-client credential its own client). A domain event implementing
	// outbox.ClientCoded overrides it.
	ClientCode string
	// AuditEnabled writes audit log rows alongside every event. The
	// platform always audits its control plane writes; consumer apps
	// should only enable this for admin / human-initiated operations,
	// not for every transactional event.
	AuditEnabled bool
	// ApplicationCode, when set, is written as each audit row's
	// applicationCode, attributing it to the application. The platform
	// requires it when the credential the outbox processor uses is
	// application-scoped (an application's own service account), and
	// refuses the row otherwise.
	ApplicationCode string
	// Now is the clock for created_at/updated_at. Default time.Now.
	Now func() time.Time
}

Config configures the outbox Sink.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults.

type Sink

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

Sink writes domain events, audit logs and dispatch jobs to the outbox_messages table. Satisfies usecasepgx.Sink. The rows and their payloads are built by package outbox (the shared outbox contract).

func NewSink

func NewSink(cfg Config) *Sink

NewSink builds a Sink. If cfg.TableName is empty, defaults to "outbox_messages".

func (*Sink) WriteAudit

func (s *Sink) WriteAudit(ctx context.Context, tx *usecasepgx.DbTx, event usecase.DomainEvent, command any) error

WriteAudit inserts an AUDIT_LOG row into outbox_messages, if audit logging is enabled. No-op otherwise.

func (*Sink) WriteDispatchJob

func (s *Sink) WriteDispatchJob(ctx context.Context, tx pgx.Tx, job outbox.DispatchJob) error

WriteDispatchJob validates job and inserts its DISPATCH_JOB row (the platform's outbox processor delivers it to POST /api/dispatch-jobs/batch). The row id is the job id (generated when the job has none).

This is the lower-level path, for a transaction the application runs itself. Inside a use case, prefer declaring the job on the plan (usecaseop's Plan.Dispatch): the plan writes it in the use case's transaction in the contract's order (after the event, before the audit row), refuses an invalid job before anything is written, and gives a job without a correlation id the event's.

func (*Sink) WriteDispatchJobs

func (s *Sink) WriteDispatchJobs(ctx context.Context, tx *usecasepgx.DbTx, jobs []outbox.DispatchJob) error

WriteDispatchJobs inserts the DISPATCH_JOB rows of jobs, in order. It is what usecaseop's Plan.Dispatch (and usecasepgx.WithDispatchJobs) write through, after the event and before the audit row. Satisfies usecasepgx.DispatchJobSink.

func (*Sink) WriteEvent

func (s *Sink) WriteEvent(ctx context.Context, tx *usecasepgx.DbTx, event usecase.DomainEvent) error

WriteEvent inserts an EVENT row into outbox_messages. The row id is the event id (also the payload's deduplicationId).

Jump to

Keyboard shortcuts

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