(db: FumaDb, options: MakeFumaClientOptions = {})
| 292 | }; |
| 293 | |
| 294 | export const makeFumaClient = (db: FumaDb, options: MakeFumaClientOptions = {}): IFumaClient => { |
| 295 | const use: IFumaClient["use"] = (label, fn) => |
| 296 | Effect.flatMap(Effect.service(activeFumaDbRef), (active) => |
| 297 | fumaEffect(label, () => fn(makeSafeFumaQuery(active ?? db, options))), |
| 298 | ).pipe(Effect.withSpan(`fumadb.${label}`)); |
| 299 | |
| 300 | const transaction = <A, E>(effect: Effect.Effect<A, E>): Effect.Effect<A, E | StorageFailure> => |
| 301 | Effect.flatMap(Effect.service(activeFumaDbRef), (active) => { |
| 302 | if (active) return effect as Effect.Effect<unknown, unknown>; |
| 303 | |
| 304 | // The outermost transaction owns the post-commit hook queue; hooks |
| 305 | // queued anywhere inside (including nested pass-through transactions) |
| 306 | // run only after THIS commit, and are discarded on rollback. |
| 307 | const commitHooks: Array<Effect.Effect<void>> = []; |
| 308 | return Effect.tryPromise({ |
| 309 | try: () => |
| 310 | db.transaction(async (transactionDb) => { |
| 311 | const exit = await Effect.runPromiseExit( |
| 312 | effect.pipe( |
| 313 | Effect.provideService(activeFumaDbRef, transactionDb), |
| 314 | Effect.provideService(pendingCommitHooksRef, commitHooks), |
| 315 | ), |
| 316 | ); |
| 317 | if (Exit.isSuccess(exit)) return exit.value; |
| 318 | |
| 319 | const failure = exit.cause.reasons.find(Cause.isFailReason); |
| 320 | // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: FumaDB transactions roll back when the callback rejects |
| 321 | if (failure) throw new TransactionEffectFailure(failure.error); |
| 322 | // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: FumaDB transactions roll back when the callback rejects |
| 323 | throw new TransactionEffectDefect(exit.cause); |
| 324 | }), |
| 325 | catch: (cause): E | StorageFailure => { |
| 326 | if (cause instanceof TransactionEffectFailure) return cause.error as E; |
| 327 | if (cause instanceof TransactionEffectDefect) { |
| 328 | return fumaFailureFromCause("transaction", cause.cause); |
| 329 | } |
| 330 | return fumaFailureFromCause("transaction", cause); |
| 331 | }, |
| 332 | }).pipe( |
| 333 | Effect.tap(() => |
| 334 | Effect.forEach(commitHooks, (hook) => hook.pipe(Effect.ignoreCause({ log: false })), { |
| 335 | discard: true, |
| 336 | }), |
| 337 | ), |
| 338 | ); |
| 339 | }).pipe(Effect.withSpan("fumadb.transaction")) as Effect.Effect<A, E | StorageFailure>; |
| 340 | |
| 341 | return { use, transaction }; |
| 342 | }; |
| 343 | |
| 344 | export class FumaClient extends Context.Service<FumaClient, IFumaClient>()("executor/FumaClient") { |
| 345 | static layer = (db: FumaDb) => Layer.succeed(this)(makeFumaClient(db)); |
no outgoing calls
no test coverage detected