Why integration events

ПроблемаДоменный факт уехал на шину как есть, со своими полями и типами. Сосед разобрал его и построился на нём: теперь переименование поля внутри домена ломает чужой сервис, и узнаём мы об этом из его алертов.

Тот же сервис, что в главах про DDD: счёт, его модули, порты и адаптеры. Здесь появляется граница контекста — та, за которой начинаются чужие сервисы. У факта на ней другое имя, другая форма и другие правила изменения.

Шаг 1 из 9

Домен уехал на шину

Справа публикация из главы про инфраструктуру. Факт уходит в outbox так, как он написан в домене: json.Marshal берёт структуру Issued целиком, а типом сообщения становится внутреннее имя invoice.issued.

Пока это читаем только мы, всё честно. Но сообщение прочитал сосед — доставка подписалась и начала собирать посылку. С этой минуты внутренняя структура домена стала контрактом: переименовал поле Total — сломал чужой парсер; добавил поле для себя — рассказал соседям то, чего им знать не нужно.

Работает это и в обратную сторону. Доставке понадобился адрес клиента, которого в факте нет, — и просьба добавить поле приходит в домен.

Доменный факт и сообщение наружу — разные вещи, и меняются они с разной скоростью. Первое мы правим когда захотим, второе — только договорившись со всеми, кто его читает.

Шаг 2 из 9

Интеграционное событие

Интеграционное событие — отдельный тип в отдельном пакете, и версия стоит прямо в его пути: infrastructure/broker/contracts/v1.

Лежит он в инфраструктуре шины, а не в корне сервиса, и это не мелочь. Форма сообщения — деталь того транспорта, которым оно едет: у шины это JSON в конверте с ключом партиции, у gRPC был бы protobuf со своими правилами совместимости, у HTTP — тело ответа и статус. Транспортов у сервиса бывает несколько, и у каждого контракт свой; общий каталог contracts делает вид, что он один.

Типы внутри примитивные: строки и int64. Ни vo.Money, ни invoice.Number за границу не уезжают — там JSON, и поля названы так, как их прочитает чужой сервис. Это DTO границы, и ничего больше.

Имя сообщения — константа с версией на конце: billing.invoice.issued.v1. По нему читателю видно, откуда сообщение, про что оно и какой схемы. Внутреннее FactName() осталось своим: его мы переименуем свободно, это имя — нет.

Набор полей здесь — решение, а не отражение факта. В контракт попадает то, что соседям действительно нужно: добавить поле потом можно, убрать — уже нет.

Шаг 3 из 9

Конверт

Данные факта — одно, метаданные сообщения — другое, поэтому они и лежат отдельно: тело в Payload, вокруг него идентификатор, тип, время и ключ.

ID нужен получателю: доставка at-least-once обещает, что сообщение придёт, но не обещает, что один раз. По этому идентификатору повтор узнают — к этому вернёмся на входящей стороне.

Key — про порядок. Брокер держит порядок внутри партиции, а не во всём топике; ключ, равный номеру счёта, кладёт все сообщения одного счёта в одну партицию: «выставлен» придёт раньше «оплачен». Между разными счетами порядка нет, и он не нужен.

Type стоит в конверте, а не только в теле: подписчик решает, разбирать ли тело, ещё не разобрав его.

Шаг 4 из 9

Кто переводит

Перевод живёт в инфраструктуре: домен про контракт не знает и знать не должен. Contract берёт доменный факт и отдаёт готовый конверт.

switch по типу факта выглядит скучно — в этом и смысл. Он явный: чтобы новое сообщение уехало наружу, кто-то должен добавить его здесь. Пока не добавил, факт остаётся внутренним, и публикация его пропускает — ErrInternalFact.

Именно это чаще всего и забывают. Публикуешь всё подряд — и каждый новый доменный факт становится публичным в день, когда его написали, молча и без чьего-либо решения.

Рядом — тот, кто Contract зовёт: broker.Publisher. Это он реализует порт invoice.Publisher, который домен объявил ещё во второй главе: перевести факт, сериализовать и отдать строку на запись. Отправлять он ничего не отправляет: шина в транзакцию не входит — это было в главе про инфраструктуру.

Записывает строку не он, а Appender — узкий порт из одного метода, объявленный здесь же, у того, кто им пользуется. За ним стоит postgres.Outbox, и после этой главы он занимается ровно одним делом: кладёт строку в таблицу. Ни домена, ни контракта в его импортах больше нет — раньше он знал и то, и другое.

Направление зависимостей от этого выпрямилось: знает про контракт тот, кто с шиной и работает, а хранилище — только про строки. Связывает их сборка: wire.Bind(new(broker.Appender), new(*postgres.Outbox)).

Шаг 5 из 9

Нам тоже присылают

Обратная сторона границы. Платёжный шлюз публикует payments.payment.received.v1, и по этому сообщению счёт становится оплаченным — значит, кто-то должен его принять и позвать сценарий справа.

Сделать это плохо можно двумя способами.

Первый: разобрать чужой JSON прямо в сценарии. Тогда application начинает знать формат соседа, и переименование поля в чужом сервисе доезжает до нашего домена.

Второй: позвать сценарий как есть. То же сообщение приедет второй раз — после ретрая брокера, после перезапуска консьюмера, после переезда партиции. Второй раз MarkPaid вернёт ErrAlreadyPaid, консьюмер сочтёт это ошибкой обработки и вернёт сообщение в очередь. И так по кругу, пока кто-нибудь не посмотрит в логи.

Шаг 6 из 9

Приём: перевести, вспомнить, простить

Хендлер лежит в слайсе платежа, рядом со своим сценарием, и делает три вещи.

Переводит. PaymentReceived — наша копия чужой схемы, и она стоит на границе. Изменится схема у соседа — правим этот файл; домен не заметит.

