From c6934e129d68975db08ea07b49fea27fe51dfaad Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 07:10:40 +0000 Subject: [PATCH 1/2] feat: add transaction and write batch support MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds withTransaction and withBatch to FirestoreService, routed through fiber-local Context.References (CurrentTransaction/CurrentBatch) so all reads and writes — including those made through repositories — are transparently redirected to the active transaction or batch. - effect-firebase: extend FirestoreServiceShape, add Firestore.withTransaction / Firestore.withBatch helpers, noop layer - @effect-firebase/admin: transactions via db.runTransaction (typed failures roll back and propagate), batches via db.batch; queries run through transaction.get; streams and deleteRecursive die inside a transaction - @effect-firebase/client: same via runTransaction/writeBatch; queries also die inside a transaction (client SDK supports document reads only) - @effect-firebase/mock: pass-through defaults - Nested withTransaction/withBatch join the ambient one; withBatch inside a transaction routes writes to the transaction Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01LJB3ye9tLX17ar4rFSSLYi --- README.md | 24 + .../lib/firestore/firestore-service.spec.ts | 417 ++++++++++++++++++ .../src/lib/firestore/firestore-service.ts | 375 ++++++++++++---- .../lib/firestore/firestore-service.spec.ts | 344 +++++++++++++++ .../src/lib/firestore/firestore-service.ts | 403 ++++++++++++----- packages/effect-firebase/README.md | 36 ++ .../src/lib/firestore/firestore-service.ts | 62 ++- .../src/lib/firestore/firestore.ts | 3 + .../src/lib/firestore/noop-layer.ts | 2 + .../src/lib/firestore/transaction.spec.ts | 40 ++ .../src/lib/firestore/transaction.ts | 86 ++++ packages/mock/README.md | 2 +- .../src/lib/firestore/firestore-service.ts | 5 + 13 files changed, 1612 insertions(+), 187 deletions(-) create mode 100644 packages/admin/src/lib/firestore/firestore-service.spec.ts create mode 100644 packages/client/src/lib/firestore/firestore-service.spec.ts create mode 100644 packages/effect-firebase/src/lib/firestore/transaction.spec.ts create mode 100644 packages/effect-firebase/src/lib/firestore/transaction.ts diff --git a/README.md b/README.md index 62342745..d2c56164 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,30 @@ const program = Effect.gen(function* () { ); ``` +### Transactions and batches + +```typescript +import { Effect } from 'effect'; +import { Firestore } from 'effect-firebase'; + +// Atomic read-modify-write across repositories +Firestore.withTransaction( + Effect.gen(function* () { + const repo = yield* PostRepository; + const post = yield* repo.getById(postId); + yield* repo.update(postId, { status: 'published' }); + }) +); + +// Stage many writes and commit them atomically +Firestore.withBatch( + Effect.gen(function* () { + const repo = yield* PostRepository; + yield* Effect.forEach(ids, (id) => repo.update(id, { status: 'archived' })); + }) +); +``` + ### Cloud Function ```typescript diff --git a/packages/admin/src/lib/firestore/firestore-service.spec.ts b/packages/admin/src/lib/firestore/firestore-service.spec.ts new file mode 100644 index 00000000..5b87d06b --- /dev/null +++ b/packages/admin/src/lib/firestore/firestore-service.spec.ts @@ -0,0 +1,417 @@ +import { describe, expect, it } from 'vitest'; +import { Cause, Data, Effect, Exit, Result, Stream } from 'effect'; +import { FirestoreService } from 'effect-firebase'; +import type { Firestore } from 'firebase-admin/firestore'; +import { layerFromFirestore } from './firestore-service.js'; + +class TestError extends Data.TaggedError('TestError')<{ reason: string }> {} + +type Op = readonly [name: string, ...args: unknown[]]; + +const makeFakeDb = () => { + const state = { + directOps: [] as Op[], + txOps: [] as Op[], + batchOps: [] as Op[], + runTransactionCalls: 0, + batchesCreated: 0, + commits: 0, + }; + + const idOf = (path: string) => path.split('/').pop() as string; + + const fakeSnapshot = (path: string, data: Record) => ({ + exists: true, + id: idOf(path), + ref: { id: idOf(path), path }, + data: () => data, + }); + + const fakeDocRef = (path: string): Record => { + const ref: Record = { + id: idOf(path), + path, + withConverter: () => ref, + get: async () => { + state.directOps.push(['get', path]); + return fakeSnapshot(path, { title: 'direct' }); + }, + set: async (...args: unknown[]) => { + state.directOps.push(['set', path, ...args]); + }, + update: async (...args: unknown[]) => { + state.directOps.push(['update', path, ...args]); + }, + delete: async () => { + state.directOps.push(['delete', path]); + }, + }; + return ref; + }; + + const fakeCollection = (path: string): Record => { + const collection: Record = { + path, + withConverter: () => collection, + doc: () => fakeDocRef(`${path}/generated-id`), + add: async (data: unknown) => { + state.directOps.push(['add', path, data]); + return fakeDocRef(`${path}/added-id`); + }, + get: async () => { + state.directOps.push(['query', path]); + return { docs: [fakeSnapshot(`${path}/1`, { title: 'direct' })] }; + }, + }; + return collection; + }; + + const tx = { + get: async (refOrQuery: { path: string; doc?: unknown }) => { + // Collection refs (queries) have a doc factory, document refs do not. + if (typeof refOrQuery.doc === 'function') { + state.txOps.push(['query', refOrQuery.path]); + return { + docs: [fakeSnapshot(`${refOrQuery.path}/1`, { title: 'tx' })], + }; + } + state.txOps.push(['get', refOrQuery.path]); + return fakeSnapshot(refOrQuery.path, { title: 'tx' }); + }, + create: (ref: { path: string }, data: unknown) => { + state.txOps.push(['create', ref.path, data]); + }, + set: (ref: { path: string }, data: unknown, options: unknown) => { + state.txOps.push(['set', ref.path, data, options]); + }, + update: (ref: { path: string }, data: unknown) => { + state.txOps.push(['update', ref.path, data]); + }, + delete: (ref: { path: string }) => { + state.txOps.push(['delete', ref.path]); + }, + }; + + const makeBatch = () => { + state.batchesCreated += 1; + return { + create: (ref: { path: string }, data: unknown) => { + state.batchOps.push(['create', ref.path, data]); + }, + set: (ref: { path: string }, data: unknown, options: unknown) => { + state.batchOps.push(['set', ref.path, data, options]); + }, + update: (ref: { path: string }, data: unknown) => { + state.batchOps.push(['update', ref.path, data]); + }, + delete: (ref: { path: string }) => { + state.batchOps.push(['delete', ref.path]); + }, + commit: async () => { + state.commits += 1; + }, + }; + }; + + const db = { + doc: fakeDocRef, + collection: fakeCollection, + recursiveDelete: async (ref: { path: string }) => { + state.directOps.push(['recursiveDelete', ref.path]); + }, + runTransaction: async (fn: (tx: unknown) => Promise) => { + state.runTransactionCalls += 1; + return fn(tx); + }, + batch: makeBatch, + }; + + return { db: db as unknown as Firestore, state }; +}; + +const run = ( + db: Firestore, + effect: Effect.Effect +) => Effect.runPromise(effect.pipe(Effect.provide(layerFromFirestore(db)))); + +const runExit = ( + db: Firestore, + effect: Effect.Effect +) => Effect.runPromiseExit(effect.pipe(Effect.provide(layerFromFirestore(db)))); + +const withService = ( + f: (service: FirestoreService['Service']) => Effect.Effect +) => Effect.flatMap(FirestoreService, f); + +describe('FirestoreService (admin)', () => { + describe('withTransaction', () => { + it('routes reads and writes through the transaction', async () => { + const { db, state } = makeFakeDb(); + await run( + db, + withService((fs) => + fs.withTransaction( + Effect.gen(function* () { + yield* fs.get('posts/1'); + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.update('posts/2', { title: 'b' }); + yield* fs.delete('posts/3'); + }) + ) + ) + ); + + expect(state.runTransactionCalls).toBe(1); + expect(state.txOps.map((op) => op[0])).toEqual([ + 'get', + 'set', + 'update', + 'delete', + ]); + expect(state.directOps).toEqual([]); + }); + + it('routes add through transaction.create with a pre-allocated ref', async () => { + const { db, state } = makeFakeDb(); + const result = await run( + db, + withService((fs) => fs.withTransaction(fs.add('posts', { title: 'a' }))) + ); + + expect(result).toEqual({ + id: 'generated-id', + path: 'posts/generated-id', + }); + expect(state.txOps).toEqual([ + ['create', 'posts/generated-id', { title: 'a' }], + ]); + }); + + it('routes queries through transaction.get', async () => { + const { db, state } = makeFakeDb(); + const results = await run( + db, + withService((fs) => fs.withTransaction(fs.query('posts', []))) + ); + + expect(state.txOps).toEqual([['query', 'posts']]); + expect(results).toHaveLength(1); + expect(results[0][1]).toEqual({ title: 'tx' }); + }); + + it('returns the result of the effect', async () => { + const { db } = makeFakeDb(); + const result = await run( + db, + withService((fs) => fs.withTransaction(Effect.succeed(42))) + ); + expect(result).toBe(42); + }); + + it('propagates typed failures from the effect', async () => { + const { db, state } = makeFakeDb(); + const exit = await runExit( + db, + withService((fs) => + fs.withTransaction( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* new TestError({ reason: 'boom' }); + }) + ) + ) + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (Exit.isFailure(exit)) { + expect(Cause.hasFails(exit.cause)).toBe(true); + const failure = Cause.findFail(exit.cause); + expect(Result.getOrThrow(failure).error).toMatchObject({ + _tag: 'TestError', + reason: 'boom', + }); + } + expect(state.runTransactionCalls).toBe(1); + }); + + it('joins the ambient transaction when nested', async () => { + const { db, state } = makeFakeDb(); + await run( + db, + withService((fs) => + fs.withTransaction( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.withTransaction(fs.set('posts/2', { title: 'b' })); + }) + ) + ) + ); + + expect(state.runTransactionCalls).toBe(1); + expect(state.txOps.map((op) => op[1])).toEqual(['posts/1', 'posts/2']); + }); + + it('dies when streaming inside a transaction', async () => { + const { db } = makeFakeDb(); + const exit = await runExit( + db, + withService((fs) => + fs.withTransaction(Stream.runCollect(fs.streamDoc('posts/1'))) + ) + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (Exit.isFailure(exit)) { + expect(Cause.hasDies(exit.cause)).toBe(true); + } + }); + + it('dies when deleteRecursive is used inside a transaction', async () => { + const { db } = makeFakeDb(); + const exit = await runExit( + db, + withService((fs) => fs.withTransaction(fs.deleteRecursive('posts/1'))) + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (Exit.isFailure(exit)) { + expect(Cause.hasDies(exit.cause)).toBe(true); + } + }); + }); + + describe('withBatch', () => { + it('stages writes on the batch and commits once', async () => { + const { db, state } = makeFakeDb(); + await run( + db, + withService((fs) => + fs.withBatch( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.update('posts/2', { title: 'b' }); + yield* fs.delete('posts/3'); + yield* fs.add('posts', { title: 'c' }); + }) + ) + ) + ); + + expect(state.batchesCreated).toBe(1); + expect(state.commits).toBe(1); + expect(state.batchOps.map((op) => op[0])).toEqual([ + 'set', + 'update', + 'delete', + 'create', + ]); + expect(state.directOps).toEqual([]); + }); + + it('reads bypass the batch and hit the database directly', async () => { + const { db, state } = makeFakeDb(); + await run( + db, + withService((fs) => + fs.withBatch( + Effect.gen(function* () { + yield* fs.get('posts/1'); + yield* fs.query('posts', []); + }) + ) + ) + ); + + expect(state.directOps.map((op) => op[0])).toEqual(['get', 'query']); + expect(state.batchOps).toEqual([]); + }); + + it('does not commit when the effect fails', async () => { + const { db, state } = makeFakeDb(); + const exit = await runExit( + db, + withService((fs) => + fs.withBatch( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* new TestError({ reason: 'boom' }); + }) + ) + ) + ); + + expect(Exit.isFailure(exit)).toBe(true); + expect(state.commits).toBe(0); + }); + + it('joins the ambient batch when nested', async () => { + const { db, state } = makeFakeDb(); + await run( + db, + withService((fs) => + fs.withBatch( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.withBatch(fs.set('posts/2', { title: 'b' })); + }) + ) + ) + ); + + expect(state.batchesCreated).toBe(1); + expect(state.commits).toBe(1); + expect(state.batchOps).toHaveLength(2); + }); + + it('routes writes to the transaction when used inside withTransaction', async () => { + const { db, state } = makeFakeDb(); + await run( + db, + withService((fs) => + fs.withTransaction(fs.withBatch(fs.set('posts/1', { title: 'a' }))) + ) + ); + + expect(state.batchesCreated).toBe(0); + expect(state.txOps.map((op) => op[0])).toEqual(['set']); + }); + + it('dies when deleteRecursive is used inside a batch', async () => { + const { db } = makeFakeDb(); + const exit = await runExit( + db, + withService((fs) => fs.withBatch(fs.deleteRecursive('posts/1'))) + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (Exit.isFailure(exit)) { + expect(Cause.hasDies(exit.cause)).toBe(true); + } + }); + }); + + describe('outside a transaction or batch', () => { + it('reads and writes go directly to the database', async () => { + const { db, state } = makeFakeDb(); + await run( + db, + withService((fs) => + Effect.gen(function* () { + yield* fs.get('posts/1'); + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.delete('posts/2'); + }) + ) + ); + + expect(state.directOps.map((op) => op[0])).toEqual([ + 'get', + 'set', + 'delete', + ]); + expect(state.txOps).toEqual([]); + expect(state.batchOps).toEqual([]); + }); + }); +}); diff --git a/packages/admin/src/lib/firestore/firestore-service.ts b/packages/admin/src/lib/firestore/firestore-service.ts index ad51ebf5..0b869555 100644 --- a/packages/admin/src/lib/firestore/firestore-service.ts +++ b/packages/admin/src/lib/firestore/firestore-service.ts @@ -1,6 +1,8 @@ import { Cause, + Context, Effect, + Exit, Layer, Array as Arr, Option, @@ -16,13 +18,50 @@ import { import type { App as FirebaseAdminApp } from 'firebase-admin/app'; import type { Snapshot } from 'effect-firebase'; import { UnknownError } from 'effect/Cause'; -import { getFirestore, type Firestore } from 'firebase-admin/firestore'; +import { + getFirestore, + type Firestore, + type Transaction, + type WriteBatch, +} from 'firebase-admin/firestore'; import { App } from '../app.js'; import { firestoreDecode, makeConverter } from './converter.js'; import { buildQuery } from './query-builder.js'; const packSnapshot = makeSnapshotPacker(firestoreDecode); +/** + * Fiber-local reference to the currently active transaction. Reads and + * writes issued while it is set are routed through the transaction, so + * repositories participate without changes. + */ +const CurrentTransaction = Context.Reference>( + '@effect-firebase/admin/CurrentTransaction', + { defaultValue: () => Option.none() } +); + +/** + * Fiber-local reference to the currently active write batch. Writes issued + * while it is set are staged on the batch; reads bypass it. + */ +const CurrentBatch = Context.Reference>( + '@effect-firebase/admin/CurrentBatch', + { defaultValue: () => Option.none() } +); + +/** + * Carries a typed Exit across the `runTransaction` promise boundary, so a + * failing effect rolls the transaction back without losing its error type. + */ +class EffectFailure { + constructor(readonly exit: Exit.Exit) {} +} + +/** + * The write-staging surface shared by `Transaction` and `WriteBatch`. + */ +type StagedWriter = Pick; + const mapError = (error: unknown) => error instanceof Error ? FirestoreError.fromError(error) @@ -89,111 +128,275 @@ const getFirestoreFromApp = (app: FirebaseAdminApp): Firestore => { const make = (db: Firestore) => { const converter = makeConverter(db); + // Writes route through the active transaction first, then the active + // batch. Both stage writes through the same create/set/update/delete + // surface; the Transaction is cast because TypeScript cannot resolve + // overloads through the Transaction | WriteBatch union. + const currentWriter: Effect.Effect> = Effect.gen( + function* () { + const tx = yield* CurrentTransaction; + if (Option.isSome(tx)) { + return Option.some(tx.value as unknown as StagedWriter); + } + return yield* CurrentBatch; + } + ); + + const assertNoTransaction = (operation: string) => + Effect.flatMap(CurrentTransaction, (tx) => + Option.isSome(tx) + ? Effect.die( + new Error( + `FirestoreService.${operation} cannot be used inside withTransaction.` + ) + ) + : Effect.void + ); + + const assertNoWriter = (operation: string) => + Effect.flatMap(currentWriter, (writer) => + Option.isSome(writer) + ? Effect.die( + new Error( + `FirestoreService.${operation} cannot be used inside withTransaction or withBatch.` + ) + ) + : Effect.void + ); + + const streamDoc = ( + path: string, + options?: Parameters[1] + ) => + Stream.callback, FirestoreError>((queue) => + Effect.acquireRelease( + Effect.sync(() => { + const docRef = db.doc(path); + return docRef.onSnapshot( + (snapshot) => { + Queue.offerUnsafe(queue, packSnapshot(snapshot, options)); + }, + (error) => { + const mappedError = mapError(error); + if (mappedError._tag === 'FirestoreError') { + Queue.failCauseUnsafe(queue, Cause.fail(mappedError)); + } else { + Queue.failCauseUnsafe( + queue, + Cause.fail(FirestoreError.fromError(error as Error)) + ); + } + } + ); + }), + (unsubscribe) => Effect.sync(() => unsubscribe()) + ) + ); + + const streamQuery = ( + collectionPath: string, + constraints: Parameters[2], + options?: Parameters[1] + ) => + Stream.callback, FirestoreError>((queue) => + Effect.acquireRelease( + Effect.sync(() => { + const query = buildQuery(db, collectionPath, constraints); + return query.onSnapshot( + (snapshot) => { + const snapshots = Arr.filterMap(snapshot.docs, (doc) => + Result.fromOption(packSnapshot(doc, options), () => void 0) + ); + Queue.offerUnsafe(queue, snapshots); + }, + (error) => { + const mappedError = mapError(error); + if (mappedError._tag === 'FirestoreError') { + Queue.failCauseUnsafe(queue, Cause.fail(mappedError)); + } else { + Queue.failCauseUnsafe( + queue, + Cause.fail(FirestoreError.fromError(error as Error)) + ); + } + } + ); + }), + (unsubscribe) => Effect.sync(() => unsubscribe()) + ) + ); + return FirestoreService.of({ get: (path, options) => - Effect.tryPromise({ - try: () => db.doc(path).get(), - catch: (error) => mapError(error), - }).pipe(Effect.map((snapshot) => packSnapshot(snapshot, options))), + Effect.gen(function* () { + const tx = yield* CurrentTransaction; + const snapshot = yield* Effect.tryPromise({ + try: () => + Option.isSome(tx) ? tx.value.get(db.doc(path)) : db.doc(path).get(), + catch: (error) => mapError(error), + }); + return packSnapshot(snapshot, options); + }), add: (path, data) => - Effect.tryPromise({ - try: async () => { - const ref = await db - .collection(path) - .withConverter(converter) - .add(data); + Effect.gen(function* () { + const writer = yield* currentWriter; + if (Option.isSome(writer)) { + const ref = db.collection(path).withConverter(converter).doc(); + yield* Effect.try({ + try: () => writer.value.create(ref, data), + catch: (error) => mapError(error), + }); return { id: ref.id, path: ref.path }; - }, - catch: (error) => mapError(error), + } + return yield* Effect.tryPromise({ + try: async () => { + const ref = await db + .collection(path) + .withConverter(converter) + .add(data); + return { id: ref.id, path: ref.path }; + }, + catch: (error) => mapError(error), + }); }), set: (path, data, options) => - Effect.tryPromise({ - try: () => - db - .doc(path) - .withConverter(converter) - .set(data, options || {}), - catch: (error) => mapError(error), + Effect.gen(function* () { + const writer = yield* currentWriter; + const ref = db.doc(path).withConverter(converter); + if (Option.isSome(writer)) { + yield* Effect.try({ + try: () => writer.value.set(ref, data, options || {}), + catch: (error) => mapError(error), + }); + return; + } + yield* Effect.tryPromise({ + try: () => ref.set(data, options || {}), + catch: (error) => mapError(error), + }); }), update: (path, data) => - Effect.tryPromise({ - try: () => db.doc(path).update(converter.toFirestore(data)), - catch: (error) => mapError(error), + Effect.gen(function* () { + const writer = yield* currentWriter; + if (Option.isSome(writer)) { + yield* Effect.try({ + try: () => + writer.value.update(db.doc(path), converter.toFirestore(data)), + catch: (error) => mapError(error), + }); + return; + } + yield* Effect.tryPromise({ + try: () => db.doc(path).update(converter.toFirestore(data)), + catch: (error) => mapError(error), + }); }), delete: (path) => - Effect.tryPromise({ - try: () => db.doc(path).withConverter(converter).delete(), - catch: (error) => mapError(error), + Effect.gen(function* () { + const writer = yield* currentWriter; + const ref = db.doc(path).withConverter(converter); + if (Option.isSome(writer)) { + yield* Effect.try({ + try: () => void writer.value.delete(ref), + catch: (error) => mapError(error), + }); + return; + } + yield* Effect.tryPromise({ + try: () => ref.delete(), + catch: (error) => mapError(error), + }); }), deleteRecursive: (path) => - Effect.tryPromise({ - try: () => db.recursiveDelete(db.doc(path)), - catch: (error) => mapError(error), - }), + assertNoWriter('deleteRecursive').pipe( + Effect.flatMap(() => + Effect.tryPromise({ + try: () => db.recursiveDelete(db.doc(path)), + catch: (error) => mapError(error), + }) + ) + ), query: (collectionPath, constraints) => - Effect.tryPromise({ - try: async () => { - const query = buildQuery(db, collectionPath, constraints); - const snapshot = await query.get(); - return Arr.filterMap(snapshot.docs, (doc) => - Result.fromOption(packSnapshot(doc), () => void 0) - ); - }, - catch: (error) => mapError(error), + Effect.gen(function* () { + const tx = yield* CurrentTransaction; + const snapshot = yield* Effect.tryPromise({ + try: () => { + const query = buildQuery(db, collectionPath, constraints); + return Option.isSome(tx) ? tx.value.get(query) : query.get(); + }, + catch: (error) => mapError(error), + }); + return Arr.filterMap(snapshot.docs, (doc) => + Result.fromOption(packSnapshot(doc), () => void 0) + ); }), streamDoc: (path, options) => - Stream.callback, FirestoreError>((queue) => - Effect.acquireRelease( - Effect.sync(() => { - const docRef = db.doc(path); - return docRef.onSnapshot( - (snapshot) => { - Queue.offerUnsafe(queue, packSnapshot(snapshot, options)); - }, - (error) => { - const mappedError = mapError(error); - if (mappedError._tag === 'FirestoreError') { - Queue.failCauseUnsafe(queue, Cause.fail(mappedError)); - } else { - Queue.failCauseUnsafe( - queue, - Cause.fail(FirestoreError.fromError(error as Error)) - ); - } - } - ); - }), - (unsubscribe) => Effect.sync(() => unsubscribe()) + Stream.unwrap( + assertNoTransaction('streamDoc').pipe( + Effect.map(() => streamDoc(path, options)) ) ), streamQuery: (collectionPath, constraints, options) => - Stream.callback, FirestoreError>((queue) => - Effect.acquireRelease( - Effect.sync(() => { - const query = buildQuery(db, collectionPath, constraints); - return query.onSnapshot( - (snapshot) => { - const snapshots = Arr.filterMap(snapshot.docs, (doc) => - Result.fromOption(packSnapshot(doc, options), () => void 0) - ); - Queue.offerUnsafe(queue, snapshots); - }, - (error) => { - const mappedError = mapError(error); - if (mappedError._tag === 'FirestoreError') { - Queue.failCauseUnsafe(queue, Cause.fail(mappedError)); - } else { - Queue.failCauseUnsafe( - queue, - Cause.fail(FirestoreError.fromError(error as Error)) - ); - } - } - ); - }), - (unsubscribe) => Effect.sync(() => unsubscribe()) + Stream.unwrap( + assertNoTransaction('streamQuery').pipe( + Effect.map(() => streamQuery(collectionPath, constraints, options)) ) ), + withTransaction: (self: Effect.Effect) => + Effect.gen(function* () { + const ambient = yield* CurrentTransaction; + // Nested transactions join the ambient one. + if (Option.isSome(ambient)) { + return yield* self; + } + const context = yield* Effect.context(); + const exit = yield* Effect.tryPromise({ + try: (signal) => + db.runTransaction((tx) => + Effect.runPromiseExit( + self.pipe( + Effect.provideService(CurrentTransaction, Option.some(tx)), + Effect.provideContext(context) + ), + { signal } + ).then((exit) => { + if (Exit.isFailure(exit)) { + // Reject so Firestore rolls the transaction back. + throw new EffectFailure(exit); + } + return exit; + }) + ), + catch: (error) => + error instanceof EffectFailure ? error : mapError(error), + }).pipe( + Effect.catch((error) => + error instanceof EffectFailure + ? Effect.succeed(error.exit as Exit.Exit) + : Effect.fail(error) + ) + ); + return yield* exit; + }), + withBatch: (self: Effect.Effect) => + Effect.gen(function* () { + const tx = yield* CurrentTransaction; + const ambient = yield* CurrentBatch; + // Inside a transaction writes are already atomic; nested batches + // join the ambient one. + if (Option.isSome(tx) || Option.isSome(ambient)) { + return yield* self; + } + const batch = db.batch(); + const result = yield* self.pipe( + Effect.provideService(CurrentBatch, Option.some(batch)) + ); + yield* Effect.tryPromise({ + try: () => batch.commit(), + catch: (error) => mapError(error), + }); + return result; + }), }); }; diff --git a/packages/client/src/lib/firestore/firestore-service.spec.ts b/packages/client/src/lib/firestore/firestore-service.spec.ts new file mode 100644 index 00000000..1fa499fa --- /dev/null +++ b/packages/client/src/lib/firestore/firestore-service.spec.ts @@ -0,0 +1,344 @@ +import { describe, expect, it, beforeEach, vi } from 'vitest'; +import { Cause, Data, Effect, Exit, Result } from 'effect'; +import { FirestoreService } from 'effect-firebase'; +import type { Firestore } from 'firebase/firestore'; + +class TestError extends Data.TaggedError('TestError')<{ reason: string }> {} + +type Op = readonly [name: string, ...args: unknown[]]; + +const h = vi.hoisted(() => { + const state = { + directOps: [] as Op[], + txOps: [] as Op[], + batchOps: [] as Op[], + runTransactionCalls: 0, + batchesCreated: 0, + commits: 0, + }; + + const reset = () => { + state.directOps = []; + state.txOps = []; + state.batchOps = []; + state.runTransactionCalls = 0; + state.batchesCreated = 0; + state.commits = 0; + }; + + const idOf = (path: string) => path.split('/').pop() as string; + + const fakeSnapshot = (path: string, data: Record) => ({ + id: idOf(path), + ref: { id: idOf(path), path }, + data: () => data, + }); + + const fakeDocRef = (path: string): Record => { + const ref: Record = { + id: idOf(path), + path, + type: 'document', + withConverter: () => ref, + }; + return ref; + }; + + const fakeCollection = (path: string): Record => { + const col: Record = { + path, + type: 'collection', + withConverter: () => col, + }; + return col; + }; + + const tx = { + get: async (ref: { path: string }) => { + state.txOps.push(['get', ref.path]); + return fakeSnapshot(ref.path, { title: 'tx' }); + }, + set: (ref: { path: string }, data: unknown, options?: unknown) => { + state.txOps.push(['set', ref.path, data, options]); + }, + update: (ref: { path: string }, data: unknown) => { + state.txOps.push(['update', ref.path, data]); + }, + delete: (ref: { path: string }) => { + state.txOps.push(['delete', ref.path]); + }, + }; + + const makeBatch = () => { + state.batchesCreated += 1; + return { + set: (ref: { path: string }, data: unknown, options?: unknown) => { + state.batchOps.push(['set', ref.path, data, options]); + }, + update: (ref: { path: string }, data: unknown) => { + state.batchOps.push(['update', ref.path, data]); + }, + delete: (ref: { path: string }) => { + state.batchOps.push(['delete', ref.path]); + }, + commit: async () => { + state.commits += 1; + }, + }; + }; + + return { + state, + reset, + fakeSnapshot, + fakeDocRef, + fakeCollection, + tx, + makeBatch, + }; +}); + +vi.mock('firebase/firestore', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + doc: (dbOrCollection: { path?: string }, path?: string) => + path !== undefined + ? h.fakeDocRef(path) + : h.fakeDocRef(`${dbOrCollection.path}/generated-id`), + collection: (_db: unknown, path: string) => h.fakeCollection(path), + query: (ref: unknown) => ref, + getDoc: async (ref: { path: string }) => { + h.state.directOps.push(['get', ref.path]); + return h.fakeSnapshot(ref.path, { title: 'direct' }); + }, + getDocs: async (q: { path: string }) => { + h.state.directOps.push(['query', q.path]); + return { docs: [h.fakeSnapshot(`${q.path}/1`, { title: 'direct' })] }; + }, + addDoc: async (col: { path: string }, data: unknown) => { + h.state.directOps.push(['add', col.path, data]); + return h.fakeDocRef(`${col.path}/added-id`); + }, + setDoc: async (ref: { path: string }, data: unknown, options?: unknown) => { + h.state.directOps.push(['set', ref.path, data, options]); + }, + updateDoc: async (ref: { path: string }, data: unknown) => { + h.state.directOps.push(['update', ref.path, data]); + }, + deleteDoc: async (ref: { path: string }) => { + h.state.directOps.push(['delete', ref.path]); + }, + runTransaction: async ( + _db: unknown, + fn: (tx: unknown) => Promise + ) => { + h.state.runTransactionCalls += 1; + return fn(h.tx); + }, + writeBatch: () => h.makeBatch(), + }; +}); + +// Imported after the mock so the service uses the mocked SDK functions. +import { layerFromFirestore } from './firestore-service.js'; + +const db = {} as Firestore; + +const run = (effect: Effect.Effect) => + Effect.runPromise(effect.pipe(Effect.provide(layerFromFirestore(db)))); + +const runExit = (effect: Effect.Effect) => + Effect.runPromiseExit(effect.pipe(Effect.provide(layerFromFirestore(db)))); + +const withService = ( + f: (service: FirestoreService['Service']) => Effect.Effect +) => Effect.flatMap(FirestoreService, f); + +beforeEach(() => { + h.reset(); +}); + +describe('FirestoreService (client)', () => { + describe('withTransaction', () => { + it('routes reads and writes through the transaction', async () => { + await run( + withService((fs) => + fs.withTransaction( + Effect.gen(function* () { + yield* fs.get('posts/1'); + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.update('posts/2', { title: 'b' }); + yield* fs.delete('posts/3'); + }) + ) + ) + ); + + expect(h.state.runTransactionCalls).toBe(1); + expect(h.state.txOps.map((op) => op[0])).toEqual([ + 'get', + 'set', + 'update', + 'delete', + ]); + expect(h.state.directOps).toEqual([]); + }); + + it('routes add through transaction.set with a pre-allocated ref', async () => { + const result = await run( + withService((fs) => fs.withTransaction(fs.add('posts', { title: 'a' }))) + ); + + expect(result).toEqual({ + id: 'generated-id', + path: 'posts/generated-id', + }); + expect(h.state.txOps).toEqual([ + ['set', 'posts/generated-id', { title: 'a' }, undefined], + ]); + }); + + it('propagates typed failures from the effect', async () => { + const exit = await runExit( + withService((fs) => + fs.withTransaction( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* new TestError({ reason: 'boom' }); + }) + ) + ) + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (Exit.isFailure(exit)) { + expect(Cause.hasFails(exit.cause)).toBe(true); + const failure = Cause.findFail(exit.cause); + expect(Result.getOrThrow(failure).error).toMatchObject({ + _tag: 'TestError', + reason: 'boom', + }); + } + }); + + it('joins the ambient transaction when nested', async () => { + await run( + withService((fs) => + fs.withTransaction( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.withTransaction(fs.set('posts/2', { title: 'b' })); + }) + ) + ) + ); + + expect(h.state.runTransactionCalls).toBe(1); + expect(h.state.txOps.map((op) => op[1])).toEqual(['posts/1', 'posts/2']); + }); + + it('dies when querying inside a transaction', async () => { + const exit = await runExit( + withService((fs) => fs.withTransaction(fs.query('posts', []))) + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (Exit.isFailure(exit)) { + expect(Cause.hasDies(exit.cause)).toBe(true); + } + }); + }); + + describe('withBatch', () => { + it('stages writes on the batch and commits once', async () => { + await run( + withService((fs) => + fs.withBatch( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.update('posts/2', { title: 'b' }); + yield* fs.delete('posts/3'); + yield* fs.add('posts', { title: 'c' }); + }) + ) + ) + ); + + expect(h.state.batchesCreated).toBe(1); + expect(h.state.commits).toBe(1); + expect(h.state.batchOps.map((op) => op[0])).toEqual([ + 'set', + 'update', + 'delete', + 'set', + ]); + expect(h.state.directOps).toEqual([]); + }); + + it('reads bypass the batch and hit the database directly', async () => { + await run( + withService((fs) => + fs.withBatch( + Effect.gen(function* () { + yield* fs.get('posts/1'); + yield* fs.query('posts', []); + }) + ) + ) + ); + + expect(h.state.directOps.map((op) => op[0])).toEqual(['get', 'query']); + expect(h.state.batchOps).toEqual([]); + }); + + it('does not commit when the effect fails', async () => { + const exit = await runExit( + withService((fs) => + fs.withBatch( + Effect.gen(function* () { + yield* fs.set('posts/1', { title: 'a' }); + yield* new TestError({ reason: 'boom' }); + }) + ) + ) + ); + + expect(Exit.isFailure(exit)).toBe(true); + expect(h.state.commits).toBe(0); + }); + + it('routes writes to the transaction when used inside withTransaction', async () => { + await run( + withService((fs) => + fs.withTransaction(fs.withBatch(fs.set('posts/1', { title: 'a' }))) + ) + ); + + expect(h.state.batchesCreated).toBe(0); + expect(h.state.txOps.map((op) => op[0])).toEqual(['set']); + }); + }); + + describe('outside a transaction or batch', () => { + it('reads and writes go directly to the database', async () => { + await run( + withService((fs) => + Effect.gen(function* () { + yield* fs.get('posts/1'); + yield* fs.set('posts/1', { title: 'a' }); + yield* fs.delete('posts/2'); + }) + ) + ); + + expect(h.state.directOps.map((op) => op[0])).toEqual([ + 'get', + 'set', + 'delete', + ]); + expect(h.state.txOps).toEqual([]); + expect(h.state.batchOps).toEqual([]); + }); + }); +}); diff --git a/packages/client/src/lib/firestore/firestore-service.ts b/packages/client/src/lib/firestore/firestore-service.ts index 4ca15aff..4dc1a459 100644 --- a/packages/client/src/lib/firestore/firestore-service.ts +++ b/packages/client/src/lib/firestore/firestore-service.ts @@ -1,6 +1,8 @@ import { Cause, + Context, Effect, + Exit, Layer, Array as Arr, Option, @@ -15,6 +17,8 @@ import { doc, getFirestore, type Firestore, + type Transaction, + type WriteBatch, getDoc, getDocs, addDoc, @@ -23,6 +27,8 @@ import { updateDoc, deleteDoc, onSnapshot, + runTransaction, + writeBatch, } from 'firebase/firestore'; import { App, layer as appLayer } from '../app.js'; import { firestoreDecode, makeConverter } from './converter.js'; @@ -32,46 +38,232 @@ const dataOptions = (options?: FirestoreDataOptions) => ({ serverTimestamps: options?.serverTimestamps ?? 'estimate', }); +/** + * Fiber-local reference to the currently active transaction. Reads and + * writes issued while it is set are routed through the transaction, so + * repositories participate without changes. + */ +const CurrentTransaction = Context.Reference>( + '@effect-firebase/client/CurrentTransaction', + { defaultValue: () => Option.none() } +); + +/** + * Fiber-local reference to the currently active write batch. Writes issued + * while it is set are staged on the batch; reads bypass it. + */ +const CurrentBatch = Context.Reference>( + '@effect-firebase/client/CurrentBatch', + { defaultValue: () => Option.none() } +); + +/** + * Carries a typed Exit across the `runTransaction` promise boundary, so a + * failing effect rolls the transaction back without losing its error type. + */ +class EffectFailure { + constructor(readonly exit: Exit.Exit) {} +} + +/** + * The write-staging surface shared by `Transaction` and `WriteBatch`. + */ +type StagedWriter = Pick; + const make = (db: Firestore) => { const converter = makeConverter(db); + // Writes route through the active transaction first, then the active + // batch. Both stage writes through the same set/update/delete surface; + // the Transaction is cast because TypeScript cannot resolve overloads + // through the Transaction | WriteBatch union. + const currentWriter: Effect.Effect> = Effect.gen( + function* () { + const tx = yield* CurrentTransaction; + if (Option.isSome(tx)) { + return Option.some(tx.value as unknown as StagedWriter); + } + return yield* CurrentBatch; + } + ); + + const assertNoTransaction = (operation: string) => + Effect.flatMap(CurrentTransaction, (tx) => + Option.isSome(tx) + ? Effect.die( + new Error( + `FirestoreService.${operation} cannot be used inside withTransaction.` + ) + ) + : Effect.void + ); + + const packDocSnapshot = ( + snapshot: { + readonly id: string; + readonly ref: { readonly path: string }; + readonly data: (options?: { + readonly serverTimestamps?: 'estimate' | 'previous' | 'none'; + }) => Record | undefined; + }, + options?: FirestoreDataOptions + ): Option.Option => { + const data = snapshot.data(dataOptions(options)); + if (!data) return Option.none(); + return Option.some([ + { id: snapshot.id, path: snapshot.ref.path }, + firestoreDecode(data), + ]); + }; + + const streamDoc = (path: string, options?: FirestoreDataOptions) => + Stream.callback, FirestoreError>((queue) => + Effect.acquireRelease( + Effect.sync(() => { + const docRef = doc(db, path); + return onSnapshot( + docRef, + (snapshot) => { + const data = snapshot.data(dataOptions(options)); + if (!data) { + Queue.offerUnsafe(queue, Option.none()); + } else { + Queue.offerUnsafe( + queue, + Option.some([ + { id: snapshot.id, path: snapshot.ref.path }, + firestoreDecode(data), + ] as const) + ); + } + }, + (error) => { + Queue.failCauseUnsafe( + queue, + Cause.fail(FirestoreError.fromError(error)) + ); + } + ); + }), + (unsubscribe) => Effect.sync(() => unsubscribe()) + ) + ); + + const streamQuery = ( + collectionPath: string, + constraints: Parameters[2], + options?: FirestoreDataOptions + ) => + Stream.callback, FirestoreError>((queue) => + Effect.acquireRelease( + Effect.sync(() => { + const q = buildQuery(db, collectionPath, constraints); + return onSnapshot( + q, + (snapshot) => { + const snapshots = Arr.filterMap(snapshot.docs, (queryDoc) => { + const data = queryDoc.data(dataOptions(options)); + if (!data) return Result.failVoid; + return Result.succeed([ + { id: queryDoc.id, path: queryDoc.ref.path }, + firestoreDecode(data), + ] as const); + }); + Queue.offerUnsafe(queue, snapshots); + }, + (error) => { + Queue.failCauseUnsafe( + queue, + Cause.fail(FirestoreError.fromError(error)) + ); + } + ); + }), + (unsubscribe) => Effect.sync(() => unsubscribe()) + ) + ); + return FirestoreService.of({ get: (path, options) => - Effect.tryPromise({ - try: () => getDoc(doc(db, path)), - catch: (error) => FirestoreError.fromError(error), - }).pipe( - Effect.map((snapshot) => { - const data = snapshot.data(dataOptions(options)); - if (!data) return Option.none(); - return Option.some([ - { id: snapshot.id, path: snapshot.ref.path }, - firestoreDecode(data), - ]); - }) - ), + Effect.gen(function* () { + const tx = yield* CurrentTransaction; + const snapshot = yield* Effect.tryPromise({ + try: () => + Option.isSome(tx) + ? tx.value.get(doc(db, path)) + : getDoc(doc(db, path)), + catch: (error) => FirestoreError.fromError(error), + }); + return packDocSnapshot(snapshot, options); + }), add: (path, data) => - Effect.tryPromise({ - try: () => addDoc(collection(db, path).withConverter(converter), data), - catch: (error) => FirestoreError.fromError(error), - }).pipe(Effect.map((ref) => ({ id: ref.id, path: ref.path }))), + Effect.gen(function* () { + const writer = yield* currentWriter; + if (Option.isSome(writer)) { + const ref = doc(collection(db, path).withConverter(converter)); + yield* Effect.try({ + try: () => void writer.value.set(ref, data), + catch: (error) => FirestoreError.fromError(error), + }); + return { id: ref.id, path: ref.path }; + } + return yield* Effect.tryPromise({ + try: () => + addDoc(collection(db, path).withConverter(converter), data), + catch: (error) => FirestoreError.fromError(error), + }).pipe(Effect.map((ref) => ({ id: ref.id, path: ref.path }))); + }), set: (path, data, options) => - Effect.tryPromise({ - try: () => - setDoc(doc(db, path).withConverter(converter), data, { - merge: options?.merge, - }), - catch: (error) => FirestoreError.fromError(error), + Effect.gen(function* () { + const writer = yield* currentWriter; + const ref = doc(db, path).withConverter(converter); + if (Option.isSome(writer)) { + yield* Effect.try({ + try: () => + void writer.value.set(ref, data, { merge: options?.merge }), + catch: (error) => FirestoreError.fromError(error), + }); + return; + } + yield* Effect.tryPromise({ + try: () => setDoc(ref, data, { merge: options?.merge }), + catch: (error) => FirestoreError.fromError(error), + }); }), update: (path, data) => - Effect.tryPromise({ - try: () => updateDoc(doc(db, path), converter.toFirestore(data)), - catch: (error) => FirestoreError.fromError(error), + Effect.gen(function* () { + const writer = yield* currentWriter; + if (Option.isSome(writer)) { + yield* Effect.try({ + try: () => + void writer.value.update( + doc(db, path), + converter.toFirestore(data) + ), + catch: (error) => FirestoreError.fromError(error), + }); + return; + } + yield* Effect.tryPromise({ + try: () => updateDoc(doc(db, path), converter.toFirestore(data)), + catch: (error) => FirestoreError.fromError(error), + }); }), delete: (path) => - Effect.tryPromise({ - try: () => deleteDoc(doc(db, path).withConverter(converter)), - catch: (error) => FirestoreError.fromError(error), + Effect.gen(function* () { + const writer = yield* currentWriter; + const ref = doc(db, path).withConverter(converter); + if (Option.isSome(writer)) { + yield* Effect.try({ + try: () => void writer.value.delete(ref), + catch: (error) => FirestoreError.fromError(error), + }); + return; + } + yield* Effect.tryPromise({ + try: () => deleteDoc(ref), + catch: (error) => FirestoreError.fromError(error), + }); }), deleteRecursive: (_path) => Effect.die( @@ -80,82 +272,95 @@ const make = (db: Firestore) => { ) ), query: (collectionPath, constraints) => - Effect.tryPromise({ - try: async () => { - const q = buildQuery(db, collectionPath, constraints); - const snapshot = await getDocs(q); - return Arr.filterMap(snapshot.docs, (queryDoc) => { - const data = queryDoc.data(); - if (!data) return Result.failVoid; - return Result.succeed([ - { id: queryDoc.id, path: queryDoc.ref.path }, - firestoreDecode(data), - ] as const); - }); - }, - catch: (error) => FirestoreError.fromError(error), - }), + // The client SDK only supports document reads inside transactions. + assertNoTransaction('query').pipe( + Effect.flatMap(() => + Effect.tryPromise({ + try: async () => { + const q = buildQuery(db, collectionPath, constraints); + const snapshot = await getDocs(q); + return Arr.filterMap(snapshot.docs, (queryDoc) => { + const data = queryDoc.data(); + if (!data) return Result.failVoid; + return Result.succeed([ + { id: queryDoc.id, path: queryDoc.ref.path }, + firestoreDecode(data), + ] as const); + }); + }, + catch: (error) => FirestoreError.fromError(error), + }) + ) + ), streamDoc: (path, options) => - Stream.callback, FirestoreError>((queue) => - Effect.acquireRelease( - Effect.sync(() => { - const docRef = doc(db, path); - return onSnapshot( - docRef, - (snapshot) => { - const data = snapshot.data(dataOptions(options)); - if (!data) { - Queue.offerUnsafe(queue, Option.none()); - } else { - Queue.offerUnsafe( - queue, - Option.some([ - { id: snapshot.id, path: snapshot.ref.path }, - firestoreDecode(data), - ] as const) - ); - } - }, - (error) => { - Queue.failCauseUnsafe( - queue, - Cause.fail(FirestoreError.fromError(error)) - ); - } - ); - }), - (unsubscribe) => Effect.sync(() => unsubscribe()) + Stream.unwrap( + assertNoTransaction('streamDoc').pipe( + Effect.map(() => streamDoc(path, options)) ) ), streamQuery: (collectionPath, constraints, options) => - Stream.callback, FirestoreError>((queue) => - Effect.acquireRelease( - Effect.sync(() => { - const q = buildQuery(db, collectionPath, constraints); - return onSnapshot( - q, - (snapshot) => { - const snapshots = Arr.filterMap(snapshot.docs, (queryDoc) => { - const data = queryDoc.data(dataOptions(options)); - if (!data) return Result.failVoid; - return Result.succeed([ - { id: queryDoc.id, path: queryDoc.ref.path }, - firestoreDecode(data), - ] as const); - }); - Queue.offerUnsafe(queue, snapshots); - }, - (error) => { - Queue.failCauseUnsafe( - queue, - Cause.fail(FirestoreError.fromError(error)) - ); - } - ); - }), - (unsubscribe) => Effect.sync(() => unsubscribe()) + Stream.unwrap( + assertNoTransaction('streamQuery').pipe( + Effect.map(() => streamQuery(collectionPath, constraints, options)) ) ), + withTransaction: (self: Effect.Effect) => + Effect.gen(function* () { + const ambient = yield* CurrentTransaction; + // Nested transactions join the ambient one. + if (Option.isSome(ambient)) { + return yield* self; + } + const context = yield* Effect.context(); + const exit = yield* Effect.tryPromise({ + try: (signal) => + runTransaction(db, (tx) => + Effect.runPromiseExit( + self.pipe( + Effect.provideService(CurrentTransaction, Option.some(tx)), + Effect.provideContext(context) + ), + { signal } + ).then((exit) => { + if (Exit.isFailure(exit)) { + // Reject so Firestore rolls the transaction back. + throw new EffectFailure(exit); + } + return exit; + }) + ), + catch: (error) => + error instanceof EffectFailure + ? error + : FirestoreError.fromError(error), + }).pipe( + Effect.catch((error) => + error instanceof EffectFailure + ? Effect.succeed(error.exit as Exit.Exit) + : Effect.fail(error) + ) + ); + return yield* exit; + }), + withBatch: (self: Effect.Effect) => + Effect.gen(function* () { + const tx = yield* CurrentTransaction; + const ambient = yield* CurrentBatch; + // Inside a transaction writes are already atomic; nested batches + // join the ambient one. + if (Option.isSome(tx) || Option.isSome(ambient)) { + return yield* self; + } + const batch = writeBatch(db); + const result = yield* self.pipe( + Effect.provideService(CurrentBatch, Option.some(batch)) + ); + yield* Effect.tryPromise({ + try: () => batch.commit(), + catch: (error) => FirestoreError.fromError(error), + }); + return result; + }), }); }; diff --git a/packages/effect-firebase/README.md b/packages/effect-firebase/README.md index 6196e1c7..4e8ef2b4 100644 --- a/packages/effect-firebase/README.md +++ b/packages/effect-firebase/README.md @@ -105,6 +105,42 @@ Query.or( Fields and operators are validated at compile time against the model. +## Transactions and batches + +`Firestore.withTransaction` runs an effect inside a Firestore transaction. Every read and write performed by the effect — including through repositories — is routed through the transaction and committed atomically: + +```typescript +import { Effect } from 'effect'; +import { Firestore } from 'effect-firebase'; + +Firestore.withTransaction( + Effect.gen(function* () { + const repo = yield* PostRepository; + const post = yield* repo.getById(postId); // transactional read + // ... all reads must happen before the first write + yield* repo.update(postId, { likes: likes + 1 }); // transactional write + }) +); +``` + +- The SDK retries the transaction on contention, so the effect may run more than once. +- Firestore requires all transactional reads to happen before the first write. +- Nested `withTransaction` calls join the ambient transaction. +- `streamDoc`, `streamQuery`, and `deleteRecursive` cannot be used inside a transaction; the client SDK additionally disallows `query`. + +`Firestore.withBatch` stages writes on a write batch and commits them atomically when the effect succeeds. When the effect fails, nothing is committed: + +```typescript +Firestore.withBatch( + Effect.gen(function* () { + const repo = yield* PostRepository; + yield* Effect.forEach(ids, (id) => repo.update(id, { status: 'archived' })); + }) +); +``` + +Batches are write-only: reads inside the effect execute immediately against the database and do not see the staged writes. A batch supports at most 500 writes. + ## Schemas `FirestoreSchema` exports platform-agnostic schemas for Firestore types: diff --git a/packages/effect-firebase/src/lib/firestore/firestore-service.ts b/packages/effect-firebase/src/lib/firestore/firestore-service.ts index 965451a8..440ba422 100644 --- a/packages/effect-firebase/src/lib/firestore/firestore-service.ts +++ b/packages/effect-firebase/src/lib/firestore/firestore-service.ts @@ -115,6 +115,65 @@ type FirestoreStreaming = { ) => Stream.Stream, FirestoreError>; }; +type FirestoreTransactions = { + /** + * Run an effect inside a Firestore transaction. + * + * Every {@link FirestoreService} read and write performed by the effect — + * including reads and writes made through repositories — is routed through + * the transaction. The transaction commits when the effect succeeds and + * rolls back when it fails. + * + * Semantics and caveats: + * - The SDK may retry the transaction on contention, re-running the effect. + * The effect must therefore be safe to run more than once; non-Firestore + * side effects inside it (logging, HTTP calls, ...) can execute multiple + * times. + * - Firestore requires all transactional reads to happen before the first + * write. Violations surface as a {@link FirestoreError} at runtime. + * - Nested `withTransaction` calls join the ambient transaction instead of + * starting a new one. + * - `streamDoc`, `streamQuery`, and `deleteRecursive` cannot participate in + * a transaction and cause a defect (`Effect.die`) when used inside one. + * - With the client SDK, `query` is not supported inside a transaction + * (only document reads are) and causes a defect. + * - Forked fibers must not outlive the transaction; all transactional work + * has to complete before the effect finishes. + * + * @param self - The effect to run inside the transaction. + * @returns The result of the effect after the transaction has committed. + */ + readonly withTransaction: ( + self: Effect.Effect + ) => Effect.Effect; + + /** + * Run an effect inside a Firestore write batch. + * + * Every {@link FirestoreService} write performed by the effect — including + * writes made through repositories — is staged on the batch and committed + * atomically when the effect succeeds. When the effect fails, nothing is + * committed. + * + * Semantics and caveats: + * - Batches are write-only. Reads (`get`, `query`, streams) inside the + * effect execute immediately against the database and do not observe the + * staged writes. + * - A batch supports at most 500 write operations. + * - Nested `withBatch` calls join the ambient batch. Inside an ambient + * transaction, `withBatch` is a no-op wrapper: writes are already atomic + * through the transaction. + * - `deleteRecursive` cannot participate in a batch and causes a defect + * (`Effect.die`) when used inside one. + * + * @param self - The effect to run inside the batch. + * @returns The result of the effect after the batch has committed. + */ + readonly withBatch: ( + self: Effect.Effect + ) => Effect.Effect; +}; + export interface FirestoreDataOptions { /** * Controls how intermediate server timestamps are handled on the client when writing data. @@ -128,7 +187,8 @@ export interface FirestoreDataOptions { export type FirestoreServiceShape = FirestoreCRUD & FirestoreQuery & - FirestoreStreaming; + FirestoreStreaming & + FirestoreTransactions; export class FirestoreService extends Context.Service< FirestoreService, diff --git a/packages/effect-firebase/src/lib/firestore/firestore.ts b/packages/effect-firebase/src/lib/firestore/firestore.ts index 9cd2cb4c..1830c600 100644 --- a/packages/effect-firebase/src/lib/firestore/firestore.ts +++ b/packages/effect-firebase/src/lib/firestore/firestore.ts @@ -11,3 +11,6 @@ export * from './model/array.js'; // Repository factory export { makeRepository } from './model/repository.js'; + +// Transaction and batch helpers +export { withTransaction, withBatch } from './transaction.js'; diff --git a/packages/effect-firebase/src/lib/firestore/noop-layer.ts b/packages/effect-firebase/src/lib/firestore/noop-layer.ts index eb139d17..6894a07a 100644 --- a/packages/effect-firebase/src/lib/firestore/noop-layer.ts +++ b/packages/effect-firebase/src/lib/firestore/noop-layer.ts @@ -24,4 +24,6 @@ export const noopLayer = Layer.succeed(FirestoreService, { query: NotInitiallized, streamDoc: NotInitiallized, streamQuery: NotInitiallized, + withTransaction: NotInitiallized, + withBatch: NotInitiallized, }); diff --git a/packages/effect-firebase/src/lib/firestore/transaction.spec.ts b/packages/effect-firebase/src/lib/firestore/transaction.spec.ts new file mode 100644 index 00000000..6c39f9d4 --- /dev/null +++ b/packages/effect-firebase/src/lib/firestore/transaction.spec.ts @@ -0,0 +1,40 @@ +import { describe, expect, it, vi } from 'vitest'; +import { Effect, Layer } from 'effect'; +import { withTransaction, withBatch } from './transaction.js'; +import { FirestoreService } from './firestore-service.js'; +import type { FirestoreServiceShape } from './firestore-service.js'; + +const makeLayer = (overrides: Partial) => + Layer.succeed(FirestoreService, overrides as FirestoreServiceShape); + +describe('withTransaction', () => { + it('delegates to FirestoreService.withTransaction', async () => { + const withTransactionMock = vi.fn( + (self: Effect.Effect) => self + ); + const result = await Effect.runPromise( + withTransaction(Effect.succeed(42)).pipe( + Effect.provide(makeLayer({ withTransaction: withTransactionMock })) + ) + ); + + expect(result).toBe(42); + expect(withTransactionMock).toHaveBeenCalledTimes(1); + }); +}); + +describe('withBatch', () => { + it('delegates to FirestoreService.withBatch', async () => { + const withBatchMock = vi.fn( + (self: Effect.Effect) => self + ); + const result = await Effect.runPromise( + withBatch(Effect.succeed('ok')).pipe( + Effect.provide(makeLayer({ withBatch: withBatchMock })) + ) + ); + + expect(result).toBe('ok'); + expect(withBatchMock).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/effect-firebase/src/lib/firestore/transaction.ts b/packages/effect-firebase/src/lib/firestore/transaction.ts new file mode 100644 index 00000000..b7f5d36d --- /dev/null +++ b/packages/effect-firebase/src/lib/firestore/transaction.ts @@ -0,0 +1,86 @@ +import { Effect } from 'effect'; +import { UnknownError } from 'effect/Cause'; +import { FirestoreError } from './errors.js'; +import { FirestoreService } from './firestore-service.js'; + +/** + * Run an effect inside a Firestore transaction. + * + * Every {@link FirestoreService} read and write performed by the effect — + * including reads and writes made through repositories — is routed through + * the transaction. The transaction commits when the effect succeeds and + * rolls back when it fails. + * + * The SDK may retry the transaction on contention, re-running the effect, so + * the effect must be safe to run more than once. See + * `FirestoreService.withTransaction` for the full semantics. + * + * @param self - The effect to run inside the transaction. + * @returns The result of the effect after the transaction has committed. + * + * @example + * ```ts + * import { Effect } from 'effect'; + * import { Firestore } from 'effect-firebase'; + * import { AccountRepository } from './account-repository.js'; + * + * const transfer = (from: AccountId, to: AccountId, amount: number) => + * Firestore.withTransaction( + * Effect.gen(function* () { + * const repo = yield* AccountRepository; + * const source = yield* repo.getById(from); + * const target = yield* repo.getById(to); + * // ... all reads happen before the first write + * yield* repo.update(from, { balance: sourceBalance - amount }); + * yield* repo.update(to, { balance: targetBalance + amount }); + * }) + * ); + * ``` + */ +export const withTransaction = ( + self: Effect.Effect +): Effect.Effect => + Effect.gen(function* () { + const firestore = yield* FirestoreService; + return yield* firestore.withTransaction(self); + }); + +/** + * Run an effect inside a Firestore write batch. + * + * Every {@link FirestoreService} write performed by the effect — including + * writes made through repositories — is staged on the batch and committed + * atomically when the effect succeeds. When the effect fails, nothing is + * committed. + * + * Batches are write-only: reads inside the effect execute immediately against + * the database and do not observe the staged writes. See + * `FirestoreService.withBatch` for the full semantics. + * + * @param self - The effect to run inside the batch. + * @returns The result of the effect after the batch has committed. + * + * @example + * ```ts + * import { Effect } from 'effect'; + * import { Firestore } from 'effect-firebase'; + * import { PostRepository } from './post-repository.js'; + * + * const archiveAll = (ids: ReadonlyArray) => + * Firestore.withBatch( + * Effect.gen(function* () { + * const repo = yield* PostRepository; + * yield* Effect.forEach(ids, (id) => + * repo.update(id, { status: 'archived' }) + * ); + * }) + * ); + * ``` + */ +export const withBatch = ( + self: Effect.Effect +): Effect.Effect => + Effect.gen(function* () { + const firestore = yield* FirestoreService; + return yield* firestore.withBatch(self); + }); diff --git a/packages/mock/README.md b/packages/mock/README.md index 1316e178..a1da126d 100644 --- a/packages/mock/README.md +++ b/packages/mock/README.md @@ -67,7 +67,7 @@ await Effect.runPromise( - In-memory only — no persistence between process restarts - Queries are evaluated in-process — behaviour may differ from real Firestore for edge cases - No security rules evaluation -- No transaction support +- `withTransaction` and `withBatch` run the effect directly — no retries, no rollback, and no staged writes - No multi-client synchronization For tests that need full Firestore semantics, use the [Firebase Emulator Suite](https://firebase.google.com/docs/emulator-suite). diff --git a/packages/mock/src/lib/firestore/firestore-service.ts b/packages/mock/src/lib/firestore/firestore-service.ts index 401e7507..8414e8f4 100644 --- a/packages/mock/src/lib/firestore/firestore-service.ts +++ b/packages/mock/src/lib/firestore/firestore-service.ts @@ -37,5 +37,10 @@ export const MockFirestoreService = ( streamQuery: () => { throw new Error('MockFirestoreService.streamQuery not implemented.'); }, + // The mock has no concurrency or staging semantics, so transactions and + // batches simply run the effect: reads and writes hit the overridden + // methods directly. + withTransaction: (self) => self, + withBatch: (self) => self, ...overrides, }); From 5d0c81fc5ae3032c685f2c6d610851acd1d24099 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 19 Jul 2026 14:20:45 +0000 Subject: [PATCH 2/2] docs: fix undefined variables in withTransaction example Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01LJB3ye9tLX17ar4rFSSLYi --- .../effect-firebase/src/lib/firestore/transaction.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/effect-firebase/src/lib/firestore/transaction.ts b/packages/effect-firebase/src/lib/firestore/transaction.ts index b7f5d36d..774a7202 100644 --- a/packages/effect-firebase/src/lib/firestore/transaction.ts +++ b/packages/effect-firebase/src/lib/firestore/transaction.ts @@ -20,7 +20,7 @@ import { FirestoreService } from './firestore-service.js'; * * @example * ```ts - * import { Effect } from 'effect'; + * import { Effect, Option } from 'effect'; * import { Firestore } from 'effect-firebase'; * import { AccountRepository } from './account-repository.js'; * @@ -28,11 +28,11 @@ import { FirestoreService } from './firestore-service.js'; * Firestore.withTransaction( * Effect.gen(function* () { * const repo = yield* AccountRepository; - * const source = yield* repo.getById(from); - * const target = yield* repo.getById(to); + * const source = Option.getOrThrow(yield* repo.getById(from)); + * const target = Option.getOrThrow(yield* repo.getById(to)); * // ... all reads happen before the first write - * yield* repo.update(from, { balance: sourceBalance - amount }); - * yield* repo.update(to, { balance: targetBalance + amount }); + * yield* repo.update(from, { balance: source.balance - amount }); + * yield* repo.update(to, { balance: target.balance + amount }); * }) * ); * ```