usecasepgx

type CommitOption ¶

type CommitOption func(*commitOptions)

CommitOption adds to what a commit writes besides the aggregate change, its event and its audit row.

func WithDispatchJobs(jobs ...outbox.DispatchJob) CommitOption

type DbTx ¶

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

DbTx is an opaque write handle passed to repository Persist methods. Wraps a pgx.Tx so repository code doesn't import pgx directly through some unrelated path; a future driver swap touches this file plus the commit.go file, nothing else.

Repositories access the underlying pgx.Tx via Inner().

func WrapTxForBootstrap(tx pgx.Tx) *DbTx

func (t *DbTx) Inner() pgx.Tx

type DispatchJobSink ¶

type DispatchJobSink interface {
	Sink
	// WriteDispatchJobs writes the jobs, in order, in the open transaction.
	// The jobs are already validated and resolved (each has its id). Must
	// not commit or roll back the tx.
	WriteDispatchJobs(ctx context.Context, tx *DbTx, jobs []outbox.DispatchJob) error
}

DispatchJobSink is a Sink that also writes dispatch jobs (outbox rows of type DISPATCH_JOB, which the platform's outbox processor delivers to POST /api/dispatch-jobs/batch). outboxpgx.Sink implements it. A commit that carries dispatch jobs through a Sink without it fails before anything is written (INTERNAL DISPATCH_JOB_UNSUPPORTED).

type Persist ¶

type Persist[A usecase.HasID] interface {
	Persist(ctx context.Context, agg *A, tx *DbTx) error
	Delete(ctx context.Context, agg *A, tx *DbTx) error
}

Persist is implemented by a repository to upsert and delete aggregates of type A within a transaction. Implement on the *repository*, not the aggregate — the CLAUDE.md "aggregates don't persist themselves" rule applies in Go too.

Example:

type OrderRepository struct{ pool *pgxpool.Pool }

func (r *OrderRepository) Persist(ctx context.Context, o *Order, tx *usecasepgx.DbTx) error {
    _, err := tx.Inner().Exec(ctx,
        `INSERT INTO orders (id, customer_id, total) VALUES ($1, $2, $3)
         ON CONFLICT (id) DO UPDATE SET customer_id = $2, total = $3`,
        o.ID, o.CustomerID, o.Total)
    return err
}

func (r *OrderRepository) Delete(ctx context.Context, o *Order, tx *usecasepgx.DbTx) error {
    _, err := tx.Inner().Exec(ctx, `DELETE FROM orders WHERE id = $1`, o.ID)
    return err
}

type Sink ¶

type Sink interface {
	// WriteEvent appends the domain event to its destination, using the
	// already-open transaction. Must not commit or roll back the tx.
	WriteEvent(ctx context.Context, tx *DbTx, event usecase.DomainEvent) error

	// WriteAudit appends an audit log row for (event, command), using
	// the same open transaction. Implementations may no-op when audit
	// logging is disabled.
	WriteAudit(ctx context.Context, tx *DbTx, event usecase.DomainEvent, command any) error
}

Sink is the destination for domain events and audit logs. The UnitOfWork doesn't know whether events go to outbox_messages (consumer apps via outboxpgx.Sink) or directly to msg_events / aud_logs (platform code via a platform-specific sink); it just calls into this interface inside the transaction it has open.

A Sink that also implements DispatchJobSink can write dispatch jobs (WithDispatchJobs / usecaseop's Plan.Dispatch).

Implementations:

• outboxpgx.Sink — consumer apps; writes to outbox_messages so fc-outbox-processor forwards events to the platform API. Also a DispatchJobSink.
• Platform's own sink (in flowcatalyst-go) — writes directly to msg_events and aud_logs. Not part of this SDK.

type SyncDeleteItem ¶

type SyncDeleteItem[A usecase.HasID] struct {
	Aggregate *A
	Event     usecase.DomainEvent
}

SyncDeleteItem pairs an aggregate to delete with the per-row domain event to emit for it. See CommitSync.

type SyncSaveItem ¶

type SyncSaveItem[A usecase.HasID] struct {
	Aggregate *A
	Event     usecase.DomainEvent
}

SyncSaveItem pairs an aggregate to upsert with the per-row domain event to emit for it (typically a Created or Updated event). See CommitSync.

type TxScopedUnitOfWork ¶

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

TxScopedUnitOfWork is a UnitOfWork bound to an externally-orchestrated transaction opened by Run. Every commit performed against it joins that single transaction. The scoped UoW does NOT close the tx; the surrounding Run does (commits on Success, rolls back on Failure).

Use TxScopedUnitOfWork.WithTx for ad-hoc writes that must be atomic with the events the scoped UoW produces (e.g. updating a non-aggregate row, raw SQL on a join table).

func (s *TxScopedUnitOfWork) Sink() Sink

func (s *TxScopedUnitOfWork) WithTx(ctx context.Context, fn func(pgx.Tx) error) error

type UnitOfWork ¶

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

UnitOfWork is the pgx-backed UnitOfWork. Each top-level Commit / CommitDelete / CommitAll / EmitEvent opens its own transaction and commits or rolls back on exit. For orchestrated transactions that span multiple use cases (or mix use cases with ad-hoc writes), use Run.

Construct with New, then pass the *UnitOfWork into every use case that needs to commit. The Sink wired here decides where domain events and audit logs land.

func New(pool *pgxpool.Pool, sink Sink) *UnitOfWork

func (u *UnitOfWork) Pool() *pgxpool.Pool

func (u *UnitOfWork) Sink() Sink