Why aggregate versioning

The problemTwo calls read the same invoice, each applied its own rule, both saved. What stayed in the database is whatever was written second — the first change is gone, and neither transaction noticed: both `UPDATE`s succeeded.

The same service as in the DDD chapters: the invoice, its modules, ports and adapters. The domain barely changes here — the invoice gains one field, and with it an answer to the question of who wins when two calls arrive at once.

Step 1 of 8

Two calls, one invoice

On the right is the payment use case as the infrastructure chapter left it: a unit of work, a repository, published facts.

While it is called one at a time, it is correct. Now two calls at once: one takes a payment, the other charges a late fee — the overdue scheduler woke up a second earlier.

Both loaded the invoice in the same state. Both applied their rule to their own copy in memory. Both saved. What stayed in the database is whatever was written second: the late fee wiped the payment, or the payment wiped the late fee — as luck would have it.

This is a lost update, and a transaction does not cure it. The transactions did not get in each other's way: each did exactly what it was asked to do, and both committed. Nothing failed, the logs are clean.

Step 2 of 8

Why not lock the row

The first idea is to add FOR UPDATE to the SELECT and make the second call wait for the first. That is pessimistic locking, and it does work.

Here is what it costs. The row stays locked until the transaction ends: while the use case computes a late fee or calls the payment gateway, everyone else on that invoice waits in line. Two use cases that take rows in a different order give you a deadlock.

And the main thing: you can only lock what happens inside one transaction. A person opened the invoice, changed the due date and pressed save a minute later — you cannot hold a transaction for a minute, and at the moment of their edit there is nothing to lock.

So the answer is not to forbid the state from changing, but to notice that it did.

Step 3 of 8

The invoice gets a version

A version is the number of the state that was read. When we write, we tell the database which number we saw; if the number there is different, someone got in between our read and our write.

The field belongs to the aggregate rather than to a table on its own: the aggregate is the consistency boundary, so it has one counter for everything inside it. Change a line of the invoice, and the invoice's version goes up.

The version adds no domain rule: Issue creates the invoice with zero, Version() only returns it, and there is no SetVersion — the number is raised by whoever writes. load.go changed too: the invoice comes back from the database with the number that was there.

Step 4 of 8

The conflict is part of the port's contract

ErrConflict is declared next to the port, and that matters: the use case has to tell "it did not save because the invoice was changed" from "the database is down". The first is an ordinary turn of events, the second is an outage.

The domain learned nothing about SQL or RowsAffected along the way. It said exactly the part that belongs to the domain: the invoice may have changed since it was read, and then the write does not happen.

Checking the version by hand — comparing it in the use case before Save — buys nothing: the same gap of time passes between the comparison and the write. The check only has force where the write happens.

Step 5 of 8

The check inside the write

UPDATE invoices SET ..., version = version + 1 WHERE number = $1 AND version = $2 — that is all of optimistic locking. The check and the write are one statement, so nothing can slip in between them.

The interesting line is the next one: RowsAffected() == 0. Zero updated rows means there is no invoice with that version. For an invoice we have just read, that means one thing: the version is already different — so ErrConflict.

Version zero means the invoice is new and is inserted. A second creation with the same number is caught by the unique index; no version is needed for that.

The read now pulls the version from the same row as the rest of the fields: it is part of the state that was read, not a separate query.

Step 6 of 8

A retry instead of an error

A conflict says: the data you decided on is stale. So the right answer is to decide again on fresh data, not to hand an error back to the caller.

That is why the whole attempt, the read included, is repeated. Repeating only the write would write the very same stale data we were trying to leave behind.

The retry is safe precisely because the whole attempt sits inside one unit of work. If the save failed, what went into the outbox rolled back too, so subscribers never heard about the cancelled attempt. Without that, every attempt would send the fact again and idempotency would become someone else's problem.

Three attempts, not an endless loop: a conflict arriving for the third time is no longer a race between two calls but load — and load is answered differently.

Step 7 of 8

When a retry is the wrong answer

A retry fits when the code decides: MarkPaid will look at the fresh invoice again and itself say that a paid invoice is not paid twice.

It does not fit when a person decided, looking at the old state. Then the version travels out with the data — a field in the projection or an ETag header — and comes back with the request to change it. If it does not match, the person is told "the invoice has changed, take another look". That is the right answer, not an error.

The difference is not in the mechanics but in who decided. The mechanics are the same: the number of the state that was read, checked at the moment of the write.

And the boundary: a version detects a conflict on one aggregate. A rule that ties two of them together cannot be held this way — there it is either one aggregate, or consistency stretched over time.

billing/GLOSSARY.md

Glossary

One meaning per word inside this context. The code, the events, the API and the person on the call all spell it the same way.

“Payment” in three departments

Department	What it calls a payment	What it is called here
Product	The purchase went through, the customer saw the checkmark	An order event. Not here
Processing	The money is held, nothing was captured	Authorization
Finance	The money showed up in a statement two days later	Settlement

Words

Authorization. The issuer agreed to hold the amount. No money moved, and the hold expires on its own.

Capture. The instruction to take what was held. From here on the money has left the payer’s account.

Settlement. The money reached our account, and the provider’s statement says so. It arrives days later, and it may not arrive at all.

Payment. One attempt to collect money for one order, with a lifecycle of its own: authorized, captured, refunded. The word means this and nothing else.

Words that are not here

Purchase. A product event: the customer saw the checkmark. It lives with orders, and there is nothing for it to mean here.

billing/invoice/domains/invoice/invoice.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/events"
	"billing/invoice/domains/invoice/vo"
)

// the invoice: the aggregate root
type Invoice struct {
	number   Number
	customer CustomerID
	lines    []Line
	total    vo.Money
	dueOn    time.Time
	status   Status
	facts    []events.Event
}

// creates the invoice and sums the lines
func Issue(
	number Number,
	customer CustomerID,
	lines []Line,
	dueOn time.Time,
) (*Invoice, error) {
	if len(lines) == 0 {
		return nil, ErrNoLines
	}

	total := vo.NewMoney(0)
	for _, line := range lines {
		total = total.Add(line.Amount())
	}

	invoice := &Invoice{
		number:   number,
		customer: customer,
		// a copy of the lines: they cannot be changed from outside
		lines:    append([]Line(nil), lines...),
		total:    total,
		dueOn:    dueOn,
		status:   StatusIssued,
	}

	// records the fact; sending it is not the aggregate's job
	invoice.facts = append(invoice.facts, events.Issued{
		Number:   string(number),
		Customer: string(customer),
		Total:    total,
	})

	return invoice, nil
}

// hands the collected facts to the outside
func (i *Invoice) Events() []events.Event { return i.facts }

func (i *Invoice) Total() vo.Money  { return i.total }
func (i *Invoice) DueOn() time.Time { return i.dueOn }
func (i *Invoice) Status() Status   { return i.status }

billing/invoice/domains/invoice/invoice.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/events"
	"billing/invoice/domains/invoice/vo"
)

// the invoice: the aggregate root
type Invoice struct {
	number   Number
	customer CustomerID
	lines    []Line
	total    vo.Money
	dueOn    time.Time
	status   Status
	// the number of the state that was read from the database
	version  int
	facts    []events.Event
}