Помнит. Remember отвечает «впервые» или «уже было», и вызов стоит внутри той же единицы работы, что и сценарий. Иначе между пометкой и изменением остаётся щель, ровно в которую и попадает повтор. Поэтому реализация Do получила одну строку: транзакция в ctx уже есть — значит, выполняем работу внутри неё, а не начинаем вторую.

Прощает. ErrAlreadyPaid — не ошибка обработки: счёт уже в нужном состоянии, сообщение подтверждаем и идём дальше. Это и есть идемпотентность на входе — два одинаковых сообщения дают один результат.

Шаг 7 из 9

Память о сообщениях

Реализация простая: таблица с уникальным идентификатором и INSERT ... ON CONFLICT DO NOTHING. Вставилась строка — сообщение новое; ноль строк — это сообщение мы уже обрабатывали.

Соединение берётся тем же conn(ctx, db), что и у остальных адаптеров, — значит, пометка ложится в транзакцию операции. Либо счёт оплачен и сообщение помечено, либо ни того, ни другого.

О чём не надо забывать: таблица растёт. Идентификаторы старше срока хранения сообщений в брокере не нужны — их чистят по времени, партициями, как обычные логи.

Шаг 8 из 9

Как менять контракт

Правило одно, и оно не про код: схему читают чужие, значит, меняем её так, чтобы читатель не заметил.

Добавить необязательное поле — можно. Старый читатель его проигнорирует.
Переименовать, удалить, сменить смысл или тип — нельзя. Это v2: новый тип сообщения, который живёт рядом с v1, пока последний читатель не переехал.

Значит, надо знать, кто читает. Без этого v1 не выключить никогда — и версии будут копиться, а не сменяться.

И граница, за которую лучше не выходить: событие говорит «что случилось», а не «сделай». Как только в контракте появляется ChargePenalty, это уже не факт, а команда — и сосед начинает управлять нашим доменом через шину.

billing/GLOSSARY.md

Словарь

Одно значение на слово внутри контекста. Код, события, API и человек на созвоне пишут его одинаково.

«Платёж» в трёх отделах

Отдел	Что он называет платежом	Как это называется здесь
Продукт	Покупка прошла, пользователь увидел галочку	Событие заказа. Здесь его нет
Процессинг	Деньги захолдированы, списания не было	Авторизация
Бухгалтерия	Деньги пришли выпиской через два дня	Поступление

Слова

Авторизация. Эмитент согласился держать сумму. Деньги не двигались, холд истекает сам.

Списание. Команда забрать захолдированное. С этого момента деньги ушли со счёта плательщика.

Поступление. Деньги дошли до нашего счёта, и это подтверждено выпиской провайдера. Приходит днями позже и может не прийти вовсе.

Платёж. Попытка получить деньги за один заказ, со своим жизненным циклом: авторизован, списан, возвращён. Слово значит это и ничего больше.

Слова, которых здесь нет

Покупка. Продуктовое событие: пользователь увидел галочку. Живёт в заказах, здесь ему делать нечего.

billing/invoice/domains/invoice/invoice.go

package invoice

import (
	"time"

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

// счёт: корень агрегата
type Invoice struct {
	number   Number
	customer CustomerID
	lines    []Line
	total    vo.Money
	dueOn    time.Time
	status   Status
	// номер состояния, которое прочитали из базы
	version  int
	facts    []events.Event
}

// создаёт счёт; версия остаётся нулевой — в базе его ещё нет
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,
		// копия строк: снаружи их не поменять
		lines:    append([]Line(nil), lines...),
		total:    total,
		dueOn:    dueOn,
		status:   StatusIssued,
	}

	// записывает факт; отправлять его — не дело агрегата
	invoice.facts = append(invoice.facts, events.Issued{
		Number:   string(number),
		Customer: string(customer),
		Total:    total,
	})

	return invoice, nil
}

// отдаёт накопленные факты наружу
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 }

// отдаёт версию; менять её снаружи нельзя
func (i *Invoice) Version() int { return i.version }

billing/invoice/domains/invoice/paid.go

package invoice

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

// отмечает счёт оплаченным; дважды оплатить нельзя
func (i *Invoice) MarkPaid() error {
	if i.status == StatusPaid {
		return ErrAlreadyPaid
	}

	i.status = StatusPaid

	// записывает факт; отправлять его — не дело агрегата
	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"
)

// приписывает пеню счёту; оплаченный счёт её не берёт
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)

	// записывает факт; отправлять его — не дело агрегата
	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"

// факт домена; имя у него внутреннее
type Event interface {
	FactName() string
}

// счёт выставлен
type Issued struct {
	Number   string
	Customer string
	Total    vo.Money
}

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

// счёт оплачен
type Paid struct {
	Number string
}

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

// счёт получил пеню
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"

// календарь рабочих дней: часть модели, а не порт к базе
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"
)

// правило: счёт, ставка и календарь — на выходе сумма
func Amount(
	overdue *invoice.Invoice,
	now time.Time,
	perWorkday vo.Money,
	calendar invoice.Calendar,
) vo.Money {
	// срок правило берёт у самого счёта, а не у вызывающего
	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"
	"errors"
)

// счёт изменили между чтением и записью
var ErrConflict = errors.New("invoice: invoice changed since it was loaded")

// порт: как счёт попадает в базу и возвращается оттуда
type Repository interface {
	ByNumber(ctx context.Context, number Number) (*Invoice, error)
	// вернёт ErrConflict, если счёт успели изменить
	Save(ctx context.Context, invoice *Invoice) error
}

billing/invoice/domains/invoice/publisher.go

package invoice

import (
	"context"

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

// порт: куда уезжают факты
type Publisher interface {
	Publish(ctx context.Context, facts ...events.Event) error
}

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

package vo

// сумма в копейках
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"
)

// контракт спецификации: один вопрос про один агрегат
type Specification interface {
	IsSatisfiedBy(inv *invoice.Invoice) bool
}

// предикат: этот счёт просрочен?
type Overdue struct {
	now      time.Time
	calendar invoice.Calendar
}

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

