Documentation
¶
Overview ¶
Package persistence is the port repositories and units of work are written against: load and store aggregates by identity, and make the state a handler wrote and the events its aggregates raised commit together or not at all.
No driver appears here and no SQL is generated here — the deliberate omission of the whole framework is an ORM. warren/persistence/postgres and its siblings implement these interfaces; this package holds the contract, the enlistment seam that makes atomic event publication possible, and an in-memory driver so the contract is exercised by CI without a database.
Index ¶
- func Collect(ctx context.Context) (context.Context, func() []domain.Event)
- func ErrNestedOptions() error
- func ErrNoTransaction(op string) error
- func ForApp(uow UnitOfWork, opts ...Option) app.UnitOfWork
- func InTransaction(ctx context.Context) bool
- func RunContract[T domain.Root[K], K domain.ID](t *testing.T, newDriver NewDriver[T, K], newAggregate func(K) T, ids ...K)
- func RunVersionedContract[T domain.Root[K], K domain.ID](t *testing.T, newDriver NewDriver[T, K], newAggregate func(K) T, ids ...K)
- func Track(ctx context.Context, aggregates ...domain.Aggregate)
- func Write(ctx context.Context, op string, root domain.Aggregate, ...) error
- type Level
- type MemoryRepository
- type MemoryUnitOfWork
- type NewDriver
- type Option
- type Repository
- type Transaction
- type UnitOfWork
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Collect ¶
Collect returns a context that gathers the aggregates enlisted beneath it and the drain that pulls their events, in enlistment order, exactly once. A UnitOfWork calls it when it opens a transaction and calls the drain between the function returning nil and the commit.
It is nesting-aware: called with a collector already on ctx it returns that same ctx and a drain returning nil, so only the outermost Do drains.
func ErrNestedOptions ¶ added in v0.2.0
func ErrNestedOptions() error
ErrNestedOptions is the error a driver returns when a nested Do is passed an Option. It is exported because every driver must return the SAME error for it — a driver in another module that re-implements the message drifts from this one, and the two then disagree in a way only a user notices.
The refusal is not fastidiousness. On a SQL database isolation and read-only are properties of a transaction's first statement, so a nested Do cannot honour an Option even in principle; silently ignoring one would mean a handler asking for Serializable and quietly getting Read Committed.
func ErrNoTransaction ¶ added in v0.2.0
ErrNoTransaction is the diagnostic a write outside a unit of work produces. It is exported for the same reason ErrNestedOptions is: every driver must produce the SAME message, and a second copy in another module drifts from this one within a month.
func ForApp ¶
func ForApp(uow UnitOfWork, opts ...Option) app.UnitOfWork
ForApp adapts a UnitOfWork to app.UnitOfWork, the one-method seam app.Transactional takes. The two differ only in Do's variadic options, which app deliberately does not know about — this adapter is the seam, so no application has to write it.
warren.Providers(persistence.ForApp) // *MemoryUnitOfWork → app.UnitOfWork
func InTransaction ¶
InTransaction reports whether a unit of work is in scope on ctx.
func RunContract ¶
func RunContract[T domain.Root[K], K domain.ID](t *testing.T, newDriver NewDriver[T, K], newAggregate func(K) T, ids ...K)
RunContract is the suite every persistence driver must pass — the memory driver here, Postgres and Mongo behind a build tag. It asserts only what every driver can promise: identity round-trips, NOT_FOUND is a code and not a nil, a rolled-back transaction leaves nothing behind, and Save enlists the aggregate so its events reach the outbox.
newAggregate builds a fresh aggregate with the given identity and at least one pending event. The suite calls it with DISTINCT, non-zero identities — a suite that only ever used the zero value could not catch a driver storing everything under one key, and would build the very zero-identity aggregate domain.NewAggregateRoot exists to prevent.
func RunVersionedContract ¶ added in v0.2.0
func RunVersionedContract[T domain.Root[K], K domain.ID](t *testing.T, newDriver NewDriver[T, K], newAggregate func(K) T, ids ...K)
RunVersionedContract certifies a driver's optimistic concurrency, and is run IN ADDITION to RunContract by drivers whose aggregates embed domain.VersionedRoot. It is separate because the version is opt-in: a driver that stores unversioned aggregates is not failing anything by not implementing it.
The suite exists because the framework shipped without it. warren.md §3.3 promised CodeConflict, Repository.Save had no expected-version seam, and a field test paying one invoice four times concurrently got four 201s and four published events. Every subtest here is that defect in miniature.
newAggregate must return an aggregate implementing domain.Versioned at version 0.
ids must hold at least versionedContractIDs DISTINCT, non-zero identities: one per subtest. Unlike RunContract, this suite cannot reuse an identity across subtests, because a store that newDriver does not actually reset — a shared Postgres table, say — would leave the row behind and every later insert would conflict for the wrong reason. Its own diagnostic would then be indistinguishable from the defect it is testing for.
func Track ¶
Track enlists aggregates with the unit of work in scope on ctx, so their events are drained into the outbox when it commits. Repositories call it from Save.
Outside any Do it is a no-op — and a no-op loses nothing: PullEvents is never called, so the events stay pending on the aggregate and a later Do publishes them.
func Write ¶ added in v0.2.0
func Write(ctx context.Context, op string, root domain.Aggregate, write func(context.Context) error) error
Write performs one aggregate write inside the unit of work in scope, and enlists root when — and only when — the write succeeded.
It is the repository's write contract in a single call, and there is no way to perform half of it. Outside a unit of work it refuses, because the row would autocommit while the aggregate's events stayed pending on an object about to go out of scope. On success it Tracks, so those events reach the outbox in the same commit. A write that returns an error enlists nothing.
This exists because the enlistment used to be a separate statement AFTER the write, and a separate statement is a statement that can be deleted. A field test deleted it: the row committed, the request returned 201, and the event evaporated with no error, no outbox row and no log line — the exact loss the outbox subsystem exists to prevent.
Be precise about what that buys, because the earlier wording ("Write is rule 1 and rule 3 fused, so there is no longer a line to delete") claimed more than is true. What Write guarantees is that it CANNOT BE HALF PERFORMED: every call checks the transaction before the write and enlists after it, and no call site can keep one half and drop the other. What it cannot do is make itself be called. A repository whose Save runs the SQL directly compiles, passes `go vet`, serves traffic, and loses every event — deleting the whole call is as easy as deleting the old Track line was.
The thing that catches THAT is the contract suite: persistence.RunContract asserts a driver's Save enlists, and `warren g repository` now generates a test that runs it, so a repository that drops Write goes red instead of green. A lint rule is deferred on capability, not on precision — `warren lint arch` is import-graph only by design.
func (r *UserRepository) Save(ctx context.Context, u *domain.User) error {
return persistence.Write(ctx, "user.Save", u, func(ctx context.Context) error {
_, err := r.db(ctx).Exec(ctx, `INSERT INTO users …`, u.ID(), u.Email)
return err
})
}
One aggregate, because one repository owns one. A write that legitimately touches more calls Track for the others inside the write function.
The check it makes is "is there a unit of work", which is precisely the precondition for Track to do anything. A DRIVER may hold a stricter one — postgres.RequireTx also insists on an ambient Postgres transaction to run SQL on — and keeps it; the two answer different questions and both are worth asking.
Types ¶
type Level ¶
type Level string
Level is a transaction isolation level. The empty level means the driver's default.
const ( // ReadCommitted is the common default. ReadCommitted Level = "read committed" // RepeatableRead prevents non-repeatable reads. RepeatableRead Level = "repeatable read" // Serializable is the level a cross-aggregate invariant needs. Its // serialization failures surface as CodeUnavailable, which is what makes // app.Retrying re-run the whole transaction. Serializable Level = "serializable" )
type MemoryRepository ¶
type MemoryRepository[T domain.Root[K], K domain.ID] struct { // contains filtered or unexported fields }
MemoryRepository is the in-process Repository for one aggregate type. Its Save calls Track, as the port's contract requires.
func NewMemoryRepository ¶
func NewMemoryRepository[T domain.Root[K], K domain.ID](uow *MemoryUnitOfWork) *MemoryRepository[T, K]
NewMemoryRepository returns an in-process Repository backed by uow.
func (*MemoryRepository[T, K]) Delete ¶
func (r *MemoryRepository[T, K]) Delete(ctx context.Context, root T) error
Delete removes the aggregate, or returns CodeNotFound.
func (*MemoryRepository[T, K]) FindByID ¶
func (r *MemoryRepository[T, K]) FindByID(ctx context.Context, id K) (T, error)
FindByID returns a COPY of the aggregate, reading the ambient transaction's pending writes first — a handler sees what it just saved, and never the object another transaction is mutating.
type MemoryUnitOfWork ¶
type MemoryUnitOfWork struct {
// contains filtered or unexported fields
}
MemoryUnitOfWork is the in-process driver: real staging, real rollback, real event draining, no database. It exists so the port is exercised by CI, so app.Transactional has something to be tested against, and so a test suite need not reach for Docker — the same reasons broker/memory and inbox.NewMemoryStore live in core.
It hands every reader its own copy of an aggregate, as a real driver does by materialising rows: two transactions cannot see each other's uncommitted mutations, a rolled-back transaction leaves the committed aggregate untouched, and a retried handler re-reads clean state instead of re-applying its change to the object it already mutated.
func NewMemoryUnitOfWork ¶
func NewMemoryUnitOfWork() *MemoryUnitOfWork
NewMemoryUnitOfWork returns an in-process UnitOfWork.
func (*MemoryUnitOfWork) Do ¶
func (u *MemoryUnitOfWork) Do(ctx context.Context, fn func(context.Context) error, opts ...Option) error
Do runs fn in a transaction. A nested call joins: it runs fn on the ambient transaction and returns its error, opening and committing nothing.
type NewDriver ¶
type NewDriver[T domain.Root[K], K domain.ID] func(t *testing.T) (UnitOfWork, Repository[T, K])
NewDriver builds a fresh unit of work and repository for one subtest.
type Option ¶
type Option func(*Transaction)
Option configures one Do.
type Repository ¶
type Repository[T domain.Root[K], K domain.ID] interface { // FindByID returns the aggregate, or an error carrying CodeNotFound. FindByID(ctx context.Context, id K) (T, error) // Save persists the whole aggregate, inserting or updating by identity, // and enlists it with the unit of work in scope by calling Track. // // The Track call is part of this contract, not an implementation detail: // a driver whose Save does not Track loses the aggregate's events, and // the contract suite asserts it. // // If root implements domain.Versioned, the write is CONDITIONAL on the // version the aggregate was loaded at. A write whose version the store has // moved past returns CodeContention and changes nothing — it is the only // thing standing between two concurrent requests and a lost update, and // RunVersionedContract certifies the code, not merely that it errored. An // aggregate that does not implement domain.Versioned is written // unconditionally, as before. // // CONTENTION and not CONFLICT, and the difference is not cosmetic: this // port doc said CodeConflict while errors.go, the postgres driver, the // memory driver and warren.md §3.3 all said CONTENTION — in the first // place a driver author reads. Nothing was written, so the next attempt // re-reads and usually wins; app.Retrying covers CONTENTION and does not // cover CONFLICT, and on a CONSUMER the CONFLICT column ACKS, which // destroys the message with its work not done. An INSERT that collides // with an existing row is the other answer — that one really is // CodeConflict, because a retry finds the row still there for ever. Save(ctx context.Context, root T) error // Delete removes the aggregate, or returns CodeNotFound. Deleting what // is not there is an error rather than a silent success: an idempotent // replay acks on NOT_FOUND anyway (§2.6), and succeeding silently hides // bugs. // // It takes the ROOT, not an identifier, and enlists it exactly as Save // does. Removing an aggregate is precisely when OrderCancelled or // AccountClosed is raised, and those events live on the caller's // instance. Taking an id cannot be repaired by loading the aggregate // inside Delete: the reloaded object is a DIFFERENT object with zero // pending events, so enlisting it publishes nothing. The contract suite // asserts the enlistment. Delete(ctx context.Context, root T) error }
Repository loads, stores, and removes a single aggregate type by identity. It is the floor every driver implements; a project's own repository interface lives in its domain package and is usually wider.
type Transaction ¶
Transaction is the resolved configuration of one Do — what a driver reads to begin the transaction it was asked for. It names no driver type.
func Configure ¶
func Configure(opts ...Option) Transaction
Configure applies opts — the call a driver makes at the top of Do.
type UnitOfWork ¶
type UnitOfWork interface {
Do(ctx context.Context, fn func(context.Context) error, opts ...Option) error
}
UnitOfWork runs a function inside one transaction and makes the state that function wrote and the events its aggregates raised commit together.
A nested Do — a handler calling Do while app.Transactional already opened one — JOINS the transaction in scope: it opens nothing, commits nothing, and drains nothing. Both patterns appear in warren.md §10 and §3.2, so erroring would break documented code, and savepoints would make the port's semantics driver-dependent (Mongo and Redis have none) while publishing events for state that was rolled back.
Source Files
Directories