// creates the invoice; the version stays zero — it is not in the database yet
func Issue(
	number Number,
	customer CustomerID,
	lines []Line,
	dueOn time.Time,
) (*Invoice, error) {
	if len(lines) == 0 {
		return nil, ErrNoLines
	}

	total := vo.NewMoney(0)
	for _, line := range lines {
		total = total.Add(line.Amount())
	}

	invoice := &Invoice{
		number:   number,
		customer: customer,
		// a copy of the lines: they cannot be changed from outside
		lines:    append([]Line(nil), lines...),
		total:    total,
		dueOn:    dueOn,
		status:   StatusIssued,
	}

	// records the fact; sending it is not the aggregate's job
	invoice.facts = append(invoice.facts, events.Issued{
		Number:   string(number),
		Customer: string(customer),
		Total:    total,
	})

	return invoice, nil
}

// hands the collected facts to the outside
func (i *Invoice) Events() []events.Event { return i.facts }

func (i *Invoice) Total() vo.Money  { return i.total }
func (i *Invoice) DueOn() time.Time { return i.dueOn }
func (i *Invoice) Status() Status   { return i.status }

// returns the version; nothing outside can change it
func (i *Invoice) Version() int { return i.version }

billing/invoice/domains/invoice/paid.go

package invoice

import "billing/invoice/domains/invoice/events"

// marks the invoice paid; paying twice is not allowed
func (i *Invoice) MarkPaid() error {
	if i.status == StatusPaid {
		return ErrAlreadyPaid
	}

	i.status = StatusPaid

	// records the fact; sending it is not the aggregate's job
	i.facts = append(i.facts, events.Paid{Number: string(i.number)})

	return nil
}

billing/invoice/domains/invoice/charged.go

package invoice

import (
	"billing/invoice/domains/invoice/events"
	"billing/invoice/domains/invoice/vo"
)

// puts the fee on the invoice; a paid invoice takes none
func (i *Invoice) ChargePenalty(amount vo.Money) error {
	if i.status == StatusPaid {
		return ErrAlreadyPaid
	}

	if amount.Equal(vo.NewMoney(0)) {
		return nil
	}

	i.lines = append(i.lines, Line{Quantity: 1, Price: amount})
	i.total = i.total.Add(amount)

	// records the fact; sending it is not the aggregate's job
	i.facts = append(i.facts, events.PenaltyCharged{
		Number: string(i.number),
		Amount: amount,
	})

	return nil
}

billing/invoice/domains/invoice/events/event.go

package events

import "billing/invoice/domains/invoice/vo"

// a domain fact; its name travels to the bus as the message type
type Event interface {
	FactName() string
}

// the invoice was issued
type Issued struct {
	Number   string
	Customer string
	Total    vo.Money
}

func (Issued) FactName() string { return "invoice.issued" }

// the invoice was paid
type Paid struct {
	Number string
}

func (Paid) FactName() string { return "invoice.paid" }

// the invoice was charged a late fee
type PenaltyCharged struct {
	Number string
	Amount vo.Money
}

func (PenaltyCharged) FactName() string { return "invoice.penalty_charged" }

billing/invoice/domains/invoice/calendar.go

package invoice

import "time"

// the working calendar: part of the model, not a storage port
type Calendar interface {
	Workdays(from time.Time, to time.Time) int
}

billing/invoice/domains/invoice/services/penalty/penalty.go

package penalty

import (
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the rule: an invoice, a rate and a calendar — an amount out
func Amount(
	overdue *invoice.Invoice,
	now time.Time,
	perWorkday vo.Money,
	calendar invoice.Calendar,
) vo.Money {
	// the rule takes the due date from the invoice itself, not from the caller
	late := calendar.Workdays(overdue.DueOn(), now)
	if late <= 0 {
		return vo.NewMoney(0)
	}

	return vo.NewMoney(perWorkday.Get() * int64(late))
}

billing/invoice/domains/invoice/repository.go

package invoice

import "context"

// port: how the invoice gets into the database and comes back
type Repository interface {
	ByNumber(ctx context.Context, number Number) (*Invoice, error)
	Save(ctx context.Context, invoice *Invoice) error
}

billing/invoice/domains/invoice/repository.go

package invoice

import (
	"context"
	"errors"
)

// the invoice changed between the read and the write
var ErrConflict = errors.New("invoice: invoice changed since it was loaded")

// port: how the invoice gets into the database and comes back
type Repository interface {
	ByNumber(ctx context.Context, number Number) (*Invoice, error)
	// returns ErrConflict if the invoice was changed in the meantime
	Save(ctx context.Context, invoice *Invoice) error
}

billing/invoice/domains/invoice/publisher.go

package invoice

import (
	"context"

	"billing/invoice/domains/invoice/events"
)

// port: where the facts go
type Publisher interface {
	Publish(ctx context.Context, facts ...events.Event) error
}

billing/invoice/domains/invoice/vo/money.go

package vo

// the amount in minor units
type Money struct {
	minor int64
}

func NewMoney(minor int64) Money {
	return Money{minor: minor}
}

func (m Money) Add(other Money) Money {
	return Money{minor: m.minor + other.minor}
}

func (m Money) Equal(other Money) bool {
	return m.minor == other.minor
}

func (m Money) Get() int64 {
	return m.minor
}

billing/invoice/domains/invoice/rules/overdue.go

package rules

import (
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the specification contract: one question about one aggregate
type Specification interface {
	IsSatisfiedBy(inv *invoice.Invoice) bool
}

// the predicate: is this invoice overdue?
type Overdue struct {
	now      time.Time
	calendar invoice.Calendar
}

func NewOverdue(now time.Time, calendar invoice.Calendar) Overdue {
	return Overdue{now: now, calendar: calendar}
}

// a paid invoice is never overdue
func (o Overdue) IsSatisfiedBy(inv *invoice.Invoice) bool {
	if inv.Status() == invoice.StatusPaid {
		return false
	}

	return o.calendar.Workdays(inv.DueOn(), o.now) > 0
}

// a second predicate: the invoice is above the limit
type OverBudget struct {
	limit vo.Money
}

func NewOverBudget(limit vo.Money) OverBudget {
	return OverBudget{limit: limit}
}

func (b OverBudget) IsSatisfiedBy(inv *invoice.Invoice) bool {
	return inv.Total().Get() > b.limit.Get()
}

// composition: requires every predicate at once
type All struct {
	specs []Specification
}

func AllOf(specs ...Specification) All {
	return All{specs: specs}
}

func (a All) IsSatisfiedBy(inv *invoice.Invoice) bool {
	for _, spec := range a.specs {
		if !spec.IsSatisfiedBy(inv) {
			return false
		}
	}

	return true
}

billing/invoice/applications/overdue/charge.go

package overdue

import (
	"context"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/rules"
	"billing/invoice/domains/invoice/services/penalty"
	"billing/invoice/domains/invoice/vo"
)

// two ports, a calendar and a rate — all the use case needs
type UseCase struct {
	uow        invoice.UnitOfWork
	invoices   invoice.Repository
	publisher  invoice.Publisher
	calendar   invoice.Calendar
	perWorkday vo.Money
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
	calendar invoice.Calendar,
	perWorkday vo.Money,
) *UseCase {
	return &UseCase{
		uow:        uow,
		invoices:   invoices,
		publisher:  publisher,
		calendar:   calendar,
		perWorkday: perWorkday,
	}
}

// load, calculate, apply, save, tell
func (u *UseCase) Run(ctx context.Context, number invoice.Number, now time.Time) error {
	return u.uow.Do(ctx, func(ctx context.Context) error {
		overdue, err := u.invoices.ByNumber(ctx, number)
		if err != nil {
			return err
		}

		// the use case asks the rule instead of trusting the caller
		if !rules.NewOverdue(now, u.calendar).IsSatisfiedBy(overdue) {
			return nil
		}

		// the rule computes the amount from the invoice, the rate and the calendar
		amount := penalty.Amount(overdue, now, u.perWorkday, u.calendar)

		// only the invoice itself can charge the fee to the invoice
		if err := overdue.ChargePenalty(amount); err != nil {
			return err
		}

		if err := u.invoices.Save(ctx, overdue); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, overdue.Events()...)
	})
}

billing/invoice/applications/issuing/issue.go

package issuing

import (
	"context"
	"time"

	"billing/invoice/domains/invoice"
)

// the dependencies are domain ports and nothing else
type UseCase struct {
	uow       invoice.UnitOfWork
	invoices  invoice.Repository
	publisher invoice.Publisher
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
) *UseCase {
	return &UseCase{uow: uow, invoices: invoices, publisher: publisher}
}

// one argument — the command, and it checks its own shape
func (u *UseCase) Handle(ctx context.Context, cmd IssueInvoice) error {
	if err := cmd.Validate(); err != nil {
		return err
	}

	issued, err := invoice.Issue(cmd.Number, cmd.Customer, cmd.Lines, cmd.DueOn)
	if err != nil {
		return err
	}

	return u.uow.Do(ctx, func(ctx context.Context) error {
		if err := u.invoices.Save(ctx, issued); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, issued.Events()...)
	})
}