// оплаченный счёт просроченным не бывает
func (o Overdue) IsSatisfiedBy(inv *invoice.Invoice) bool {
	if inv.Status() == invoice.StatusPaid {
		return false
	}

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

// второй предикат: счёт больше лимита
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()
}

// сборка: требует все предикаты сразу
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"
)

// два порта, календарь и ставка — всё, что нужно сценарию
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,
	}
}

// достать, посчитать, применить, сохранить, рассказать
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
		}

		// сценарий спрашивает у правила, а не верит вызывающему
		if !rules.NewOverdue(now, u.calendar).IsSatisfiedBy(overdue) {
			return nil
		}

		// правило считает сумму по счёту, ставке и календарю
		amount := penalty.Amount(overdue, now, u.perWorkday, u.calendar)

		// приписать пеню счёту может только сам счёт
		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"
)

// зависимости — только порты домена
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}
}

// один аргумент: команда, и она же проверяет свою форму
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"
	"errors"

	"billing/invoice/domains/invoice"
)

// сколько раз пробовать при конфликте
const attempts = 3

// зависимости — только порты домена
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}
}

// повторяет попытку целиком, пока счёт меняет кто-то другой
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)

		// конфликт — не ошибка вызова: данные устарели
		if errors.Is(err, invoice.ErrConflict) {
			continue
		}

		return err
	}

	return err
}

// одна попытка: достать, применить, сохранить, рассказать
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"
)

// собирает счёт из того, что лежало в базе
func Load(
	number Number,
	customer CustomerID,
	lines []Line,
	total vo.Money,
	dueOn time.Time,
	status Status,
	// версия приезжает вместе с данными
	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"

// порт: выполнить работу целиком или не выполнять
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"
)

// реализация порта хранения на Postgres
type Invoices struct {
	db *sql.DB
}

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

// достаёт счёт вместе с версией строки
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
}

// пишет счёт, если его версия в базе всё ещё та
func (r *Invoices) Save(ctx context.Context, saved *invoice.Invoice) error {
	db := conn(ctx, r.db)

	// нулевая версия — счёт новый, его вставляют
	if saved.Version() == 0 {
		if _, err := db.ExecContext(ctx, insertInvoice, saved); err != nil {
			return err
		}

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

		return err
	}

	// условие по версии стоит в самом UPDATE
	result, err := db.ExecContext(ctx, updateInvoice, saved, saved.Version())
	if err != nil {
		return err
	}

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

	// ни одной строки — значит, версия в базе другая
	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{}

// реализация единицы работы: транзакция Postgres
type UnitOfWork struct {
	db *sql.DB
}

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

// вся работа внутри одной транзакции
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 {
		// ошибка внутри — откат целиком
		_ = tx.Rollback()

		return err
	}

	return tx.Commit()
}

// транзакция из ctx, если операция уже началась
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/uow.go

package postgres

import (
	"context"
	"database/sql"
)

type txKey struct{}

// реализация единицы работы: транзакция Postgres
type UnitOfWork struct {
	db *sql.DB
}

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

// вся работа внутри одной транзакции
func (u *UnitOfWork) Do(ctx context.Context, work func(ctx context.Context) error) error {
	// операция уже началась — выполняем внутри неё, без второй транзакции
	if _, ok := ctx.Value(txKey{}).(*sql.Tx); ok {
		return work(ctx)
	}

	tx, err := u.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}

	if err := work(context.WithValue(ctx, txKey{}, tx)); err != nil {
		// ошибка внутри — откат целиком
		_ = tx.Rollback()

		return err
	}

	return tx.Commit()
}

// транзакция из ctx, если операция уже началась
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"
)

// реализация публикации: факты строками в базе
type Outbox struct {
	db *sql.DB
}

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

// пишет факты в ту же транзакцию
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
		}

		// имя факта и тело сообщения
		if _, err := db.ExecContext(ctx, insertOutbox, event.FactName(), body); err != nil {
			return err
		}
	}

	return nil
}

billing/invoice/infrastructure/postgres/outbox.go

package postgres

import (
	"context"
	"database/sql"
)

// реализация публикации: факты строками в базе
type Outbox struct {
	db *sql.DB
}

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

// пишет строку сообщения; про домен и контракт не знает
func (o *Outbox) Append(ctx context.Context, messageType string, key string, body []byte) error {
	// тип, ключ и тело — всё, что нужно отправителю
	_, err := conn(ctx, o.db).ExecContext(ctx, insertOutbox, messageType, key, body)

	return err
}

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

package dto

import (
	"errors"
	"time"

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

// тело запроса: то, в чём данные пересекают границу
type IssueRequest struct {
	Number   string    `json:"number"`
	Customer string    `json:"customer"`
	DueOn    time.Time `json:"due_on"`
	Lines    []Line    `json:"lines"`
}

// строка счёта так, как её присылают
type Line struct {
	AmountMinor int64 `json:"amount_minor"`
}

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

// проверяется форма запроса, а не правила предметной области
func (r IssueRequest) Validate() error {
	if r.Number == "" {
		return ErrEmptyNumber
	}

	return nil
}

// что из запроса нужно сценарию, уже в типах домена
type Issue struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

// переводит тело запроса в типы домена
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"
)

// ручка владеет сценарием и больше ничем
type Handler struct {
	issue *issuing.UseCase
}

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

// POST /invoices: разобрать, перевести, позвать
func (h *Handler) Issue(w http.ResponseWriter, r *http.Request) {
	var body dto.IssueRequest

	// тело не читается — это ошибка клиента
	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
	}

	// перевод в типы домена — дело пакета dto
	args := body.ToDomain()

	// транспорт собирает команду — дальше едет она, а не россыпь аргументов
	cmd := issuing.IssueInvoice{
		Number:   args.Number,
		Customer: args.Customer,
		Lines:    args.Lines,
		DueOn:    args.DueOn,
	}

	// сценарий получает доменные сущности, а не запрос
	err := h.issue.Handle(r.Context(), cmd)
	if err != nil {
		// ошибку домена переводит в код ответа отдельная функция
		fail(w, err)

		return
	}

	// 201 и пустое тело: номер клиент прислал сам
	w.WriteHeader(http.StatusCreated)
}

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

