feat(database): afterCommit runs work after the outermost COMMIT

.changeset/after-commit.md

@@ -0,0 +1,5 @@
+---
+"@maple-dev/effect-orm": minor
+---
+
+`Db.afterCommit(effect)` queues work, such as an SMS or an HTTP call, to run after the outermost transaction commits, and never after a rollback. It delegates to Effect's `SqlClient.afterCommit` and, like `requireTransaction`, adds `Transaction` to the requirements, so queueing outside a transaction does not compile. A failing hook does not fail the committed transaction: its cause goes to the `ErrorReporter`s in context and to an error log, and the next hook still runs.

design/transactions.md

@@ -650,3 +650,22 @@ What landed differently from the sections above:
 - **Not yet done from phase 2:** the over-the-wire `@effect/sql-pg` suite (two connections,
   real serialization failures), and filing the upstream Effect issue for orDie'd COMMIT and
   ROLLBACK. The PGlite suite simulates contention with `RAISE ... USING ERRCODE`.
+
+## 10. `afterCommit` (issue #73)
+
+`Db.afterCommit(effect)` queues work that must wait for COMMIT: a vendor call, a job enqueue.
+Effect 4.0.6 ships `SqlClient.afterCommit`, which already runs callbacks after the outermost
+COMMIT, drops those of a rolled-back transaction or savepoint, hands a kept savepoint's callbacks
+to its parent, and dies when queued after the transaction ended. `Db.afterCommit` delegates to
+it on the client of the open `Db.transaction` and adds only:
+
+- **`Transaction` in `R`.** Effect's version runs the work at once outside a transaction; here
+  that does not compile, like `requireTransaction`.
+- **A failing hook does not fail the transaction.** The data is committed; a typed failure at
+  that point would invite a retry that writes twice. The cause goes to
+  `ErrorReporter.report` and `Effect.logError`, and the next hook still runs.
+- **The hook's context drops the open-state and `Transaction`** as Effect drops its own
+  transaction service, so a statement in a hook does not trip the closed-transaction guard.
+
+At most once: hooks are not persisted. Guaranteed delivery is an outbox table written inside
+the transaction, which needs no library support.

docs/database.md

@@ -89,6 +89,7 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti
 | `execute(statement)` | Run a statement and discard its rows |
 | `transaction(options?)` | Run an effect in a transaction; data-first or pipeable |
 | `requireTransaction` | Mark an effect as correct only inside a transaction |
+| `afterCommit(effect)` | Queue an effect to run after the outermost transaction commits, never after a rollback |
 | `retryContention(options?)` | Re-run on serialization failure or deadlock |
 | `Transaction` | The open transaction (`depth`, `isolationLevel`, `accessMode`) |
 | `isContention(error)` | Whether an error is a serialization failure or deadlock |
@@ -196,6 +197,47 @@ Effect's own `Effect.tx` treats `Effect.Transaction`. A helper that works either
 most reads, needs no marker. Inside a transaction, `yield* Db.Transaction` gives its `depth`
 and settings.
 