billing/invoice/applications/payment/pay.go

package payment

import (
	"context"

	"billing/invoice/domains/invoice"
)

// the dependencies are domain ports and nothing else
type UseCase struct {
	uow       invoice.UnitOfWork
	invoices  invoice.Repository
	publisher invoice.Publisher
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
) *UseCase {
	return &UseCase{uow: uow, invoices: invoices, publisher: publisher}
}

// load, call the method, save, tell — as one operation
func (u *UseCase) Run(ctx context.Context, number invoice.Number) error {
	return u.uow.Do(ctx, func(ctx context.Context) error {
		paid, err := u.invoices.ByNumber(ctx, number)
		if err != nil {
			return err
		}

		if err := paid.MarkPaid(); err != nil {
			return err
		}

		if err := u.invoices.Save(ctx, paid); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, paid.Events()...)
	})
}

billing/invoice/applications/payment/pay.go

package payment

import (
	"context"
	"errors"

	"billing/invoice/domains/invoice"
)

// how many times to try on a conflict
const attempts = 3

// the dependencies are domain ports and nothing else
type UseCase struct {
	uow       invoice.UnitOfWork
	invoices  invoice.Repository
	publisher invoice.Publisher
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
) *UseCase {
	return &UseCase{uow: uow, invoices: invoices, publisher: publisher}
}

// retries the whole attempt while someone else is changing the invoice
func (u *UseCase) Run(ctx context.Context, number invoice.Number) error {
	var err error

	for attempt := 0; attempt < attempts; attempt++ {
		err = u.pay(ctx, number)

		// a conflict is not a bad call: the data went stale
		if errors.Is(err, invoice.ErrConflict) {
			continue
		}

		return err
	}

	return err
}

// one attempt: load, apply, save, tell
func (u *UseCase) pay(ctx context.Context, number invoice.Number) error {
	return u.uow.Do(ctx, func(ctx context.Context) error {
		paid, err := u.invoices.ByNumber(ctx, number)
		if err != nil {
			return err
		}

		if err := paid.MarkPaid(); err != nil {
			return err
		}

		if err := u.invoices.Save(ctx, paid); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, paid.Events()...)
	})
}

billing/invoice/domains/invoice/load.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/vo"
)

// rebuilds the invoice from what was in the database
func Load(
	number Number,
	customer CustomerID,
	lines []Line,
	total vo.Money,
	dueOn time.Time,
	status Status,
) *Invoice {
	return &Invoice{
		number:   number,
		customer: customer,
		lines:    lines,
		total:    total,
		dueOn:    dueOn,
		status:   status,
	}
}

billing/invoice/domains/invoice/load.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/vo"
)

// rebuilds the invoice from what was in the database
func Load(
	number Number,
	customer CustomerID,
	lines []Line,
	total vo.Money,
	dueOn time.Time,
	status Status,
	// the version arrives together with the data
	version int,
) *Invoice {
	return &Invoice{
		number:   number,
		customer: customer,
		lines:    lines,
		total:    total,
		dueOn:    dueOn,
		status:   status,
		version:  version,
	}
}

billing/invoice/domains/invoice/uow.go

package invoice

import "context"

// port: run the work as a whole or do not run it
type UnitOfWork interface {
	Do(ctx context.Context, work func(ctx context.Context) error) error
}

billing/invoice/infrastructure/postgres/invoices.go

package postgres

import (
	"context"
	"database/sql"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the storage port implemented on Postgres
type Invoices struct {
	db *sql.DB
}

func NewInvoices(db *sql.DB) *Invoices {
	return &Invoices{db: db}
}

// loads the invoice and rebuilds the aggregate
func (r *Invoices) ByNumber(ctx context.Context, number invoice.Number) (*invoice.Invoice, error) {
	var (
		customer string
		minor    int64
		dueOn    time.Time
		status   string
	)

	err := conn(ctx, r.db).
		QueryRowContext(ctx, selectInvoice, string(number)).
		Scan(&customer, &minor, &dueOn, &status)
	if err != nil {
		return nil, err
	}

	lines, err := r.lines(ctx, number)
	if err != nil {
		return nil, err
	}

	// rebuilding the aggregate is the domain's job
	return invoice.Load(
		number,
		invoice.CustomerID(customer),
		lines,
		vo.NewMoney(minor),
		dueOn,
		invoice.Status(status),
	), nil
}

// writes the invoice and its lines
func (r *Invoices) Save(ctx context.Context, saved *invoice.Invoice) error {
	db := conn(ctx, r.db)

	if _, err := db.ExecContext(ctx, upsertInvoice, saved); err != nil {
		return err
	}

	_, err := db.ExecContext(ctx, replaceLines, saved)

	return err
}

billing/invoice/infrastructure/postgres/invoices.go

package postgres

import (
	"context"
	"database/sql"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the storage port implemented on Postgres
type Invoices struct {
	db *sql.DB
}

func NewInvoices(db *sql.DB) *Invoices {
	return &Invoices{db: db}
}

// loads the invoice together with the row version
func (r *Invoices) ByNumber(ctx context.Context, number invoice.Number) (*invoice.Invoice, error) {
	var (
		customer string
		minor    int64
		dueOn    time.Time
		status   string
		version  int
	)

	err := conn(ctx, r.db).
		QueryRowContext(ctx, selectInvoice, string(number)).
		Scan(&customer, &minor, &dueOn, &status, &version)
	if err != nil {
		return nil, err
	}

	lines, err := r.lines(ctx, number)
	if err != nil {
		return nil, err
	}

	return invoice.Load(
		number,
		invoice.CustomerID(customer),
		lines,
		vo.NewMoney(minor),
		dueOn,
		invoice.Status(status),
		version,
	), nil
}

// writes the invoice if its version in the database is still the same
func (r *Invoices) Save(ctx context.Context, saved *invoice.Invoice) error {
	db := conn(ctx, r.db)

	// version zero — the invoice is new, so it is inserted
	if saved.Version() == 0 {
		if _, err := db.ExecContext(ctx, insertInvoice, saved); err != nil {
			return err
		}

		_, err := db.ExecContext(ctx, replaceLines, saved)

		return err
	}

	// the version check sits inside the UPDATE itself
	result, err := db.ExecContext(ctx, updateInvoice, saved, saved.Version())
	if err != nil {
		return err
	}

	rows, err := result.RowsAffected()
	if err != nil {
		return err
	}

	// no rows — the version in the database is a different one
	if rows == 0 {
		return invoice.ErrConflict
	}

	_, err = db.ExecContext(ctx, replaceLines, saved)

	return err
}

billing/invoice/infrastructure/postgres/uow.go

package postgres

import (
	"context"
	"database/sql"
)

type txKey struct{}

// the unit of work implemented as a Postgres transaction
type UnitOfWork struct {
	db *sql.DB
}

func NewUnitOfWork(db *sql.DB) *UnitOfWork {
	return &UnitOfWork{db: db}
}

// all of the work inside one transaction
func (u *UnitOfWork) Do(ctx context.Context, work func(ctx context.Context) error) error {
	tx, err := u.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}

	if err := work(context.WithValue(ctx, txKey{}, tx)); err != nil {
		// an error inside — rolled back as a whole
		_ = tx.Rollback()

		return err
	}

	return tx.Commit()
}

// the transaction from ctx, if the operation has already begun
func conn(ctx context.Context, db *sql.DB) interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
} {
	if tx, ok := ctx.Value(txKey{}).(*sql.Tx); ok {
		return tx
	}

	return db
}