package http

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

	"billing/invoice/domains/invoice"
)

// переводит ошибку домена в код ответа
func fail(w http.ResponseWriter, err error) {
	switch {
	// правило домена нарушено — это ошибка запроса, а не сервера
	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:
		// причина остаётся в логе, наружу — код
		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"

// адрес ручки объявляет сам модуль
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"
)

// сборка уехала в сгенерированную функцию
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"
)

// набор адаптеров: один на все приложения сервиса
var adapters = wire.NewSet(
	postgres.NewUnitOfWork,
	postgres.NewInvoices,
	postgres.NewOutbox,
	// какая реализация стоит за портом
	wire.Bind(new(invoice.UnitOfWork), new(*postgres.UnitOfWork)),
	wire.Bind(new(invoice.Repository), new(*postgres.Invoices)),
	wire.Bind(new(invoice.Publisher), new(*postgres.Outbox)),
)

// что надо собрать — по типу результата
func initHandler(dsn string) (*issuinghttp.Handler, error) {
	wire.Build(openDB, adapters, issuing.New, issuinghttp.NewHandler)

	// тело-заглушка: настоящее напишет генератор
	return nil, nil
}

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

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/broker"
	"billing/invoice/infrastructure/postgres"
	"billing/invoice/applications/issuing"
	issuinghttp "billing/invoice/applications/issuing/transport/http"
)

// набор адаптеров: один на все приложения сервиса
var adapters = wire.NewSet(
	postgres.NewUnitOfWork,
	postgres.NewInvoices,
	postgres.NewOutbox,
	broker.NewPublisher,
	// какая реализация стоит за портом
	wire.Bind(new(invoice.UnitOfWork), new(*postgres.UnitOfWork)),
	wire.Bind(new(invoice.Repository), new(*postgres.Invoices)),
	// порт публикации теперь за брокером, а outbox — за его узким портом
	wire.Bind(new(invoice.Publisher), new(*broker.Publisher)),
	wire.Bind(new(broker.Appender), new(*postgres.Outbox)),
)

// что надо собрать — по типу результата
func initHandler(dsn string) (*issuinghttp.Handler, error) {
	wire.Build(openDB, adapters, issuing.New, issuinghttp.NewHandler)

	// тело-заглушка: настоящее напишет генератор
	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"
)

// команда: что сделать и с какими данными
type IssueInvoice struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

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

// проверяется форма, а не правила предметной области
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"
)

// карточка счёта для экрана, а не агрегат
type Card struct {
	Number  string
	Total   int64
	Status  string
	Overdue bool
}

// запрос: что просят прочитать, названное типом
type CardQuery struct {
	Number invoice.Number
}

// порт чтения: отдаёт готовую карточку
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"
)

// запрос отдаёт ровно то, что показывают на экране
const selectCard = `
SELECT number,
       total_minor,
       status,
       due_on < now() AND status <> 'paid' AS overdue
  FROM invoices
 WHERE number = $1`

// реализация порта чтения: один запрос, без агрегата
type Cards struct {
	db *sql.DB
}

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

// читает карточку счёта по номеру
func (c *Cards) Card(ctx context.Context, query views.CardQuery) (views.Card, error) {
	var card views.Card

	// поля запроса ложатся прямо в проекцию
	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
}

billing/invoice/infrastructure/broker/contracts/v1/events.go

package v1

// имена сообщений с версией на конце
const (
	InvoiceIssuedType = "billing.invoice.issued.v1"
	InvoicePaidType   = "billing.invoice.paid.v1"
)

// интеграционное событие: то, что видят соседи
type InvoiceIssued struct {
	Number     string `json:"number"`
	Customer   string `json:"customer"`
	TotalMinor int64  `json:"total_minor"`
}

// полей ровно столько, сколько нужно читателю
type InvoicePaid struct {
	Number string `json:"number"`
}

billing/invoice/infrastructure/broker/contracts/v1/envelope.go

package v1

import (
	"encoding/json"
	"time"
)

// конверт: метаданные сообщения отдельно от данных факта
type Envelope struct {
	ID         string          `json:"id"`
	Type       string          `json:"type"`
	OccurredAt time.Time       `json:"occurred_at"`
	// ключ порядка: сообщения одного счёта идут по порядку
	Key     string          `json:"key"`
	Payload json.RawMessage `json:"payload"`
}

billing/invoice/infrastructure/broker/publisher.go

package broker

