feat(postgres): add TransactionAwareAggregateRepository for unit-of-work pattern

.golangci.yaml

@@ -19,7 +19,7 @@ linters:
     - errname
     - errorlint
     - exhaustive
-    - exhaustruct
+    - exhaustruct_v5
     - exptostd
     - fatcontext
     - funlen
@@ -122,7 +122,7 @@ linters:
     rules:
       - path: _test\.go
         linters:
-          - exhaustruct
+          - exhaustruct_v5
           - funlen
       # NOTE: false positive to ignore in this file.
       - path: command/scenario_test\.go

README.md

@@ -179,6 +179,31 @@ In `eventually` there are a couple of implementation types:
   Pick this when you want snapshot-style reads but still need the event
   log for projections, auditing, or downstream consumers.
 
+- **`postgres.TransactionAwareAggregateRepository`** uses a caller-owned
+  PostgreSQL transaction for both snapshot reads and writes. Supply a function
+  that retrieves a `pgx.Tx` from the operation's context:
+
+  ```go
+  userRepository := postgres.NewTransactionAwareAggregateRepository(
+      unitofwork.TxFromContext, // func(context.Context) (pgx.Tx, bool)
+      UserType,
+      userSerde,
+      messageSerde,
+  )
+  ```
+
+  This repository requires a transaction for every `Get` and `Save`; a missing
+  transaction returns `postgres.ErrTransactionRequired`. It needs no connection
+  pool and never begins, commits, or rolls back transactions. Configure table
+  names with the same `WithAggregateTableName`, `WithEventsTableName`, and
+  `WithStreamsTableName` options used by `AggregateRepository`.
+
+  The unit of work must open a **Serializable** transaction, propagate it through
+  context, and roll back if any operation fails. `Save` success means the snapshot
+  and events are staged; only the unit-of-work commit makes them durable. After a
+  failed or rolled-back save, discard affected aggregate instances and reload
+  before retrying the command, because saving flushes their recorded events.
+
 ### CQRS with Commands and Queries
 
 CQRS - or _Command/Query Responsibility Segregation_ - splits the write path from the read path.

command/command.go

@@ -54,7 +54,7 @@ func FromGenericEnvelope[T Command](cmd GenericEnvelope) (Envelope[T], bool) {
 		}, true
 	}
 
-	return Envelope[T]{}, false //nolint:exhaustruct // This is a zero value anyway.
+	return Envelope[T]{}, false //nolint:exhaustruct_v5 // This is a zero value anyway.
 }
 
 // ToEnvelope is a convenience function that wraps the provided Command type

examples/todolist/internal/todolist/add_item_command.go

@@ -20,7 +20,7 @@ type AddItemCommand struct {
 // Name implements message.Message.
 func (AddItemCommand) Name() string { return "AddTodoListItem" }
 
-//nolint:exhaustruct // Interface implementation assertion.
+//nolint:exhaustruct_v5 // Interface implementation assertion.
 var _ command.Handler[AddItemCommand] = AddItemCommandHandler{}
 
 // AddItemCommandHandler is the command.Handler for AddItemCommand commands.

examples/todolist/internal/todolist/create_command.go

@@ -18,7 +18,7 @@ type CreateCommand struct {
 // Name implements message.Message.
 func (CreateCommand) Name() string { return "CreateTodoList" }
 
-//nolint:exhaustruct // Interface implementation assertion.
+//nolint:exhaustruct_v5 // Interface implementation assertion.
 var _ command.Handler[CreateCommand] = CreateCommandHandler{}
 
 // CreateCommandHandler is the Command Handler for CreateCommand commands.

examples/todolist/internal/todolist/delete_item_command.go

@@ -16,7 +16,7 @@ type DeleteItemCommand struct {
 // Name implements message.Message.
 func (DeleteItemCommand) Name() string { return "DeleteTodoListItem" }
 
-//nolint:exhaustruct // Interface implementation assertion.
+//nolint:exhaustruct_v5 // Interface implementation assertion.
 var _ command.Handler[DeleteItemCommand] = DeleteItemCommandHandler{}
 
 // DeleteItemCommandHandler is the command.Handler for DeleteItemCommand

examples/todolist/internal/todolist/get_query.go

@@ -17,7 +17,7 @@ type GetQuery struct {
 // Name implements message.Message.
 func (GetQuery) Name() string { return "GetTodoList" }
 
-//nolint:exhaustruct // Interface implementation assertion.
+//nolint:exhaustruct_v5 // Interface implementation assertion.
 var _ query.Handler[GetQuery, *TodoList] = GetQueryHandler{}
 
 // GetQueryHandler handles a GetQuery by returning the TodoList specified

examples/todolist/internal/todolist/mark_item_as_done_command.go

@@ -17,7 +17,7 @@ type MarkItemAsDoneCommand struct {
 // Name implements message.Message.
 func (MarkItemAsDoneCommand) Name() string { return "MarkTodoListItemAsDone" }
 
-//nolint:exhaustruct // Interface implementation assertion.
+//nolint:exhaustruct_v5 // Interface implementation assertion.
 var _ command.Handler[MarkItemAsDoneCommand] = MarkItemAsDoneCommandHandler{}
 
 // MarkItemAsDoneCommandHandler is the command.Handler for

examples/todolist/internal/todolist/mark_item_as_pending_command.go

@@ -17,7 +17,7 @@ type MarkItemAsPendingCommand struct {
 // Name implements message.Message.
 func (MarkItemAsPendingCommand) Name() string { return "MarkTodoListItemAsPending" }
 
-//nolint:exhaustruct // Interface implementation assertion.
+//nolint:exhaustruct_v5 // Interface implementation assertion.
 var _ command.Handler[MarkItemAsPendingCommand] = MarkItemAsPendingCommandHandler{}
 
 // MarkItemAsPendingCommandHandler is the command.Handler for

examples/todolist/internal/todolist/todolist.go

@@ -80,7 +80,7 @@ func (tl *TodoList) Apply(evt event.Event) error {
 		tl.CreationTime = evt.CreationTime
 
 	case ItemWasAdded:
-		item := &Item{} //nolint:exhaustruct // Applied below.
+		item := &Item{} //nolint:exhaustruct_v5 // Applied below.
 		if err := item.Apply(evt); err != nil {
 			return fmt.Errorf("todolist.TodoList.Apply: failed to apply item event, %w", err)
 		}

examples/todolist/main.go

@@ -100,7 +100,7 @@ func run() error { //nolint:funlen // Single linear wire-up of the service; spli
 		connectgrpcreflect.NewStaticReflector(todolistv1connect.TodoListServiceName),
 	))
 
-	srv := &http.Server{ //nolint:exhaustruct // Stdlib struct with many optional fields; defaults are fine.
+	srv := &http.Server{ //nolint:exhaustruct_v5 // Stdlib struct with many optional fields; defaults are fine.
 		Addr:              cfg.Server.Address,
 		Handler:           mux,
 		ReadTimeout:       cfg.Server.ReadTimeout,

flake.nix

@@ -13,8 +13,8 @@
             config.allowUnfree = true;
           };
 
-          go = pkgs.go_1_26;
-          withOurGoVersion = pkg: pkg.override { buildGoModule = pkgs.buildGo126Module; };
+          go = pkgs.go_1_27;
+          withOurGoVersion = pkg: pkg.override { buildGoModule = pkgs.buildGo127Module; };
         in
         {
           devShells.default = with pkgs; mkShell {