billing/invoice/infrastructure/postgres/outbox.go

package postgres

import (
	"context"
	"database/sql"
	"encoding/json"

	"billing/invoice/domains/invoice/events"
)

// publishing implemented as rows in the database
type Outbox struct {
	db *sql.DB
}

func NewOutbox(db *sql.DB) *Outbox {
	return &Outbox{db: db}
}

// writes the facts into the same transaction
func (o *Outbox) Publish(ctx context.Context, facts ...events.Event) error {
	db := conn(ctx, o.db)

	for _, event := range facts {
		body, err := json.Marshal(event)
		if err != nil {
			return err
		}

		// the fact name and the message body
		if _, err := db.ExecContext(ctx, insertOutbox, event.FactName(), body); err != nil {
			return err
		}
	}

	return nil
}

billing/invoice/applications/issuing/transport/http/dto/issue.go

package dto

import (
	"errors"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the request body: what the data crosses the boundary in
type IssueRequest struct {
	Number   string    `json:"number"`
	Customer string    `json:"customer"`
	DueOn    time.Time `json:"due_on"`
	Lines    []Line    `json:"lines"`
}

// an invoice line the way it is sent
type Line struct {
	AmountMinor int64 `json:"amount_minor"`
}

var ErrEmptyNumber = errors.New("dto: number is empty")

// the shape of the request is checked, not the rules of the domain
func (r IssueRequest) Validate() error {
	if r.Number == "" {
		return ErrEmptyNumber
	}

	return nil
}

// what the use case needs from the request, already in domain types
type Issue struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

// translates the request body into domain types
func (r IssueRequest) ToDomain() Issue {
	lines := make([]invoice.Line, 0, len(r.Lines))
	for _, item := range r.Lines {
		lines = append(lines, invoice.NewLine(vo.NewMoney(item.AmountMinor)))
	}

	return Issue{
		Number:   invoice.Number(r.Number),
		Customer: invoice.CustomerID(r.Customer),
		Lines:    lines,
		DueOn:    r.DueOn,
	}
}

billing/invoice/applications/issuing/transport/http/handler.go

package http

import (
	"encoding/json"
	"net/http"

	"billing/invoice/applications/issuing"
	"billing/invoice/applications/issuing/transport/http/dto"
)

// the handler owns the use case and nothing else
type Handler struct {
	issue *issuing.UseCase
}

func NewHandler(issue *issuing.UseCase) *Handler {
	return &Handler{issue: issue}
}

// POST /invoices: parse, translate, call
func (h *Handler) Issue(w http.ResponseWriter, r *http.Request) {
	var body dto.IssueRequest

	// the body does not parse — that is the client's error
	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
		http.Error(w, "malformed json", http.StatusBadRequest)

		return
	}

	if err := body.Validate(); err != nil {
		http.Error(w, err.Error(), http.StatusUnprocessableEntity)

		return
	}

	// translating into domain types is the dto package's job
	args := body.ToDomain()

	// the transport builds the command — it travels on, not a handful of arguments
	cmd := issuing.IssueInvoice{
		Number:   args.Number,
		Customer: args.Customer,
		Lines:    args.Lines,
		DueOn:    args.DueOn,
	}

	// the use case gets domain entities, not a request
	err := h.issue.Handle(r.Context(), cmd)
	if err != nil {
		// a separate function turns a domain error into a status code
		fail(w, err)

		return
	}

	// 201 and an empty body: the client sent the number itself
	w.WriteHeader(http.StatusCreated)
}

billing/invoice/applications/issuing/transport/http/errors.go

package http

import (
	"errors"
	"log/slog"
	"net/http"

	"billing/invoice/domains/invoice"
)

// turns a domain error into a status code
func fail(w http.ResponseWriter, err error) {
	switch {
	// a broken domain rule is the request's error, not the server's
	case errors.Is(err, invoice.ErrNoLines):
		http.Error(w, "invoice has no lines", http.StatusUnprocessableEntity)
	case errors.Is(err, invoice.ErrAlreadyPaid):
		http.Error(w, "invoice is already paid", http.StatusConflict)
	default:
		// the cause stays in the log, the code goes out
		slog.Error("issue invoice", "error", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
	}
}

billing/invoice/applications/issuing/transport/http/routes.go

package http

import "net/http"

// the module declares the address of its own handler
func (h *Handler) Register(mux *http.ServeMux) {
	mux.HandleFunc("POST /invoices", h.Issue)
}

billing/invoice/cmd/api/main.go

package main

import (
	"log"
	"net/http"
	"os"

	_ "github.com/lib/pq"
)

// the assembly moved into a generated function
func main() {
	handler, err := initHandler(os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatal(err)
	}

	mux := http.NewServeMux()
	handler.Register(mux)

	log.Fatal(http.ListenAndServe(":8080", mux))
}

billing/invoice/cmd/api/wire.go

//go:build wireinject

package main

import (
	"database/sql"

	"github.com/google/wire"

	"billing/invoice/domains/invoice"
	"billing/invoice/infrastructure/postgres"
	"billing/invoice/applications/issuing"
	issuinghttp "billing/invoice/applications/issuing/transport/http"
)

// the set of adapters: one for all applications of the service
var adapters = wire.NewSet(
	postgres.NewUnitOfWork,
	postgres.NewInvoices,
	postgres.NewOutbox,
	// which implementation stands behind the port
	wire.Bind(new(invoice.UnitOfWork), new(*postgres.UnitOfWork)),
	wire.Bind(new(invoice.Repository), new(*postgres.Invoices)),
	wire.Bind(new(invoice.Publisher), new(*postgres.Outbox)),
)

// what to assemble — stated by the result type
func initHandler(dsn string) (*issuinghttp.Handler, error) {
	wire.Build(openDB, adapters, issuing.New, issuinghttp.NewHandler)

	// a stub body: the real one is written by the generator
	return nil, nil
}

func openDB(dsn string) (*sql.DB, error) {
	return sql.Open("postgres", dsn)
}

billing/invoice/applications/issuing/command.go

package issuing

import (
	"errors"
	"time"

	"billing/invoice/domains/invoice"
)

// the command: what to do and with what data
type IssueInvoice struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

var ErrNoNumber = errors.New("issuing: invoice number is empty")

// the shape is checked here, not the domain rules
func (c IssueInvoice) Validate() error {
	if c.Number == "" {
		return ErrNoNumber
	}

	return nil
}

billing/invoice/applications/views/card.go

package views

import (
	"context"

	"billing/invoice/domains/invoice"
)

// an invoice card for the screen, not an aggregate
type Card struct {
	Number  string
	Total   int64
	Status  string
	Overdue bool
}

// the query: what is asked for, named as a type
type CardQuery struct {
	Number invoice.Number
}

// the read port: hands out a ready card
type Cards interface {
	Card(ctx context.Context, query CardQuery) (Card, error)
}

billing/invoice/infrastructure/postgres/cards.go

package postgres

import (
	"context"
	"database/sql"

	"billing/invoice/applications/views"
	"billing/invoice/domains/invoice"
)

// the query returns exactly what the screen shows
const selectCard = `
SELECT number,
       total_minor,
       status,
       due_on < now() AND status <> 'paid' AS overdue
  FROM invoices
 WHERE number = $1`

// the read port implemented: one query, no aggregate
type Cards struct {
	db *sql.DB
}

func NewCards(db *sql.DB) *Cards {
	return &Cards{db: db}
}

// reads an invoice card by its number
func (c *Cards) Card(ctx context.Context, query views.CardQuery) (views.Card, error) {
	var card views.Card

	// the columns land straight in the projection
	err := c.db.QueryRowContext(ctx, selectCard, string(query.Number)).
		Scan(&card.Number, &card.Total, &card.Status, &card.Overdue)
	if err != nil {
		return views.Card{}, err
	}

	return card, nil
}

Step 8 of 8

What we ended up with

Click a file to read it.

domains/invoice/invoice.go — the version field and Version().
domains/invoice/load.go — the version arrives from the database with the data.
domains/invoice/repository.go — ErrConflict next to the port.
infrastructure/postgres/invoices.go — the version check and RowsAffected.
applications/payment/pay.go — the whole attempt retried.

The domain grew by one field and one error. Everything else is a write that checks, and a use case that knows how to start over.

billing/GLOSSARY.md

Glossary

One meaning per word inside this context. The code, the events, the API and the person on the call all spell it the same way.

“Payment” in three departments

Department	What it calls a payment	What it is called here
Product	The purchase went through, the customer saw the checkmark	An order event. Not here
Processing	The money is held, nothing was captured	Authorization
Finance	The money showed up in a statement two days later	Settlement

Words

Authorization. The issuer agreed to hold the amount. No money moved, and the hold expires on its own.

Capture. The instruction to take what was held. From here on the money has left the payer’s account.

Settlement. The money reached our account, and the provider’s statement says so. It arrives days later, and it may not arrive at all.

Payment. One attempt to collect money for one order, with a lifecycle of its own: authorized, captured, refunded. The word means this and nothing else.

Words that are not here

Purchase. A product event: the customer saw the checkmark. It lives with orders, and there is nothing for it to mean here.

billing/invoice/domains/invoice/invoice.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/events"
	"billing/invoice/domains/invoice/vo"
)

// the invoice: the aggregate root
type Invoice struct {
	number   Number
	customer CustomerID
	lines    []Line
	total    vo.Money
	dueOn    time.Time
	status   Status
	facts    []events.Event
}

// creates the invoice and sums the lines
func Issue(
	number Number,
	customer CustomerID,
	lines []Line,
	dueOn time.Time,
) (*Invoice, error) {
	if len(lines) == 0 {
		return nil, ErrNoLines
	}

	total := vo.NewMoney(0)
	for _, line := range lines {
		total = total.Add(line.Amount())
	}

	invoice := &Invoice{
		number:   number,
		customer: customer,
		// a copy of the lines: they cannot be changed from outside
		lines:    append([]Line(nil), lines...),
		total:    total,
		dueOn:    dueOn,
		status:   StatusIssued,
	}

	// records the fact; sending it is not the aggregate's job
	invoice.facts = append(invoice.facts, events.Issued{
		Number:   string(number),
		Customer: string(customer),
		Total:    total,
	})

	return invoice, nil
}

// hands the collected facts to the outside
func (i *Invoice) Events() []events.Event { return i.facts }

func (i *Invoice) Total() vo.Money  { return i.total }
func (i *Invoice) DueOn() time.Time { return i.dueOn }
func (i *Invoice) Status() Status   { return i.status }

billing/invoice/domains/invoice/invoice.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/events"
	"billing/invoice/domains/invoice/vo"
)

// the invoice: the aggregate root
type Invoice struct {
	number   Number
	customer CustomerID
	lines    []Line
	total    vo.Money
	dueOn    time.Time
	status   Status
	// the number of the state that was read from the database
	version  int
	facts    []events.Event
}

// creates the invoice; the version stays zero — it is not in the database yet
func Issue(
	number Number,
	customer CustomerID,
	lines []Line,
	dueOn time.Time,
) (*Invoice, error) {
	if len(lines) == 0 {
		return nil, ErrNoLines
	}

	total := vo.NewMoney(0)
	for _, line := range lines {
		total = total.Add(line.Amount())
	}

	invoice := &Invoice{
		number:   number,
		customer: customer,
		// a copy of the lines: they cannot be changed from outside
		lines:    append([]Line(nil), lines...),
		total:    total,
		dueOn:    dueOn,
		status:   StatusIssued,
	}

	// records the fact; sending it is not the aggregate's job
	invoice.facts = append(invoice.facts, events.Issued{
		Number:   string(number),
		Customer: string(customer),
		Total:    total,
	})

	return invoice, nil
}

// hands the collected facts to the outside
func (i *Invoice) Events() []events.Event { return i.facts }

func (i *Invoice) Total() vo.Money  { return i.total }
func (i *Invoice) DueOn() time.Time { return i.dueOn }
func (i *Invoice) Status() Status   { return i.status }

// returns the version; nothing outside can change it
func (i *Invoice) Version() int { return i.version }

billing/invoice/domains/invoice/paid.go

package invoice

import "billing/invoice/domains/invoice/events"

// marks the invoice paid; paying twice is not allowed
func (i *Invoice) MarkPaid() error {
	if i.status == StatusPaid {
		return ErrAlreadyPaid
	}

	i.status = StatusPaid

	// records the fact; sending it is not the aggregate's job
	i.facts = append(i.facts, events.Paid{Number: string(i.number)})

	return nil
}

billing/invoice/domains/invoice/charged.go

package invoice

import (
	"billing/invoice/domains/invoice/events"
	"billing/invoice/domains/invoice/vo"
)

// puts the fee on the invoice; a paid invoice takes none
func (i *Invoice) ChargePenalty(amount vo.Money) error {
	if i.status == StatusPaid {
		return ErrAlreadyPaid
	}

	if amount.Equal(vo.NewMoney(0)) {
		return nil
	}

	i.lines = append(i.lines, Line{Quantity: 1, Price: amount})
	i.total = i.total.Add(amount)

	// records the fact; sending it is not the aggregate's job
	i.facts = append(i.facts, events.PenaltyCharged{
		Number: string(i.number),
		Amount: amount,
	})

	return nil
}

billing/invoice/domains/invoice/events/event.go

package events

import "billing/invoice/domains/invoice/vo"

// a domain fact; its name travels to the bus as the message type
type Event interface {
	FactName() string
}