import (
	"context"
	"encoding/json"
	"errors"
	"time"

	"github.com/google/uuid"

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

// порт записи: куда лечь строке сообщения
type Appender interface {
	Append(ctx context.Context, messageType string, key string, body []byte) error
}

// реализация порта публикации: переводит и отдаёт на запись
type Publisher struct {
	rows Appender
}

func NewPublisher(rows Appender) *Publisher {
	return &Publisher{rows: rows}
}

// по сообщению на факт, и ни одной отправки
func (p *Publisher) Publish(ctx context.Context, facts ...events.Event) error {
	for _, fact := range facts {
		message, err := Contract(fact, uuid.NewString(), time.Now().UTC())

		// факт без контракта наружу не уезжает
		if errors.Is(err, ErrInternalFact) {
			continue
		}

		if err != nil {
			return err
		}

		body, err := json.Marshal(message)
		if err != nil {
			return err
		}

		// строка ложится в ту же транзакцию, что и счёт
		if err := p.rows.Append(ctx, message.Type, message.Key, body); err != nil {
			return err
		}
	}

	return nil
}

billing/invoice/infrastructure/broker/outbound.go

package broker

import (
	"encoding/json"
	"errors"
	"time"

	v1 "billing/invoice/infrastructure/broker/contracts/v1"
	"billing/invoice/domains/invoice/events"
)

// факт, у которого нет контракта наружу
var ErrInternalFact = errors.New("broker: fact is not published outside")

// переводит доменный факт в сообщение
func Contract(fact events.Event, id string, at time.Time) (v1.Envelope, error) {
	var (
		messageType string
		key         string
		payload     any
	)

	switch event := fact.(type) {
	// явный разбор: что уезжает, решают здесь
	case events.Issued:
		messageType = v1.InvoiceIssuedType
		key = event.Number
		payload = v1.InvoiceIssued{
			Number:     event.Number,
			Customer:   event.Customer,
			TotalMinor: event.Total.Get(),
		}
	case events.Paid:
		messageType = v1.InvoicePaidType
		key = event.Number
		payload = v1.InvoicePaid{Number: event.Number}
	// новый факт наружу сам не уедет
	default:
		return v1.Envelope{}, ErrInternalFact
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return v1.Envelope{}, err
	}

	return v1.Envelope{
		ID:         id,
		Type:       messageType,
		OccurredAt: at,
		Key:        key,
		Payload:    body,
	}, nil
}

billing/invoice/applications/payment/transport/consumer/received.go

package consumer

import (
	"context"
	"encoding/json"
	"errors"

	v1 "billing/invoice/infrastructure/broker/contracts/v1"
	"billing/invoice/domains/invoice"
	"billing/invoice/applications/payment"
)

// схема соседа, как мы её читаем
type PaymentReceived struct {
	InvoiceNumber string `json:"invoice_number"`
	AmountMinor   int64  `json:"amount_minor"`
}

// порт: видели ли мы это сообщение раньше
type Inbox interface {
	Remember(ctx context.Context, id string) (bool, error)
}

// единица работы, память о сообщениях и сценарий
type Handler struct {
	uow   invoice.UnitOfWork
	inbox Inbox
	pay   *payment.UseCase
}

func NewHandler(uow invoice.UnitOfWork, inbox Inbox, pay *payment.UseCase) *Handler {
	return &Handler{uow: uow, inbox: inbox, pay: pay}
}

// разобрать, проверить повтор, позвать сценарий
func (h *Handler) Handle(ctx context.Context, message v1.Envelope) error {
	var received PaymentReceived
	if err := json.Unmarshal(message.Payload, &received); err != nil {
		return err
	}

	return h.uow.Do(ctx, func(ctx context.Context) error {
		// то же сообщение второй раз — ничего не делаем
		first, err := h.inbox.Remember(ctx, message.ID)
		if err != nil {
			return err
		}

		if !first {
			return nil
		}

		err = h.pay.Run(ctx, invoice.Number(received.InvoiceNumber))

		// повтор не ошибка: счёт уже в нужном состоянии
		if errors.Is(err, invoice.ErrAlreadyPaid) {
			return nil
		}

		return err
	})
}

billing/invoice/infrastructure/postgres/inbox.go

package postgres

import (
	"context"
	"database/sql"
)

// реализация: таблица обработанных сообщений
type Inbox struct {
	db *sql.DB
}

func NewInbox(db *sql.DB) *Inbox {
	return &Inbox{db: db}
}

// вставка по уникальному id в той же транзакции
func (i *Inbox) Remember(ctx context.Context, id string) (bool, error) {
	result, err := conn(ctx, i.db).ExecContext(ctx, insertHandled, id)
	if err != nil {
		return false, err
	}

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

	// вставилось — значит, сообщение новое
	return rows == 1, nil
}

Шаг 9 из 9

Что получилось

Щёлкни по файлу, чтобы прочитать.

infrastructure/broker/contracts/v1/events.go — что видят соседи.
infrastructure/broker/contracts/v1/envelope.go — конверт: id, тип, время, ключ.
infrastructure/broker/outbound.go — перевод факта в сообщение.
infrastructure/broker/publisher.go — публикация: переводит и отдаёт строку на запись.
infrastructure/postgres/outbox.go — одна строка в таблицу, и больше ничего.
applications/payment/transport/consumer/received.go — приём чужого сообщения.
infrastructure/postgres/inbox.go — память об обработанных.
cmd/api/wire.go — сборка: порт публикации теперь за broker.Publisher.

Домен не изменился ни на строку: у него по-прежнему свои факты и своё имя для каждого. Всё новое легло на границу — по обе её стороны.

billing/GLOSSARY.md

Словарь

Одно значение на слово внутри контекста. Код, события, API и человек на созвоне пишут его одинаково.

«Платёж» в трёх отделах

Отдел	Что он называет платежом	Как это называется здесь
Продукт	Покупка прошла, пользователь увидел галочку	Событие заказа. Здесь его нет
Процессинг	Деньги захолдированы, списания не было	Авторизация
Бухгалтерия	Деньги пришли выпиской через два дня	Поступление

Слова

Авторизация. Эмитент согласился держать сумму. Деньги не двигались, холд истекает сам.

Списание. Команда забрать захолдированное. С этого момента деньги ушли со счёта плательщика.

Слова, которых здесь нет

Покупка. Продуктовое событие: пользователь увидел галочку. Живёт в заказах, здесь ему делать нечего.

billing/invoice/domains/invoice/invoice.go

package invoice

import (
	"time"

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

// счёт: корень агрегата
type Invoice struct {
	number   Number
	customer CustomerID
	lines    []Line
	total    vo.Money
	dueOn    time.Time
	status   Status
	// номер состояния, которое прочитали из базы
	version  int
	facts    []events.Event
}

// создаёт счёт; версия остаётся нулевой — в базе его ещё нет
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,
		// копия строк: снаружи их не поменять
		lines:    append([]Line(nil), lines...),
		total:    total,
		dueOn:    dueOn,
		status:   StatusIssued,
	}

	// записывает факт; отправлять его — не дело агрегата
	invoice.facts = append(invoice.facts, events.Issued{
		Number:   string(number),
		Customer: string(customer),
		Total:    total,
	})

	return invoice, nil
}