+### After commit
+
+A vendor call (an SMS, an HTTP request, a job enqueue) must not run inside a transaction: it
+holds the connection while it waits, and it cannot be rolled back. `afterCommit` queues the
+call and runs it only after COMMIT succeeds:
+
+```ts
+const bookTour = Effect.fn("bookTour")(function* (tour: NewTour) {
+	yield* Db.execute(Db.sql`INSERT INTO tours VALUES (${tour.id}, ${tour.phone})`)
+	yield* Db.afterCommit(sms.send(tour.phone, "Your tour is booked"))
+}, Db.requireTransaction)
+
+Db.transaction(bookTour(tour))  // the SMS goes out after COMMIT, or not at all
+```
+
+Like `requireTransaction`, `afterCommit` adds `Transaction` to the requirements, so a helper
+deep in the call stack can queue work without knowing where the transaction starts. The
+requirements of the queued effect stay in `R`: they must be met where it is queued.
+
+Effect's own `SqlClient.afterCommit` does the queueing; `afterCommit` adds the `Transaction`
+requirement and the failure policy below.
+
+- **Order.** Hooks run one at a time, in the order they were queued, after COMMIT and outside
+  the transaction. A statement in a hook runs on its own connection, not the transaction's.
+- **Rollback.** A rolled-back transaction runs nothing. A savepoint that rolls back drops what
+  it queued; a savepoint that is kept hands its hooks to the outer transaction. With `retry`,
+  only the attempt that commits runs its hooks.
+- **Context.** A hook runs with the services of the place it was queued, minus the
+  transaction.
+- **Failures.** A hook that fails or dies does not fail the transaction, because the data is
+  already committed and a caller that retries would write it twice. The hook's cause goes to
+  the `ErrorReporter`s in context (`Effect.withErrorReporting` uses the same ones) and to an
+  error log, and the next hook still runs. When the caller must know that the call failed,
+  handle the error inside the hook.
+- **At most once.** A crash before the hooks run skips them. Hooks run before `transaction`
+  returns and uninterruptibly, so keep them short and fork long-running work. When the work
+  must happen, write it to an outbox table inside the transaction instead.
+
+`afterCommit` dies when nothing would run the work: a `Transaction` provided by hand, or a
+transaction that has already ended (a fiber forked inside it and not joined).
+
 ### Settings
 
 | Option | Postgres |

docs/reference.md

@@ -512,7 +512,7 @@ placeholders, double-quoted identifiers.
 | `ClickHouseStatementFromString`, `splitTerminalClauses`, `maskLiteralsAndComments`               | `/sql`            |
 | Schema tooling: `isSchemaObject`, `makeSnapshot`, `renderSchema`, `diffSchemas`, `diffPgSchemas`, `fromDrizzleSnapshot`, snapshot and entity schemas (`PgEnumEntity`, `PgUniqueEntity`, …), `Hint`; for SQLite, `sqliteEntitiesOf`, `renderSqliteSchema`, `renderSqliteCreateTable`, `renderSqliteCreateIndex`, `diffSqliteSchemas`, `fromDrizzleSqliteSnapshot`, `SqliteMigrationOp`, `renderSqliteOp`, `renderSqliteMigrationSql`, `labelOfSqlite`, `rebuildName`, the `SqliteSnapshot` schema and the `SqliteTableEntity`, `SqliteColumnEntity`, `SqliteIndexEntity`, `SqliteUniqueEntity` schemas | `/schema`         |
 | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors                  | `/migrate`        |