// the invoice was issued
type Issued struct {
	Number   string
	Customer string
	Total    vo.Money
}

func (Issued) FactName() string { return "invoice.issued" }

// the invoice was paid
type Paid struct {
	Number string
}

func (Paid) FactName() string { return "invoice.paid" }

// the invoice was charged a late fee
type PenaltyCharged struct {
	Number string
	Amount vo.Money
}

func (PenaltyCharged) FactName() string { return "invoice.penalty_charged" }

billing/invoice/domains/invoice/calendar.go

package invoice

import "time"

// the working calendar: part of the model, not a storage port
type Calendar interface {
	Workdays(from time.Time, to time.Time) int
}

billing/invoice/domains/invoice/services/penalty/penalty.go

package penalty

import (
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the rule: an invoice, a rate and a calendar — an amount out
func Amount(
	overdue *invoice.Invoice,
	now time.Time,
	perWorkday vo.Money,
	calendar invoice.Calendar,
) vo.Money {
	// the rule takes the due date from the invoice itself, not from the caller
	late := calendar.Workdays(overdue.DueOn(), now)
	if late <= 0 {
		return vo.NewMoney(0)
	}

	return vo.NewMoney(perWorkday.Get() * int64(late))
}

billing/invoice/domains/invoice/repository.go

package invoice

import "context"

// port: how the invoice gets into the database and comes back
type Repository interface {
	ByNumber(ctx context.Context, number Number) (*Invoice, error)
	Save(ctx context.Context, invoice *Invoice) error
}

billing/invoice/domains/invoice/repository.go

package invoice

import (
	"context"
	"errors"
)

// the invoice changed between the read and the write
var ErrConflict = errors.New("invoice: invoice changed since it was loaded")

// port: how the invoice gets into the database and comes back
type Repository interface {
	ByNumber(ctx context.Context, number Number) (*Invoice, error)
	// returns ErrConflict if the invoice was changed in the meantime
	Save(ctx context.Context, invoice *Invoice) error
}

billing/invoice/domains/invoice/publisher.go

package invoice

import (
	"context"

	"billing/invoice/domains/invoice/events"
)

// port: where the facts go
type Publisher interface {
	Publish(ctx context.Context, facts ...events.Event) error
}

billing/invoice/domains/invoice/vo/money.go

package vo

// the amount in minor units
type Money struct {
	minor int64
}

func NewMoney(minor int64) Money {
	return Money{minor: minor}
}

func (m Money) Add(other Money) Money {
	return Money{minor: m.minor + other.minor}
}

func (m Money) Equal(other Money) bool {
	return m.minor == other.minor
}

func (m Money) Get() int64 {
	return m.minor
}

billing/invoice/domains/invoice/rules/overdue.go

package rules

import (
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the specification contract: one question about one aggregate
type Specification interface {
	IsSatisfiedBy(inv *invoice.Invoice) bool
}

// the predicate: is this invoice overdue?
type Overdue struct {
	now      time.Time
	calendar invoice.Calendar
}

func NewOverdue(now time.Time, calendar invoice.Calendar) Overdue {
	return Overdue{now: now, calendar: calendar}
}

// a paid invoice is never overdue
func (o Overdue) IsSatisfiedBy(inv *invoice.Invoice) bool {
	if inv.Status() == invoice.StatusPaid {
		return false
	}

	return o.calendar.Workdays(inv.DueOn(), o.now) > 0
}

// a second predicate: the invoice is above the limit
type OverBudget struct {
	limit vo.Money
}

func NewOverBudget(limit vo.Money) OverBudget {
	return OverBudget{limit: limit}
}

func (b OverBudget) IsSatisfiedBy(inv *invoice.Invoice) bool {
	return inv.Total().Get() > b.limit.Get()
}

// composition: requires every predicate at once
type All struct {
	specs []Specification
}

func AllOf(specs ...Specification) All {
	return All{specs: specs}
}

func (a All) IsSatisfiedBy(inv *invoice.Invoice) bool {
	for _, spec := range a.specs {
		if !spec.IsSatisfiedBy(inv) {
			return false
		}
	}

	return true
}

billing/invoice/applications/overdue/charge.go

package overdue

import (
	"context"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/rules"
	"billing/invoice/domains/invoice/services/penalty"
	"billing/invoice/domains/invoice/vo"
)

// two ports, a calendar and a rate — all the use case needs
type UseCase struct {
	uow        invoice.UnitOfWork
	invoices   invoice.Repository
	publisher  invoice.Publisher
	calendar   invoice.Calendar
	perWorkday vo.Money
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
	calendar invoice.Calendar,
	perWorkday vo.Money,
) *UseCase {
	return &UseCase{
		uow:        uow,
		invoices:   invoices,
		publisher:  publisher,
		calendar:   calendar,
		perWorkday: perWorkday,
	}
}

// load, calculate, apply, save, tell
func (u *UseCase) Run(ctx context.Context, number invoice.Number, now time.Time) error {
	return u.uow.Do(ctx, func(ctx context.Context) error {
		overdue, err := u.invoices.ByNumber(ctx, number)
		if err != nil {
			return err
		}

		// the use case asks the rule instead of trusting the caller
		if !rules.NewOverdue(now, u.calendar).IsSatisfiedBy(overdue) {
			return nil
		}

		// the rule computes the amount from the invoice, the rate and the calendar
		amount := penalty.Amount(overdue, now, u.perWorkday, u.calendar)

		// only the invoice itself can charge the fee to the invoice
		if err := overdue.ChargePenalty(amount); err != nil {
			return err
		}

		if err := u.invoices.Save(ctx, overdue); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, overdue.Events()...)
	})
}

billing/invoice/applications/issuing/issue.go

package issuing

import (
	"context"
	"time"

	"billing/invoice/domains/invoice"
)

// the dependencies are domain ports and nothing else
type UseCase struct {
	uow       invoice.UnitOfWork
	invoices  invoice.Repository
	publisher invoice.Publisher
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
) *UseCase {
	return &UseCase{uow: uow, invoices: invoices, publisher: publisher}
}

// one argument — the command, and it checks its own shape
func (u *UseCase) Handle(ctx context.Context, cmd IssueInvoice) error {
	if err := cmd.Validate(); err != nil {
		return err
	}

	issued, err := invoice.Issue(cmd.Number, cmd.Customer, cmd.Lines, cmd.DueOn)
	if err != nil {
		return err
	}

	return u.uow.Do(ctx, func(ctx context.Context) error {
		if err := u.invoices.Save(ctx, issued); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, issued.Events()...)
	})
}

billing/invoice/applications/payment/pay.go

package payment

import (
	"context"

	"billing/invoice/domains/invoice"
)

// the dependencies are domain ports and nothing else
type UseCase struct {
	uow       invoice.UnitOfWork
	invoices  invoice.Repository
	publisher invoice.Publisher
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
) *UseCase {
	return &UseCase{uow: uow, invoices: invoices, publisher: publisher}
}

// load, call the method, save, tell — as one operation
func (u *UseCase) Run(ctx context.Context, number invoice.Number) error {
	return u.uow.Do(ctx, func(ctx context.Context) error {
		paid, err := u.invoices.ByNumber(ctx, number)
		if err != nil {
			return err
		}

		if err := paid.MarkPaid(); err != nil {
			return err
		}

		if err := u.invoices.Save(ctx, paid); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, paid.Events()...)
	})
}

billing/invoice/applications/payment/pay.go

package payment

import (
	"context"
	"errors"

	"billing/invoice/domains/invoice"
)

// how many times to try on a conflict
const attempts = 3

// the dependencies are domain ports and nothing else
type UseCase struct {
	uow       invoice.UnitOfWork
	invoices  invoice.Repository
	publisher invoice.Publisher
}

func New(
	uow invoice.UnitOfWork,
	invoices invoice.Repository,
	publisher invoice.Publisher,
) *UseCase {
	return &UseCase{uow: uow, invoices: invoices, publisher: publisher}
}

// retries the whole attempt while someone else is changing the invoice
func (u *UseCase) Run(ctx context.Context, number invoice.Number) error {
	var err error

	for attempt := 0; attempt < attempts; attempt++ {
		err = u.pay(ctx, number)

		// a conflict is not a bad call: the data went stale
		if errors.Is(err, invoice.ErrConflict) {
			continue
		}

		return err
	}

	return err
}

// one attempt: load, apply, save, tell
func (u *UseCase) pay(ctx context.Context, number invoice.Number) error {
	return u.uow.Do(ctx, func(ctx context.Context) error {
		paid, err := u.invoices.ByNumber(ctx, number)
		if err != nil {
			return err
		}

		if err := paid.MarkPaid(); err != nil {
			return err
		}

		if err := u.invoices.Save(ctx, paid); err != nil {
			return err
		}

		return u.publisher.Publish(ctx, paid.Events()...)
	})
}

billing/invoice/domains/invoice/load.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/vo"
)

// rebuilds the invoice from what was in the database
func Load(
	number Number,
	customer CustomerID,
	lines []Line,
	total vo.Money,
	dueOn time.Time,
	status Status,
) *Invoice {
	return &Invoice{
		number:   number,
		customer: customer,
		lines:    lines,
		total:    total,
		dueOn:    dueOn,
		status:   status,
	}
}

billing/invoice/domains/invoice/load.go

package invoice

import (
	"time"

	"billing/invoice/domains/invoice/vo"
)

// rebuilds the invoice from what was in the database
func Load(
	number Number,
	customer CustomerID,
	lines []Line,
	total vo.Money,
	dueOn time.Time,
	status Status,
	// the version arrives together with the data
	version int,
) *Invoice {
	return &Invoice{
		number:   number,
		customer: customer,
		lines:    lines,
		total:    total,
		dueOn:    dueOn,
		status:   status,
		version:  version,
	}
}

billing/invoice/domains/invoice/uow.go

package invoice

import "context"

// port: run the work as a whole or do not run it
type UnitOfWork interface {
	Do(ctx context.Context, work func(ctx context.Context) error) error
}

billing/invoice/infrastructure/postgres/invoices.go

package postgres

import (
	"context"
	"database/sql"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the storage port implemented on Postgres
type Invoices struct {
	db *sql.DB
}

func NewInvoices(db *sql.DB) *Invoices {
	return &Invoices{db: db}
}

// loads the invoice and rebuilds the aggregate
func (r *Invoices) ByNumber(ctx context.Context, number invoice.Number) (*invoice.Invoice, error) {
	var (
		customer string
		minor    int64
		dueOn    time.Time
		status   string
	)

	err := conn(ctx, r.db).
		QueryRowContext(ctx, selectInvoice, string(number)).
		Scan(&customer, &minor, &dueOn, &status)
	if err != nil {
		return nil, err
	}

	lines, err := r.lines(ctx, number)
	if err != nil {
		return nil, err
	}

	// rebuilding the aggregate is the domain's job
	return invoice.Load(
		number,
		invoice.CustomerID(customer),
		lines,
		vo.NewMoney(minor),
		dueOn,
		invoice.Status(status),
	), nil
}

// writes the invoice and its lines
func (r *Invoices) Save(ctx context.Context, saved *invoice.Invoice) error {
	db := conn(ctx, r.db)

	if _, err := db.ExecContext(ctx, upsertInvoice, saved); err != nil {
		return err
	}

	_, err := db.ExecContext(ctx, replaceLines, saved)

	return err
}

billing/invoice/infrastructure/postgres/invoices.go

package postgres

import (
	"context"
	"database/sql"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the storage port implemented on Postgres
type Invoices struct {
	db *sql.DB
}

func NewInvoices(db *sql.DB) *Invoices {
	return &Invoices{db: db}
}

// loads the invoice together with the row version
func (r *Invoices) ByNumber(ctx context.Context, number invoice.Number) (*invoice.Invoice, error) {
	var (
		customer string
		minor    int64
		dueOn    time.Time
		status   string
		version  int
	)

	err := conn(ctx, r.db).
		QueryRowContext(ctx, selectInvoice, string(number)).
		Scan(&customer, &minor, &dueOn, &status, &version)
	if err != nil {
		return nil, err
	}

	lines, err := r.lines(ctx, number)
	if err != nil {
		return nil, err
	}

	return invoice.Load(
		number,
		invoice.CustomerID(customer),
		lines,
		vo.NewMoney(minor),
		dueOn,
		invoice.Status(status),
		version,
	), nil
}

// writes the invoice if its version in the database is still the same
func (r *Invoices) Save(ctx context.Context, saved *invoice.Invoice) error {
	db := conn(ctx, r.db)

	// version zero — the invoice is new, so it is inserted
	if saved.Version() == 0 {
		if _, err := db.ExecContext(ctx, insertInvoice, saved); err != nil {
			return err
		}

		_, err := db.ExecContext(ctx, replaceLines, saved)

		return err
	}

	// the version check sits inside the UPDATE itself
	result, err := db.ExecContext(ctx, updateInvoice, saved, saved.Version())
	if err != nil {
		return err
	}

	rows, err := result.RowsAffected()
	if err != nil {
		return err
	}

	// no rows — the version in the database is a different one
	if rows == 0 {
		return invoice.ErrConflict
	}

	_, err = db.ExecContext(ctx, replaceLines, saved)

	return err
}

billing/invoice/infrastructure/postgres/uow.go

package postgres

import (
	"context"
	"database/sql"
)

type txKey struct{}

// the unit of work implemented as a Postgres transaction
type UnitOfWork struct {
	db *sql.DB
}

func NewUnitOfWork(db *sql.DB) *UnitOfWork {
	return &UnitOfWork{db: db}
}

// all of the work inside one transaction
func (u *UnitOfWork) Do(ctx context.Context, work func(ctx context.Context) error) error {
	tx, err := u.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}

	if err := work(context.WithValue(ctx, txKey{}, tx)); err != nil {
		// an error inside — rolled back as a whole
		_ = tx.Rollback()

		return err
	}

	return tx.Commit()
}

// the transaction from ctx, if the operation has already begun
func conn(ctx context.Context, db *sql.DB) interface {
	ExecContext(context.Context, string, ...any) (sql.Result, error)
	QueryRowContext(context.Context, string, ...any) *sql.Row
} {
	if tx, ok := ctx.Value(txKey{}).(*sql.Tx); ok {
		return tx
	}

	return db
}

billing/invoice/infrastructure/postgres/outbox.go

package postgres

import (
	"context"
	"database/sql"
	"encoding/json"

	"billing/invoice/domains/invoice/events"
)

// publishing implemented as rows in the database
type Outbox struct {
	db *sql.DB
}

func NewOutbox(db *sql.DB) *Outbox {
	return &Outbox{db: db}
}

// writes the facts into the same transaction
func (o *Outbox) Publish(ctx context.Context, facts ...events.Event) error {
	db := conn(ctx, o.db)

	for _, event := range facts {
		body, err := json.Marshal(event)
		if err != nil {
			return err
		}

		// the fact name and the message body
		if _, err := db.ExecContext(ctx, insertOutbox, event.FactName(), body); err != nil {
			return err
		}
	}

	return nil
}

billing/invoice/applications/issuing/transport/http/dto/issue.go

package dto

import (
	"errors"
	"time"

	"billing/invoice/domains/invoice"
	"billing/invoice/domains/invoice/vo"
)

// the request body: what the data crosses the boundary in
type IssueRequest struct {
	Number   string    `json:"number"`
	Customer string    `json:"customer"`
	DueOn    time.Time `json:"due_on"`
	Lines    []Line    `json:"lines"`
}

// an invoice line the way it is sent
type Line struct {
	AmountMinor int64 `json:"amount_minor"`
}

var ErrEmptyNumber = errors.New("dto: number is empty")

// the shape of the request is checked, not the rules of the domain
func (r IssueRequest) Validate() error {
	if r.Number == "" {
		return ErrEmptyNumber
	}

	return nil
}

// what the use case needs from the request, already in domain types
type Issue struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

// translates the request body into domain types
func (r IssueRequest) ToDomain() Issue {
	lines := make([]invoice.Line, 0, len(r.Lines))
	for _, item := range r.Lines {
		lines = append(lines, invoice.NewLine(vo.NewMoney(item.AmountMinor)))
	}

	return Issue{
		Number:   invoice.Number(r.Number),
		Customer: invoice.CustomerID(r.Customer),
		Lines:    lines,
		DueOn:    r.DueOn,
	}
}

billing/invoice/applications/issuing/transport/http/handler.go

package http

import (
	"encoding/json"
	"net/http"

	"billing/invoice/applications/issuing"
	"billing/invoice/applications/issuing/transport/http/dto"
)

// the handler owns the use case and nothing else
type Handler struct {
	issue *issuing.UseCase
}

func NewHandler(issue *issuing.UseCase) *Handler {
	return &Handler{issue: issue}
}

// POST /invoices: parse, translate, call
func (h *Handler) Issue(w http.ResponseWriter, r *http.Request) {
	var body dto.IssueRequest

	// the body does not parse — that is the client's error
	if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
		http.Error(w, "malformed json", http.StatusBadRequest)

		return
	}

	if err := body.Validate(); err != nil {
		http.Error(w, err.Error(), http.StatusUnprocessableEntity)

		return
	}

	// translating into domain types is the dto package's job
	args := body.ToDomain()

	// the transport builds the command — it travels on, not a handful of arguments
	cmd := issuing.IssueInvoice{
		Number:   args.Number,
		Customer: args.Customer,
		Lines:    args.Lines,
		DueOn:    args.DueOn,
	}

	// the use case gets domain entities, not a request
	err := h.issue.Handle(r.Context(), cmd)
	if err != nil {
		// a separate function turns a domain error into a status code
		fail(w, err)

		return
	}

	// 201 and an empty body: the client sent the number itself
	w.WriteHeader(http.StatusCreated)
}