// отдаёт накопленные факты наружу
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 }

// отдаёт версию; менять её снаружи нельзя
func (i *Invoice) Version() int { return i.version }

billing/invoice/domains/invoice/paid.go

package invoice

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

// отмечает счёт оплаченным; дважды оплатить нельзя
func (i *Invoice) MarkPaid() error {
	if i.status == StatusPaid {
		return ErrAlreadyPaid
	}

	i.status = StatusPaid

	// записывает факт; отправлять его — не дело агрегата
	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"
)

// приписывает пеню счёту; оплаченный счёт её не берёт
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)

	// записывает факт; отправлять его — не дело агрегата
	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"

// факт домена; имя у него внутреннее
type Event interface {
	FactName() string
}

// счёт выставлен
type Issued struct {
	Number   string
	Customer string
	Total    vo.Money
}

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

// счёт оплачен
type Paid struct {
	Number string
}

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

// счёт получил пеню
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"

// календарь рабочих дней: часть модели, а не порт к базе
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"
)

// правило: счёт, ставка и календарь — на выходе сумма
func Amount(
	overdue *invoice.Invoice,
	now time.Time,
	perWorkday vo.Money,
	calendar invoice.Calendar,
) vo.Money {
	// срок правило берёт у самого счёта, а не у вызывающего
	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"
	"errors"
)

// счёт изменили между чтением и записью
var ErrConflict = errors.New("invoice: invoice changed since it was loaded")

// порт: как счёт попадает в базу и возвращается оттуда
type Repository interface {
	ByNumber(ctx context.Context, number Number) (*Invoice, error)
	// вернёт ErrConflict, если счёт успели изменить
	Save(ctx context.Context, invoice *Invoice) error
}

billing/invoice/domains/invoice/publisher.go

package invoice

import (
	"context"

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

// порт: куда уезжают факты
type Publisher interface {
	Publish(ctx context.Context, facts ...events.Event) error
}

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

package vo

// сумма в копейках
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"
)

// контракт спецификации: один вопрос про один агрегат
type Specification interface {
	IsSatisfiedBy(inv *invoice.Invoice) bool
}

// предикат: этот счёт просрочен?
type Overdue struct {
	now      time.Time
	calendar invoice.Calendar
}

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

// оплаченный счёт просроченным не бывает
func (o Overdue) IsSatisfiedBy(inv *invoice.Invoice) bool {
	if inv.Status() == invoice.StatusPaid {
		return false
	}

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

// второй предикат: счёт больше лимита
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()
}

// сборка: требует все предикаты сразу
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"
)

// два порта, календарь и ставка — всё, что нужно сценарию
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,
	}
}

// достать, посчитать, применить, сохранить, рассказать
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
		}

		// сценарий спрашивает у правила, а не верит вызывающему
		if !rules.NewOverdue(now, u.calendar).IsSatisfiedBy(overdue) {
			return nil
		}

		// правило считает сумму по счёту, ставке и календарю
		amount := penalty.Amount(overdue, now, u.perWorkday, u.calendar)

		// приписать пеню счёту может только сам счёт
		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"
)

// зависимости — только порты домена
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}
}

// один аргумент: команда, и она же проверяет свою форму
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"
	"errors"

	"billing/invoice/domains/invoice"
)

// сколько раз пробовать при конфликте
const attempts = 3

// зависимости — только порты домена
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}
}

// повторяет попытку целиком, пока счёт меняет кто-то другой
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)

		// конфликт — не ошибка вызова: данные устарели
		if errors.Is(err, invoice.ErrConflict) {
			continue
		}

		return err
	}

	return err
}

// одна попытка: достать, применить, сохранить, рассказать
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"
)

// собирает счёт из того, что лежало в базе
func Load(
	number Number,
	customer CustomerID,
	lines []Line,
	total vo.Money,
	dueOn time.Time,
	status Status,
	// версия приезжает вместе с данными
	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"

// порт: выполнить работу целиком или не выполнять
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"
)

// реализация порта хранения на Postgres
type Invoices struct {
	db *sql.DB
}

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

// достаёт счёт вместе с версией строки
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
}

// пишет счёт, если его версия в базе всё ещё та
func (r *Invoices) Save(ctx context.Context, saved *invoice.Invoice) error {
	db := conn(ctx, r.db)

	// нулевая версия — счёт новый, его вставляют
	if saved.Version() == 0 {
		if _, err := db.ExecContext(ctx, insertInvoice, saved); err != nil {
			return err
		}

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

		return err
	}

	// условие по версии стоит в самом UPDATE
	result, err := db.ExecContext(ctx, updateInvoice, saved, saved.Version())
	if err != nil {
		return err
	}

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

	// ни одной строки — значит, версия в базе другая
	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{}

// реализация единицы работы: транзакция Postgres
type UnitOfWork struct {
	db *sql.DB
}

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

// вся работа внутри одной транзакции
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 {
		// ошибка внутри — откат целиком
		_ = tx.Rollback()

		return err
	}

	return tx.Commit()
}

// транзакция из ctx, если операция уже началась
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/uow.go

package postgres

import (
	"context"
	"database/sql"
)

type txKey struct{}

// реализация единицы работы: транзакция Postgres
type UnitOfWork struct {
	db *sql.DB
}

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

// вся работа внутри одной транзакции
func (u *UnitOfWork) Do(ctx context.Context, work func(ctx context.Context) error) error {
	// операция уже началась — выполняем внутри неё, без второй транзакции
	if _, ok := ctx.Value(txKey{}).(*sql.Tx); ok {
		return work(ctx)
	}

	tx, err := u.db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}

	if err := work(context.WithValue(ctx, txKey{}, tx)); err != nil {
		// ошибка внутри — откат целиком
		_ = tx.Rollback()

		return err
	}

	return tx.Commit()
}

// транзакция из ctx, если операция уже началась
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"
)

// реализация публикации: факты строками в базе
type Outbox struct {
	db *sql.DB
}

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