-| `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `retryContention`, `Transaction` | `/database` |
+| `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` |
 | `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` |
 | `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` |
 | `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` |

src/database.ts

@@ -8,6 +8,7 @@
 export {
 	Database,
 	Transaction,
+	afterCommit,
 	execute,
 	fromSqlClient,
 	isContention,

src/database/after-commit.test.ts

@@ -0,0 +1,190 @@
+// `afterCommit` against a real Effect SqlClient (`@effect/sql-pglite`): work
+// that must wait for COMMIT, such as a vendor call, and must never run after a
+// rollback.
+
+import { PgliteClient } from "@effect/sql-pglite"
+import { assert, expect, layer } from "@effect/vitest"
+import { Cause, Context, Deferred, Effect, ErrorReporter, Exit, Fiber, Layer, Logger, Ref, Schedule } from "effect"
+import * as SqlClient from "effect/sql/SqlClient"
+import * as CH from "../ch/index"
+import * as PG from "../postgres"
+import { postgresDialect } from "../pg/dialect"
+import * as Db from "../database"
+
+const Live = Layer.effect(
+	Db.Database,
+	Effect.gen(function* () {
+		return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: postgresDialect })
+	}),
+).pipe(Layer.provideMerge(PgliteClient.layer({ postgresqlconf: "timezone = 'UTC'" })))
+
+let tables = 0
+const freshTable = Effect.gen(function* () {
+	const name = `after_commit_${++tables}`
+	yield* Db.execute(Db.sql`CREATE TABLE ${Db.sql.identifier(name)} (id int4 PRIMARY KEY)`)
+	return name
+})
+const insert = (table: string, id: number) => Db.execute(Db.sql`INSERT INTO ${Db.sql.identifier(table)} VALUES (${id})`)
+const ids = (table: string) =>
+	Db.run(CH.from(CH.table(table, { id: PG.int4 })).select("id").orderBy(["id", "asc"])).pipe(
+		Effect.map((rows) => rows.map((row) => row.id)),
+	)
+
+class Domain extends Error {
+	readonly _tag = "Domain"
+}
+
+class Sms extends Context.Service<Sms, { readonly sent: Array<string> }>()("test/Sms") {}
+
+const dies = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+	Effect.map(Effect.exit(effect), (exit) => {
+		assert(Exit.isFailure(exit) && Cause.hasDies(exit.cause))
+		return Cause.squash(exit.cause)
+	})
+
+layer(Live, { excludeTestServices: true })("afterCommit on PGlite", (it) => {
+	it.effect("runs after COMMIT, outside the transaction, and never after a rollback", () =>
+		Effect.gen(function* () {
+			const table = yield* freshTable
+			const seen = yield* Ref.make<ReadonlyArray<unknown>>([])
+			const record = (label: string) =>
+				Effect.gen(function* () {
+					const open = yield* Effect.serviceOption(Db.Transaction)
+					const rows = yield* ids(table)
+					yield* Ref.update(seen, (all) => [...all, { label, ids: rows, open: open._tag === "Some" }])
+				})
+
+			const value = yield* Db.transaction(
+				Effect.gen(function* () {
+					yield* insert(table, 1)
+					yield* Db.afterCommit(record("sms"))
+					expect(yield* Ref.get(seen)).toEqual([])
+					return "done"
+				}),
+			)
+			expect(value).toBe("done")
+			expect(yield* Ref.get(seen)).toEqual([{ label: "sms", ids: [1], open: false }])
+
+			const error = yield* Effect.flip(
+				Db.transaction(
+					insert(table, 2).pipe(
+						Effect.andThen(Db.afterCommit(record("rolled back"))),
+						Effect.andThen(Effect.fail(new Domain("no"))),
+					),
+				),
+			)
+			assert(error instanceof Domain)
+			expect(yield* Ref.get(seen)).toHaveLength(1)
+		}),
+	)
+
+	it.effect("runs in queue order; a rolled-back savepoint drops its hooks, a kept one hands them up", () =>
+		Effect.gen(function* () {
+			const ran = yield* Ref.make<ReadonlyArray<string>>([])
+			const queue = (label: string) => Db.afterCommit(Ref.update(ran, (all) => [...all, label]))
+			yield* Db.transaction(
+				Effect.gen(function* () {
+					yield* queue("outer 1")
+					yield* Db.transaction(queue("kept savepoint"))
+					yield* Db.transaction(queue("dropped savepoint").pipe(Effect.andThen(Effect.fail(new Domain("inner"))))).pipe(
+						Effect.ignore,
+					)
+					yield* Db.transaction(Db.transaction(queue("two deep")))
+					yield* queue("outer 2")
+					expect(yield* Ref.get(ran)).toEqual([])
+				}),
+			)
+			expect(yield* Ref.get(ran)).toEqual(["outer 1", "kept savepoint", "two deep", "outer 2"])
+
+			// A kept savepoint inside an outer transaction that rolls back runs nothing.
+			yield* Ref.set(ran, [])
+			yield* Db.transaction(Db.transaction(queue("inner")).pipe(Effect.andThen(Effect.fail(new Domain("outer"))))).pipe(
+				Effect.ignore,
+			)
+			expect(yield* Ref.get(ran)).toEqual([])
+		}),
+	)
+
+	it.effect("a failing hook does not fail the committed transaction: it is reported and logged, and the next hook runs", () =>
+		Effect.gen(function* () {
+			const table = yield* freshTable
+			const reported: Array<unknown> = []
+			const logged: Array<{ level: string; message: unknown; failure: unknown }> = []
+			const reporter = ErrorReporter.make(({ cause }) => void reported.push(Cause.squash(cause)))
+			const logger = Logger.make(({ logLevel, message, cause }) => {
+				logged.push({ level: logLevel, message, failure: Cause.squash(cause) })
+			})
+			const ran = yield* Ref.make(0)
+			const vendorDown = new Domain("vendor down")
+			const bug = new Error("bug in a hook")
+
+			const value = yield* Db.transaction(
+				Effect.gen(function* () {
+					yield* insert(table, 1)
+					yield* Db.afterCommit(Effect.fail(vendorDown))
+					yield* Db.afterCommit(Effect.die(bug))
+					yield* Db.afterCommit(Ref.update(ran, (n) => n + 1))
+					return "committed"
+				}),
+			).pipe(Effect.provide(Layer.merge(ErrorReporter.layer([reporter]), Logger.layer([logger]))))
+
+			expect(value).toBe("committed")
+			expect(yield* ids(table)).toEqual([1])
+			expect(yield* Ref.get(ran)).toBe(1)
+			expect(reported).toEqual([vendorDown, bug])
+			expect(logged.map((entry) => [entry.level, entry.failure])).toEqual([
+				["Error", vendorDown],
+				["Error", bug],
+			])
+		}),
+	)
+
+	it.effect("a hook sees the services of the place it was queued, and its R stays required", () =>
+		Effect.gen(function* () {
+			const sent: Array<string> = []
+			const sendSms = (text: string) => Effect.flatMap(Effect.service(Sms), (sms) => Effect.sync(() => void sms.sent.push(text)))
+			yield* Db.transaction(Db.afterCommit(sendSms("tour booked")).pipe(Effect.provideService(Sms, { sent })))
+			expect(sent).toEqual(["tour booked"])
+		}),
+	)
+
+	it.effect("a retried attempt drops its hooks; only the attempt that commits runs them", () =>
+		Effect.gen(function* () {
+			const ran = yield* Ref.make<ReadonlyArray<number>>([])
+			const attempts = yield* Ref.make(0)
+			yield* Db.transaction(
+				Effect.gen(function* () {
+					const attempt = yield* Ref.updateAndGet(attempts, (n) => n + 1)
+					yield* Db.afterCommit(Ref.update(ran, (all) => [...all, attempt]))
+					if (attempt < 3) yield* Db.execute({ sql: "DO $$ BEGIN RAISE EXCEPTION 'forced' USING ERRCODE = '40001'; END $$" })
+				}),
+				{ retry: { times: 3, schedule: Schedule.spaced("1 millis") } },
+			)
+			expect(yield* Ref.get(ran)).toEqual([3])
+		}),
+	)
+
+	it.effect("runs under a plain sql.withTransaction too; dies where nothing will run it", () =>
+		Effect.gen(function* () {
+			const sql = yield* SqlClient.SqlClient
+			const ran = yield* Ref.make(0)
+			const work = Ref.update(ran, (n) => n + 1)
+
+			// Effect's own transaction queues it: it runs once the outer COMMIT succeeds.
+			yield* sql.withTransaction(
+				Db.transaction(Db.afterCommit(work)).pipe(Effect.andThen(Effect.flatMap(Ref.get(ran), (n) => Effect.sync(() => expect(n).toBe(0))))),
+			)
+			expect(yield* Ref.get(ran)).toBe(1)
+
+			// Transaction provided by hand, with no transaction behind it.
+			yield* dies(Db.afterCommit(work).pipe(Effect.provideService(Db.Transaction, { depth: 0, isolationLevel: undefined, accessMode: undefined })))
+
+			// A fiber forked in the transaction that queues work after it ended.
+			const release = yield* Deferred.make<void>()
+			const straggler = yield* Db.transaction(Deferred.await(release).pipe(Effect.andThen(Db.afterCommit(work)), Effect.forkDetach))
+			yield* Deferred.succeed(release, undefined)
+			yield* dies(Fiber.join(straggler))
+			expect(yield* Ref.get(ran)).toBe(1)
+		}),
+	)
+})

src/database/database.test-d.ts

@@ -1,6 +1,6 @@
-// Type-level tests: rows from `run`, `requireTransaction`, and what `transaction` removes.
+// Type-level tests: rows from `run`, `requireTransaction`, `afterCommit`, and what `transaction` removes.
 
-import { Effect, Schema } from "effect"
+import { Context, Effect, Schema } from "effect"
 import { expectTypeOf } from "expect-type"
 import * as CH from "../ch/index"
 import * as PG from "../postgres"
@@ -58,6 +58,16 @@ expectTypeOf(Db.transaction(Effect.fail(new Domain()))).toEqualTypeOf<
 const unwrapped: Effect.Effect<number, Db.DatabaseError, Db.Database> = revoke("f")
 void unwrapped
 
+// `afterCommit` needs a transaction, swallows the hook's errors into its report,
+// and keeps the hook's own requirements: they must be met where it is queued.
+class Sms extends Context.Service<Sms, { readonly send: (text: string) => Effect.Effect<void, Domain> }>()("Sms") {}
+const sendLater = Db.afterCommit(Effect.flatMap(Effect.service(Sms), (sms) => sms.send("hi")))
+expectTypeOf(sendLater).toEqualTypeOf<Effect.Effect<void, never, Sms | Db.Transaction>>()
+expectTypeOf(Db.transaction(sendLater)).toEqualTypeOf<Effect.Effect<void, Db.DatabaseError | Db.TransactionError, Sms | Db.Database>>()
+// @ts-expect-error Transaction is still required
+const notQueued: Effect.Effect<void, never, Sms> = sendLater
+void notQueued
+
 // Settings are typed: an unknown isolation level does not compile.
 // @ts-expect-error not an isolation level
 void Db.transaction(Effect.void, { isolationLevel: "read uncommitted" })

src/database/database.ts

@@ -7,7 +7,7 @@
 // statements outliving their transaction, and contention retry. See
 // `design/transactions.md`.
 
-import { Cause, Context, Effect, Exit, Layer, Option, Schedule, Schema } from "effect"
+import { Cause, Context, Effect, ErrorReporter, Exit, Layer, Option, Schedule, Schema } from "effect"
 import { dual } from "effect/Function"
 import * as SqlClient from "effect/sql/SqlClient"
 import { isSqlError } from "effect/sql/SqlError"
@@ -528,6 +528,43 @@ export const execute = (statement: StatementInput): Effect.Effect<void, Database
 export const requireTransaction = <A, E, R>(self: Effect.Effect<A, E, R>): Effect.Effect<A, E, R | Transaction> =>
 	Effect.andThen(Effect.service(Transaction), self)
 
+/**
+ * Queue `effect` to run after the outermost transaction commits, and never if
+ * it rolls back: an SMS, an HTTP call, a job enqueue. Effect's own
+ * `SqlClient.afterCommit` does the queueing (savepoints, ordering, rollback);
+ * this adds `Transaction` to the requirements, like `requireTransaction`, so
+ * work queued outside a transaction does not compile instead of running at once.
+ *
+ * A hook that fails does not fail the committed transaction: its cause goes to
+ * the `ErrorReporter`s in context and to an error log. Hooks run at most once;
+ * work that must happen belongs in an outbox table written in the transaction.
+ *
+ * ```ts
+ * yield* Db.afterCommit(sms.send(phone, text))
+ * ```
+ */
+export const afterCommit = <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<void, never, R | Transaction> =>
+	Effect.gen(function* () {
+		yield* Transaction
+		const state = yield* Effect.serviceOption(TransactionOpen)
+		if (Option.isNone(state)) {
+			return yield* Effect.die(new Error("Db.afterCommit: no transaction opened by Db.transaction is open"))
+		}
+		yield* state.value.client.afterCommit(
+			effect.pipe(
+				Effect.catchCause((cause) =>
+					ErrorReporter.report(cause).pipe(
+						Effect.andThen(Effect.logError("afterCommit hook failed after the transaction committed", cause)),
+					),
+				),
+				Effect.asVoid,
+				// Effect drops its own transaction from the hook's context; drop ours too,
+				// so a statement in the hook does not trip the closed-transaction guard.
+				Effect.updateContext((context: Context.Context<R>) => Context.omit(TransactionOpen, Transaction)(context) as Context.Context<R>),
+			),
+		)
+	})
+
 /**
  * `transaction` on the `Database` in context. Data-first or pipeable, so it
  * also fits as an `Effect.fn` pipe argument: