(db: FumaDb, options: MakeFumaClientOptions = {})
| 347 | }; |
| 348 | |
| 349 | export const makeFumaClient = (db: FumaDb, options: MakeFumaClientOptions = {}): IFumaClient => { |
| 350 | const use: IFumaClient["use"] = (label, fn) => |
| 351 | Effect.flatMap(Effect.service(activeFumaDbRef), (active) => |
| 352 | fumaEffect(label, () => fn(makeSafeFumaQuery(active ?? db, options))), |
| 353 | ).pipe(Effect.withSpan(`fumadb.${label}`)); |
| 354 | |
| 355 | const transaction = <A, E>(effect: Effect.Effect<A, E>): Effect.Effect<A, E | StorageFailure> => |
| 356 | Effect.flatMap(Effect.service(activeFumaDbRef), (active) => { |
| 357 | if (active) return effect as Effect.Effect<unknown, unknown>; |
| 358 | |
| 359 | // The outermost transaction owns the post-commit hook queue; hooks |
| 360 | // queued anywhere inside (including nested pass-through transactions) |
| 361 | // run only after THIS commit, and are discarded on rollback. |
| 362 | const commitHooks: PendingCommitHook[] = []; |
| 363 | return Effect.contextWith((context) => |
| 364 | Effect.tryPromise({ |
| 365 | try: () => |
| 366 | db.transaction(async (transactionDb) => { |
| 367 | const exit = await Effect.runPromiseExitWith(context)( |
| 368 | effect.pipe( |
| 369 | Effect.provideService(activeFumaDbRef, transactionDb), |
| 370 | Effect.provideService(pendingCommitHooksRef, commitHooks), |
| 371 | ), |
| 372 | ); |
| 373 | if (Exit.isSuccess(exit)) return exit.value; |
| 374 | |
| 375 | const failure = exit.cause.reasons.find(Cause.isFailReason); |
| 376 | // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: FumaDB transactions roll back when the callback rejects |
| 377 | if (failure) throw new TransactionEffectFailure(failure.error); |
| 378 | // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: FumaDB transactions roll back when the callback rejects |
| 379 | throw new TransactionEffectDefect(exit.cause); |
| 380 | }), |
| 381 | catch: (cause): E | StorageFailure => { |
| 382 | if (cause instanceof TransactionEffectFailure) return cause.error as E; |
| 383 | if (cause instanceof TransactionEffectDefect) { |
| 384 | return fumaFailureFromCause("transaction", cause.cause); |
| 385 | } |
| 386 | return fumaFailureFromCause("transaction", cause); |
| 387 | }, |
| 388 | }).pipe(Effect.tap(() => runCommitHooks(commitHooks))), |
| 389 | ); |
| 390 | }).pipe(Effect.withSpan("fumadb.transaction")) as Effect.Effect<A, E | StorageFailure>; |
| 391 | |
| 392 | return { use, transaction }; |
| 393 | }; |
| 394 | |
| 395 | export class FumaClient extends Context.Service<FumaClient, IFumaClient>()("executor/FumaClient") { |
| 396 | static layer = (db: FumaDb) => Layer.succeed(this)(makeFumaClient(db)); |
no outgoing calls
no test coverage detected