// пишет факты в ту же транзакцию
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
		}

		// имя факта и тело сообщения
		if _, err := db.ExecContext(ctx, insertOutbox, event.FactName(), body); err != nil {
			return err
		}
	}

	return nil
}

billing/invoice/infrastructure/postgres/outbox.go

package postgres

import (
	"context"
	"database/sql"
)

// реализация публикации: факты строками в базе
type Outbox struct {
	db *sql.DB
}

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

// пишет строку сообщения; про домен и контракт не знает
func (o *Outbox) Append(ctx context.Context, messageType string, key string, body []byte) error {
	// тип, ключ и тело — всё, что нужно отправителю
	_, err := conn(ctx, o.db).ExecContext(ctx, insertOutbox, messageType, key, body)

	return err
}

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

package dto

import (
	"errors"
	"time"

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

// тело запроса: то, в чём данные пересекают границу
type IssueRequest struct {
	Number   string    `json:"number"`
	Customer string    `json:"customer"`
	DueOn    time.Time `json:"due_on"`
	Lines    []Line    `json:"lines"`
}

// строка счёта так, как её присылают
type Line struct {
	AmountMinor int64 `json:"amount_minor"`
}

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

// проверяется форма запроса, а не правила предметной области
func (r IssueRequest) Validate() error {
	if r.Number == "" {
		return ErrEmptyNumber
	}

	return nil
}

// что из запроса нужно сценарию, уже в типах домена
type Issue struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

// переводит тело запроса в типы домена
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"
)

// ручка владеет сценарием и больше ничем
type Handler struct {
	issue *issuing.UseCase
}

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

// POST /invoices: разобрать, перевести, позвать
func (h *Handler) Issue(w http.ResponseWriter, r *http.Request) {
	var body dto.IssueRequest

	// тело не читается — это ошибка клиента
	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
	}

	// перевод в типы домена — дело пакета dto
	args := body.ToDomain()

	// транспорт собирает команду — дальше едет она, а не россыпь аргументов
	cmd := issuing.IssueInvoice{
		Number:   args.Number,
		Customer: args.Customer,
		Lines:    args.Lines,
		DueOn:    args.DueOn,
	}

	// сценарий получает доменные сущности, а не запрос
	err := h.issue.Handle(r.Context(), cmd)
	if err != nil {
		// ошибку домена переводит в код ответа отдельная функция
		fail(w, err)

		return
	}

	// 201 и пустое тело: номер клиент прислал сам
	w.WriteHeader(http.StatusCreated)
}

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

package http

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

	"billing/invoice/domains/invoice"
)

// переводит ошибку домена в код ответа
func fail(w http.ResponseWriter, err error) {
	switch {
	// правило домена нарушено — это ошибка запроса, а не сервера
	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:
		// причина остаётся в логе, наружу — код
		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"

// адрес ручки объявляет сам модуль
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"
)

// сборка уехала в сгенерированную функцию
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"
)

// набор адаптеров: один на все приложения сервиса
var adapters = wire.NewSet(
	postgres.NewUnitOfWork,
	postgres.NewInvoices,
	postgres.NewOutbox,
	// какая реализация стоит за портом
	wire.Bind(new(invoice.UnitOfWork), new(*postgres.UnitOfWork)),
	wire.Bind(new(invoice.Repository), new(*postgres.Invoices)),
	wire.Bind(new(invoice.Publisher), new(*postgres.Outbox)),
)

// что надо собрать — по типу результата
func initHandler(dsn string) (*issuinghttp.Handler, error) {
	wire.Build(openDB, adapters, issuing.New, issuinghttp.NewHandler)

	// тело-заглушка: настоящее напишет генератор
	return nil, nil
}

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

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/broker"
	"billing/invoice/infrastructure/postgres"
	"billing/invoice/applications/issuing"
	issuinghttp "billing/invoice/applications/issuing/transport/http"
)

// набор адаптеров: один на все приложения сервиса
var adapters = wire.NewSet(
	postgres.NewUnitOfWork,
	postgres.NewInvoices,
	postgres.NewOutbox,
	broker.NewPublisher,
	// какая реализация стоит за портом
	wire.Bind(new(invoice.UnitOfWork), new(*postgres.UnitOfWork)),
	wire.Bind(new(invoice.Repository), new(*postgres.Invoices)),
	// порт публикации теперь за брокером, а outbox — за его узким портом
	wire.Bind(new(invoice.Publisher), new(*broker.Publisher)),
	wire.Bind(new(broker.Appender), new(*postgres.Outbox)),
)

// что надо собрать — по типу результата
func initHandler(dsn string) (*issuinghttp.Handler, error) {
	wire.Build(openDB, adapters, issuing.New, issuinghttp.NewHandler)

	// тело-заглушка: настоящее напишет генератор
	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"
)

// команда: что сделать и с какими данными
type IssueInvoice struct {
	Number   invoice.Number
	Customer invoice.CustomerID
	Lines    []invoice.Line
	DueOn    time.Time
}

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

// проверяется форма, а не правила предметной области
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"
)

// карточка счёта для экрана, а не агрегат
type Card struct {
	Number  string
	Total   int64
	Status  string
	Overdue bool
}

// запрос: что просят прочитать, названное типом
type CardQuery struct {
	Number invoice.Number
}

// порт чтения: отдаёт готовую карточку
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"
)

// запрос отдаёт ровно то, что показывают на экране
const selectCard = `
SELECT number,
       total_minor,
       status,
       due_on < now() AND status <> 'paid' AS overdue
  FROM invoices
 WHERE number = $1`

// реализация порта чтения: один запрос, без агрегата
type Cards struct {
	db *sql.DB
}

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

// читает карточку счёта по номеру
func (c *Cards) Card(ctx context.Context, query views.CardQuery) (views.Card, error) {
	var card views.Card

	// поля запроса ложатся прямо в проекцию
	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
}

billing/invoice/infrastructure/broker/contracts/v1/events.go

package v1

