import { Cause, Context, Data, Effect, Exit, Layer, Predicate } from "effect";import type { AbstractQuery } from "@executor-js/fumadb/query";import type { AnySchema, AnyTable, Schema as FumaSchema } from "@executor-js/fumadb/schema"; export class StorageError extends Data.TaggedError("StorageError")<{  readonly message: string;  readonly cause: unknown;}> {} export class UniqueViolationError extends Data.TaggedError("UniqueViolationError")<{  readonly model?: string;}> {} /** * The database connection itself failed — the statement never got a verdict. * Distinct from `StorageError` (a query the backend answered with an error) * because the two need different responses: a connection fault is either a * transient socket loss worth retrying on a FRESH pool, or a pool-lifetime * bug that must stay loud. * * `retryable` says which. It is a property of the fault, not a policy: *   - `true` — the socket died underneath a live pool, or was never *     established because the backend was unreachable (`CONNECTION_CLOSED`, *     `CONNECT_TIMEOUT`, `ECONNREFUSED`, `ECONNRESET`). Reconnecting can *     succeed. *   - `false` — the pool was already torn down or the socket belongs to a *     different request context (`CONNECTION_ENDED`, `CONNECTION_DESTROYED`, *     the workerd cross-request I/O rejection). Retrying is futile by *     construction — postgres.js rejects every query once `end()` has been *     called — so these must surface rather than be papered over. * * No retry consumes this yet; classification lands first so the retry seam * (and the request-scope fix behind these faults) can be designed against a * typed signal instead of driver strings. */export class StorageConnectionError extends Data.TaggedError("StorageConnectionError")<{  readonly message: string;  readonly label: string;  readonly code: string;  readonly retryable: boolean;  readonly cause: unknown;}> {} export type StorageFailure = StorageError | StorageConnectionError | UniqueViolationError; export type FumaTables = Record<string, AnyTable>;type EmptyFumaSchema = FumaSchema<"latest", Record<never, never>>;export type TablesToFumaSchema<TTables extends FumaTables | undefined> = TTables extends FumaTables  ? string extends keyof TTables    ? AnySchema    : FumaSchema<"latest", TTables>  : EmptyFumaSchema;export type FumaDb<TSchema extends AnySchema = AnySchema> = AbstractQuery<TSchema>;export type FumaQuery<TSchema extends AnySchema = AnySchema> = Omit<  AbstractQuery<TSchema>,  "internal" | "withContext" | "transaction"> & {  readonly transaction: <A>(run: (db: FumaQuery<TSchema>) => Promise<A>) => Promise<A>;};export type FumaRow<TTable extends AnyTable> = Omit<  {    readonly [K in keyof TTable["columns"]]: TTable["columns"][K]["$out"];  },  "row_id">; const isUniqueViolation = (cause: unknown): boolean => {  let current = cause;  for (let i = 0; i < 5; i += 1) {    const err =      current && typeof current === "object" ? (current as Record<string, unknown>) : null;    if (!err) return false;    const code = err["code"];    // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: database drivers expose unique-violation details on native error messages    const message = err["message"];    const innerCause = err["cause"];    if (code === "23505") return true;    if (      typeof message === "string" &&      /unique constraint|duplicate key|violates unique constraint/i.test(message)    ) {      return true;    }    if (!innerCause || innerCause === current) return false;    current = innerCause;  }  return false;}; /** * postgres.js raises connection faults through `Errors.connection(code, …)`, * which puts the code on `error.code` (see `postgres/src/errors.js`). Workerd's * cross-request I/O rejection is a plain `Error` with no code, so it is matched * on its fixed runtime text and given a synthetic code. */const RETRYABLE_CONNECTION_CODES: ReadonlySet<string> = new Set([  "CONNECTION_CLOSED",  "CONNECT_TIMEOUT",  "ECONNREFUSED",  "ECONNRESET",]);const FATAL_CONNECTION_CODES: ReadonlySet<string> = new Set([  "CONNECTION_ENDED",  "CONNECTION_DESTROYED",]);const CROSS_REQUEST_IO_CODE = "CROSS_REQUEST_IO";const CROSS_REQUEST_IO_PATTERN = /cannot perform i\/o on behalf of a different request/i; /** Walk a cause chain, newest first, at most 5 links deep (as `isUniqueViolation` does). */const walkCauses = (cause: unknown, visit: (err: Record<string, unknown>) => boolean): boolean => {  let current = cause;  for (let i = 0; i < 5; i += 1) {    const err =      current && typeof current === "object" ? (current as Record<string, unknown>) : null;    if (!err) return false;    if (visit(err)) return true;    const innerCause = err["cause"];    if (!innerCause || innerCause === current) return false;    current = innerCause;  }  return false;}; /** First driver error code in the cause chain, if any. Never the error text. */const causeCode = (cause: unknown): string | undefined => {  let found: string | undefined;  walkCauses(cause, (err) => {    const code = err["code"];    if (typeof code === "string" && code.length > 0) {      found = code;      return true;    }    return false;  });  return found;}; /** Connection-fault code in the cause chain, if this is a connection fault at all. */const connectionFaultCode = (cause: unknown): string | undefined => {  let found: string | undefined;  walkCauses(cause, (err) => {    const code = err["code"];    if (      typeof code === "string" &&      (RETRYABLE_CONNECTION_CODES.has(code) || FATAL_CONNECTION_CODES.has(code))    ) {      found = code;      return true;    }    // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: workerd's cross-request I/O rejection carries no code, only this fixed message    const message = err["message"];    if (typeof message === "string" && CROSS_REQUEST_IO_PATTERN.test(message)) {      found = CROSS_REQUEST_IO_CODE;      return true;    }    return false;  });  return found;}; /** * Build the failure message from stable inputs only — the call-site label and * the driver's error code — never the driver's text. * * The driver text is `Failed query: <sql>\nparams: <bound values>` * (`drizzle-orm/errors.js`). Copying it into the message made error reporting * group by statement shape, splitting one defect across a report per table and * WHERE-clause, and printed bound parameters (organization ids, user ids, * user-chosen connection names) into report titles. The full driver error is * preserved on `cause`, which the reporter still receives as a chained * exception. */const stableMessage = (label: string, code: string | undefined): string =>  code ? `FumaDB ${label} failed: ${code}` : `FumaDB ${label} failed`; export const isStorageFailure = (error: unknown): error is StorageFailure =>  Predicate.isTagged(error, "StorageError") ||  Predicate.isTagged(error, "StorageConnectionError") ||  Predicate.isTagged(error, "UniqueViolationError"); export const fumaFailureFromCause = (label: string, cause: unknown): StorageFailure => {  if (isStorageFailure(cause)) return cause;  if (isUniqueViolation(cause)) return new UniqueViolationError({ model: label });  const connectionCode = connectionFaultCode(cause);  if (connectionCode !== undefined) {    return new StorageConnectionError({      message: stableMessage(label, connectionCode),      label,      code: connectionCode,      retryable: RETRYABLE_CONNECTION_CODES.has(connectionCode),      cause,    });  }  return new StorageError({    message: stableMessage(label, causeCode(cause)),    cause,  });}; export const fumaEffect = <A>(  label: string,  run: () => Promise<A>,): Effect.Effect<A, StorageFailure> =>  Effect.tryPromise({    try: run,    catch: (cause) => fumaFailureFromCause(label, cause),  }); export const activeFumaDbRef = Context.Reference<FumaDb | null>("executor/ActiveFumaDb", {  defaultValue: () => null,}); // Post-commit hooks. `transaction()` nests by pass-through (an inner// `transaction()` call inside an active transaction just runs its effect), so// an effect that must observe only DURABLE changes cannot simply run "after// the transaction" — after an inner pass-through the outer transaction may// still roll back. `afterCommit` solves this structurally: while a transaction// is active the effect is queued on the outermost transaction's hook list and// runs after its commit; with no active transaction it runs immediately.// Hooks are best-effort observers: failures and defects are swallowed, and a// rolled-back transaction discards its queue.const pendingCommitHooksRef = Context.Reference<Array<Effect.Effect<void>> | null>(  "executor/PendingCommitHooks",  { defaultValue: () => null },); export const afterCommit = (effect: Effect.Effect<void>): Effect.Effect<void> =>  Effect.flatMap(Effect.service(pendingCommitHooksRef), (hooks) =>    hooks      ? Effect.sync(() => {          hooks.push(effect);        })      : effect.pipe(Effect.ignoreCause({ log: false })),  ); class TransactionEffectFailure {  constructor(readonly error: unknown) {}} class TransactionEffectDefect {  constructor(readonly cause: unknown) {}} export type IFumaClient<TSchema extends AnySchema = AnySchema> = Readonly<{  use: <A>(    label: string,    fn: (db: FumaQuery<TSchema>) => Promise<A>,  ) => Effect.Effect<A, StorageFailure>;  transaction: <A, E>(effect: Effect.Effect<A, E>) => Effect.Effect<A, E | StorageFailure>;}>; export interface MakeFumaClientOptions {  readonly tables?: ReadonlySet<string>;} const isAllowedTable = (tables: ReadonlySet<string> | undefined, table: PropertyKey): boolean =>  tables === undefined || (typeof table === "string" && tables.has(table)); const assertAllowedTable = (tables: ReadonlySet<string> | undefined, table: PropertyKey): void => {  if (isAllowedTable(tables, table)) return;  // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: plugin-facing FumaDB facade rejects unavailable tables synchronously before query execution  throw new StorageError({    message: `FumaDB table "${String(table)}" is not available through this storage boundary.`,    cause: undefined,  });}; const makeSafeFumaQuery = <TSchema extends AnySchema>(  db: FumaDb<TSchema>,  options: MakeFumaClientOptions,): FumaQuery<TSchema> => {  const table = <TableName extends keyof TSchema["tables"]>(name: TableName): TableName => {    assertAllowedTable(options.tables, name);    return name;  };   const query: FumaQuery<TSchema> = {    count: (name, value) => db.count(table(name), value),    create: (name, value) => db.create(table(name), value),    createMany: (name, values) => db.createMany(table(name), values),    deleteMany: (name, value) => db.deleteMany(table(name), value),    findFirst: (name, value) => db.findFirst(table(name), value),    findMany: (name, value) => db.findMany(table(name), value),    transaction: (run) =>      db.transaction((transactionDb) => run(makeSafeFumaQuery(transactionDb, options))),    updateMany: (name, value) => db.updateMany(table(name), value),    upsert: (name, value) => db.upsert(table(name), value),    upsertMany: (name, value) => db.upsertMany(table(name), value),  };   return Object.freeze(query);}; export const makeFumaClient = (db: FumaDb, options: MakeFumaClientOptions = {}): IFumaClient => {  const use: IFumaClient["use"] = (label, fn) =>    Effect.flatMap(Effect.service(activeFumaDbRef), (active) =>      fumaEffect(label, () => fn(makeSafeFumaQuery(active ?? db, options))),    ).pipe(Effect.withSpan(`fumadb.${label}`));   const transaction = <A, E>(effect: Effect.Effect<A, E>): Effect.Effect<A, E | StorageFailure> =>    Effect.flatMap(Effect.service(activeFumaDbRef), (active) => {      if (active) return effect as Effect.Effect<unknown, unknown>;       // The outermost transaction owns the post-commit hook queue; hooks      // queued anywhere inside (including nested pass-through transactions)      // run only after THIS commit, and are discarded on rollback.      const commitHooks: Array<Effect.Effect<void>> = [];      return Effect.tryPromise({        try: () =>          db.transaction(async (transactionDb) => {            const exit = await Effect.runPromiseExit(              effect.pipe(                Effect.provideService(activeFumaDbRef, transactionDb),                Effect.provideService(pendingCommitHooksRef, commitHooks),              ),            );            if (Exit.isSuccess(exit)) return exit.value;             const failure = exit.cause.reasons.find(Cause.isFailReason);            // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: FumaDB transactions roll back when the callback rejects            if (failure) throw new TransactionEffectFailure(failure.error);            // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: FumaDB transactions roll back when the callback rejects            throw new TransactionEffectDefect(exit.cause);          }),        catch: (cause): E | StorageFailure => {          if (cause instanceof TransactionEffectFailure) return cause.error as E;          if (cause instanceof TransactionEffectDefect) {            return fumaFailureFromCause("transaction", cause.cause);          }          return fumaFailureFromCause("transaction", cause);        },      }).pipe(        Effect.tap(() =>          Effect.forEach(commitHooks, (hook) => hook.pipe(Effect.ignoreCause({ log: false })), {            discard: true,          }),        ),      );    }).pipe(Effect.withSpan("fumadb.transaction")) as Effect.Effect<A, E | StorageFailure>;   return { use, transaction };}; export class FumaClient extends Context.Service<FumaClient, IFumaClient>()("executor/FumaClient") {  static layer = (db: FumaDb) => Layer.succeed(this)(makeFumaClient(db));}