billing/invoice/applications/issuing/transport/http/errors.go

package http

import (
	"errors"
	"log/slog"
	"net/http"

	"billing/invoice/domains/invoice"
)

// turns a domain error into a status code
func fail(w http.ResponseWriter, err error) {
	switch {
	// a broken domain rule is the request's error, not the server's
	case errors.Is(err, invoice.ErrNoLines):
		http.Error(w, "invoice has no lines", http.StatusUnprocessableEntity)
	case errors.Is(err, invoice.ErrAlreadyPaid):
		http.Error(w, "invoice is already paid", http.StatusConflict)
	default:
		// the cause stays in the log, the code goes out
		slog.Error("issue invoice", "error", err)
		http.Error(w, "internal error", http.StatusInternalServerError)
	}
}

billing/invoice/applications/issuing/transport/http/routes.go

package http

import "net/http"

// the module declares the address of its own handler
func (h *Handler) Register(mux *http.ServeMux) {
	mux.HandleFunc("POST /invoices", h.Issue)
}

billing/invoice/cmd/api/main.go

package main

import (
	"log"
	"net/http"
	"os"

	_ "github.com/lib/pq"
)

// the assembly moved into a generated function
func main() {
	handler, err := initHandler(os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatal(err)
	}

	mux := http.NewServeMux()
	handler.Register(mux)

	log.Fatal(http.ListenAndServe(":8080", mux))
}

billing/invoice/cmd/api/wire.go

//go:build wireinject

package main

import (
	"database/sql"

	"github.com/google/wire"

	"billing/invoice/domains/invoice"
	"billing/invoice/infrastructure/postgres"
	"billing/invoice/applications/issuing"
	issuinghttp "billing/invoice/applications/issuing/transport/http"
)

// the set of adapters: one for all applications of the service
var adapters = wire.NewSet(
	postgres.NewUnitOfWork,
	postgres.NewInvoices,
	postgres.NewOutbox,
	// which implementation stands behind the port
	wire.Bind(new(invoice.UnitOfWork), new(*postgres.UnitOfWork)),
	wire.Bind(new(invoice.Repository), new(*postgres.Invoices)),
	wire.Bind(new(invoice.Publisher), new(*postgres.Outbox)),
)

// what to assemble — stated by the result type
func initHandler(dsn string) (*issuinghttp.Handler, error) {
	wire.Build(openDB, adapters, issuing.New, issuinghttp.NewHandler)

	// a stub body: the real one is written by the generator
	return nil, nil
}

func openDB(dsn string) (*sql.DB, error) {
	return sql.Open("postgres", dsn)
}

billing/invoice/applications/issuing/command.go

package issuing

import (
	"errors"
	"time"

	"billing/invoice/domains/invoice"
)

// the command: what to do and with what data
type IssueInvoice struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

var ErrNoNumber = errors.New("issuing: invoice number is empty")

// the shape is checked here, not the domain rules
func (c IssueInvoice) Validate() error {
	if c.Number == "" {
		return ErrNoNumber
	}

	return nil
}

billing/invoice/applications/views/card.go

package views

import (
	"context"

	"billing/invoice/domains/invoice"
)

// an invoice card for the screen, not an aggregate
type Card struct {
	Number  string
	Total   int64
	Status  string
	Overdue bool
}

// the query: what is asked for, named as a type
type CardQuery struct {
	Number invoice.Number
}

// the read port: hands out a ready card
type Cards interface {
	Card(ctx context.Context, query CardQuery) (Card, error)
}

billing/invoice/infrastructure/postgres/cards.go

package postgres

import (
	"context"
	"database/sql"

	"billing/invoice/applications/views"
	"billing/invoice/domains/invoice"
)

// the query returns exactly what the screen shows
const selectCard = `
SELECT number,
       total_minor,
       status,
       due_on < now() AND status <> 'paid' AS overdue
  FROM invoices
 WHERE number = $1`

// the read port implemented: one query, no aggregate
type Cards struct {
	db *sql.DB
}

func NewCards(db *sql.DB) *Cards {
	return &Cards{db: db}
}

// reads an invoice card by its number
func (c *Cards) Card(ctx context.Context, query views.CardQuery) (views.Card, error) {
	var card views.Card

	// the columns land straight in the projection
	err := c.db.QueryRowContext(ctx, selectCard, string(query.Number)).
		Scan(&card.Number, &card.Total, &card.Status, &card.Overdue)
	if err != nil {
		return views.Card{}, err
	}

	return card, nil
}

Use ← and → to move between steps.