// имена сообщений с версией на конце
const (
	InvoiceIssuedType = "billing.invoice.issued.v1"
	InvoicePaidType   = "billing.invoice.paid.v1"
)

// интеграционное событие: то, что видят соседи
type InvoiceIssued struct {
	Number     string `json:"number"`
	Customer   string `json:"customer"`
	TotalMinor int64  `json:"total_minor"`
}

// полей ровно столько, сколько нужно читателю
type InvoicePaid struct {
	Number string `json:"number"`
}

billing/invoice/infrastructure/broker/contracts/v1/envelope.go

package v1

import (
	"encoding/json"
	"time"
)

// конверт: метаданные сообщения отдельно от данных факта
type Envelope struct {
	ID         string          `json:"id"`
	Type       string          `json:"type"`
	OccurredAt time.Time       `json:"occurred_at"`
	// ключ порядка: сообщения одного счёта идут по порядку
	Key     string          `json:"key"`
	Payload json.RawMessage `json:"payload"`
}

billing/invoice/infrastructure/broker/publisher.go

package broker

import (
	"context"
	"encoding/json"
	"errors"
	"time"

	"github.com/google/uuid"

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

// порт записи: куда лечь строке сообщения
type Appender interface {
	Append(ctx context.Context, messageType string, key string, body []byte) error
}

// реализация порта публикации: переводит и отдаёт на запись
type Publisher struct {
	rows Appender
}

func NewPublisher(rows Appender) *Publisher {
	return &Publisher{rows: rows}
}

// по сообщению на факт, и ни одной отправки
func (p *Publisher) Publish(ctx context.Context, facts ...events.Event) error {
	for _, fact := range facts {
		message, err := Contract(fact, uuid.NewString(), time.Now().UTC())

		// факт без контракта наружу не уезжает
		if errors.Is(err, ErrInternalFact) {
			continue
		}

		if err != nil {
			return err
		}

		body, err := json.Marshal(message)
		if err != nil {
			return err
		}

		// строка ложится в ту же транзакцию, что и счёт
		if err := p.rows.Append(ctx, message.Type, message.Key, body); err != nil {
			return err
		}
	}

	return nil
}

billing/invoice/infrastructure/broker/outbound.go

package broker

import (
	"encoding/json"
	"errors"
	"time"

	v1 "billing/invoice/infrastructure/broker/contracts/v1"
	"billing/invoice/domains/invoice/events"
)

// факт, у которого нет контракта наружу
var ErrInternalFact = errors.New("broker: fact is not published outside")

// переводит доменный факт в сообщение
func Contract(fact events.Event, id string, at time.Time) (v1.Envelope, error) {
	var (
		messageType string
		key         string
		payload     any
	)

	switch event := fact.(type) {
	// явный разбор: что уезжает, решают здесь
	case events.Issued:
		messageType = v1.InvoiceIssuedType
		key = event.Number
		payload = v1.InvoiceIssued{
			Number:     event.Number,
			Customer:   event.Customer,
			TotalMinor: event.Total.Get(),
		}
	case events.Paid:
		messageType = v1.InvoicePaidType
		key = event.Number
		payload = v1.InvoicePaid{Number: event.Number}
	// новый факт наружу сам не уедет
	default:
		return v1.Envelope{}, ErrInternalFact
	}

	body, err := json.Marshal(payload)
	if err != nil {
		return v1.Envelope{}, err
	}

	return v1.Envelope{
		ID:         id,
		Type:       messageType,
		OccurredAt: at,
		Key:        key,
		Payload:    body,
	}, nil
}

billing/invoice/applications/payment/transport/consumer/received.go

package consumer

import (
	"context"
	"encoding/json"
	"errors"

	v1 "billing/invoice/infrastructure/broker/contracts/v1"
	"billing/invoice/domains/invoice"
	"billing/invoice/applications/payment"
)

// схема соседа, как мы её читаем
type PaymentReceived struct {
	InvoiceNumber string `json:"invoice_number"`
	AmountMinor   int64  `json:"amount_minor"`
}

// порт: видели ли мы это сообщение раньше
type Inbox interface {
	Remember(ctx context.Context, id string) (bool, error)
}

// единица работы, память о сообщениях и сценарий
type Handler struct {
	uow   invoice.UnitOfWork
	inbox Inbox
	pay   *payment.UseCase
}

func NewHandler(uow invoice.UnitOfWork, inbox Inbox, pay *payment.UseCase) *Handler {
	return &Handler{uow: uow, inbox: inbox, pay: pay}
}

// разобрать, проверить повтор, позвать сценарий
func (h *Handler) Handle(ctx context.Context, message v1.Envelope) error {
	var received PaymentReceived
	if err := json.Unmarshal(message.Payload, &received); err != nil {
		return err
	}

	return h.uow.Do(ctx, func(ctx context.Context) error {
		// то же сообщение второй раз — ничего не делаем
		first, err := h.inbox.Remember(ctx, message.ID)
		if err != nil {
			return err
		}

		if !first {
			return nil
		}

		err = h.pay.Run(ctx, invoice.Number(received.InvoiceNumber))

		// повтор не ошибка: счёт уже в нужном состоянии
		if errors.Is(err, invoice.ErrAlreadyPaid) {
			return nil
		}

		return err
	})
}

billing/invoice/infrastructure/postgres/inbox.go

package postgres

import (
	"context"
	"database/sql"
)

// реализация: таблица обработанных сообщений
type Inbox struct {
	db *sql.DB
}

func NewInbox(db *sql.DB) *Inbox {
	return &Inbox{db: db}
}

// вставка по уникальному id в той же транзакции
func (i *Inbox) Remember(ctx context.Context, id string) (bool, error) {
	result, err := conn(ctx, i.db).ExecContext(ctx, insertHandled, id)
	if err != nil {
		return false, err
	}

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

	// вставилось — значит, сообщение новое
	return rows == 1, nil
}

Листать шаги можно стрелками ← и →.