Skip to content

Commit bf021ca

Browse files
authored
feat(postgres): add TransactionAwareAggregateRepository for unit-of-work pattern (#369)
This PR adds `TransactionAwareAggregateRepository` to `postgres` package, for caller-owned transactions (e.g. Unit of Work pattern). `TransactionAwareAggregateRepository` implements `aggregate.Repository` without requiring a connection pool. Instead, it expects a `pgx.Tx` be injected into context, and requires specifying the function to fetch the `pgx.Tx` from context. ```go repository := postgres.NewTransactionAwareAggregateRepository( unitofwork.TxFromContext, AggregateType, aggregateSerde, messageSerde, postgres.WithAggregateTableName[ID, *Root]("aggregates"), ) ```
1 parent 09cfa5e commit bf021ca

24 files changed

Lines changed: 516 additions & 142 deletions

‎.golangci.yaml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ linters:
1919
- errname
2020
- errorlint
2121
- exhaustive
22-
- exhaustruct
22+
- exhaustruct_v5
2323
- exptostd
2424
- fatcontext
2525
- funlen
@@ -122,7 +122,7 @@ linters:
122122
rules:
123123
- path: _test\.go
124124
linters:
125-
- exhaustruct
125+
- exhaustruct_v5
126126
- funlen
127127
# NOTE: false positive to ignore in this file.
128128
- path: command/scenario_test\.go

‎README.md‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -179,6 +179,31 @@ In `eventually` there are a couple of implementation types:
179179
Pick this when you want snapshot-style reads but still need the event
180180
log for projections, auditing, or downstream consumers.
181181

182+
- **`postgres.TransactionAwareAggregateRepository`** uses a caller-owned
183+
PostgreSQL transaction for both snapshot reads and writes. Supply a function
184+
that retrieves a `pgx.Tx` from the operation's context:
185+
186+
```go
187+
userRepository := postgres.NewTransactionAwareAggregateRepository(
188+
unitofwork.TxFromContext, // func(context.Context) (pgx.Tx, bool)
189+
UserType,
190+
userSerde,
191+
messageSerde,
192+
)
193+
```
194+
195+
This repository requires a transaction for every `Get` and `Save`; a missing
196+
transaction returns `postgres.ErrTransactionRequired`. It needs no connection
197+
pool and never begins, commits, or rolls back transactions. Configure table
198+
names with the same `WithAggregateTableName`, `WithEventsTableName`, and
199+
`WithStreamsTableName` options used by `AggregateRepository`.
200+
201+
The unit of work must open a **Serializable** transaction, propagate it through
202+
context, and roll back if any operation fails. `Save` success means the snapshot
203+
and events are staged; only the unit-of-work commit makes them durable. After a
204+
failed or rolled-back save, discard affected aggregate instances and reload
205+
before retrying the command, because saving flushes their recorded events.
206+
182207
### CQRS with Commands and Queries
183208

184209
CQRS - or _Command/Query Responsibility Segregation_ - splits the write path from the read path.

‎command/command.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,7 +54,7 @@ func FromGenericEnvelope[T Command](cmd GenericEnvelope) (Envelope[T], bool) {
5454
}, true
5555
}
5656

57-
return Envelope[T]{}, false //nolint:exhaustruct // This is a zero value anyway.
57+
return Envelope[T]{}, false //nolint:exhaustruct_v5 // This is a zero value anyway.
5858
}
5959

6060
// ToEnvelope is a convenience function that wraps the provided Command type

‎examples/todolist/internal/todolist/add_item_command.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ type AddItemCommand struct {
2020
// Name implements message.Message.
2121
func (AddItemCommand) Name() string { return "AddTodoListItem" }
2222

23-
//nolint:exhaustruct // Interface implementation assertion.
23+
//nolint:exhaustruct_v5 // Interface implementation assertion.
2424
var _ command.Handler[AddItemCommand] = AddItemCommandHandler{}
2525

2626
// AddItemCommandHandler is the command.Handler for AddItemCommand commands.

‎examples/todolist/internal/todolist/create_command.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ type CreateCommand struct {
1818
// Name implements message.Message.
1919
func (CreateCommand) Name() string { return "CreateTodoList" }
2020

21-
//nolint:exhaustruct // Interface implementation assertion.
21+
//nolint:exhaustruct_v5 // Interface implementation assertion.
2222
var _ command.Handler[CreateCommand] = CreateCommandHandler{}
2323

2424
// CreateCommandHandler is the Command Handler for CreateCommand commands.

‎examples/todolist/internal/todolist/delete_item_command.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ type DeleteItemCommand struct {
1616
// Name implements message.Message.
1717
func (DeleteItemCommand) Name() string { return "DeleteTodoListItem" }
1818

19-
//nolint:exhaustruct // Interface implementation assertion.
19+
//nolint:exhaustruct_v5 // Interface implementation assertion.
2020
var _ command.Handler[DeleteItemCommand] = DeleteItemCommandHandler{}
2121

2222
// DeleteItemCommandHandler is the command.Handler for DeleteItemCommand

‎examples/todolist/internal/todolist/get_query.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ type GetQuery struct {
1717
// Name implements message.Message.
1818
func (GetQuery) Name() string { return "GetTodoList" }
1919

20-
//nolint:exhaustruct // Interface implementation assertion.
20+
//nolint:exhaustruct_v5 // Interface implementation assertion.
2121
var _ query.Handler[GetQuery, *TodoList] = GetQueryHandler{}
2222

2323
// GetQueryHandler handles a GetQuery by returning the TodoList specified

‎examples/todolist/internal/todolist/mark_item_as_done_command.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ type MarkItemAsDoneCommand struct {
1717
// Name implements message.Message.
1818
func (MarkItemAsDoneCommand) Name() string { return "MarkTodoListItemAsDone" }
1919

20-
//nolint:exhaustruct // Interface implementation assertion.
20+
//nolint:exhaustruct_v5 // Interface implementation assertion.
2121
var _ command.Handler[MarkItemAsDoneCommand] = MarkItemAsDoneCommandHandler{}
2222

2323
// MarkItemAsDoneCommandHandler is the command.Handler for

‎examples/todolist/internal/todolist/mark_item_as_pending_command.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,7 @@ type MarkItemAsPendingCommand struct {
1717
// Name implements message.Message.
1818
func (MarkItemAsPendingCommand) Name() string { return "MarkTodoListItemAsPending" }
1919

20-
//nolint:exhaustruct // Interface implementation assertion.
20+
//nolint:exhaustruct_v5 // Interface implementation assertion.
2121
var _ command.Handler[MarkItemAsPendingCommand] = MarkItemAsPendingCommandHandler{}
2222

2323
// MarkItemAsPendingCommandHandler is the command.Handler for

‎examples/todolist/internal/todolist/todolist.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ func (tl *TodoList) Apply(evt event.Event) error {
8080
tl.CreationTime = evt.CreationTime
8181

8282
case ItemWasAdded:
83-
item := &Item{} //nolint:exhaustruct // Applied below.
83+
item := &Item{} //nolint:exhaustruct_v5 // Applied below.
8484
if err := item.Apply(evt); err != nil {
8585
return fmt.Errorf("todolist.TodoList.Apply: failed to apply item event, %w", err)
8686
}

0 commit comments

Comments
 (0)