fix: type DatabaseError on pure-save and SQL/Pg store adapters

packages/effect-app/src/Model/Repository/ext.ts

@@ -5,7 +5,7 @@ import * as Request from "effect/Request"
 import * as RequestResolver from "effect/RequestResolver"
 import * as Array from "../../Array.ts"
 import type { NonEmptyArray } from "../../Array.ts"
-import { type InvalidStateError, NotFoundError, type OptimisticConcurrencyException } from "../../client/errors.ts"
+import { type DatabaseError, type InvalidStateError, NotFoundError, type OptimisticConcurrencyException } from "../../client/errors.ts"
 import * as Effect from "../../Effect.ts"
 import * as Option from "../../Option.ts"
 import { type FixEnv, type PureEnv, runTerm } from "../../Pure.ts"
@@ -89,7 +89,11 @@ export const extendRepo = <
     A
   >(
     gen: Effect.Effect<readonly [Iterable<P>, Iterable<Evt>, A], E, R>
-  ) {
+  ): Effect.Effect<
+    A,
+    E | InvalidStateError | OptimisticConcurrencyException | DatabaseError,
+    R | RSchema | RPublish
+  > {
     return Effect.flatMap(gen, ([items, events, a]) => repo.saveAndPublish(items, events).pipe(Effect.map(() => a)))
   }
 
@@ -113,6 +117,8 @@ export const extendRepo = <
     )
   }
 
+  type PureStoreError = InvalidStateError | OptimisticConcurrencyException | DatabaseError
+
   const queryAndSavePure: {
     <A, E2, R2, T2 extends T>(
       q: (
@@ -121,8 +127,10 @@ export const extendRepo = <
       pure: Effect.Effect<A, E2, FixEnv<R2, Evt, T, T2>>
     ): Effect.Effect<
       A,
-      InvalidStateError | OptimisticConcurrencyException | NotFoundError<ItemType> | E2,
-      Exclude<R2, {
+      PureStoreError | NotFoundError<ItemType> | E2,
+      | RSchema
+      | RPublish
+      | Exclude<R2, {
         env: PureEnv<Evt, T, T2>
       }>
     >
@@ -136,7 +144,7 @@ export const extendRepo = <
       pure: Effect.Effect<A, E2, FixEnv<R2, Evt, readonly T[], readonly T2[]>>
     ): Effect.Effect<
       A,
-      InvalidStateError | OptimisticConcurrencyException | E2,
+      PureStoreError | E2,
       | RSchema
       | RPublish
       | Exclude<R2, {
@@ -154,31 +162,33 @@ export const extendRepo = <
       batch: "batched" | number
     ): Effect.Effect<
       A[],
-      InvalidStateError | OptimisticConcurrencyException | E2,
+      PureStoreError | E2,
       | RSchema
       | RPublish
       | Exclude<R2, {
         env: PureEnv<Evt, readonly T[], readonly T2[]>
       }>
     >
-  } = (q, pure, batch?: "batched" | number) =>
+  } = ((q: any, pure: any, batch?: "batched" | number) =>
+    // Overload dispatch: query returns T | T[]; pure helpers are generic on item shape.
+    // Runtime path is query → pure term → saveAndPublish (which raises DatabaseError).
     repo.query(q).pipe(
       Effect.andThen((_) =>
         Array.isArray(_)
           ? batch === undefined
-            ? saveManyWithPure_(_ as any, pure as any)
-            : saveManyWithPureBatched_(_ as any, pure as any, batch === "batched" ? 100 : batch)
-          : saveWithPure_(_ as any, pure as any)
+            ? saveManyWithPure_(_ as T[], pure)
+            : saveManyWithPureBatched_(_ as T[], pure, batch === "batched" ? 100 : batch)
+          : saveWithPure_(_ as T, pure)
       )
-    ) as any
+    )) as typeof queryAndSavePure
 
   const saveManyWithPure: {
     <R, A, E, S1 extends T, S2 extends T>(
       items: Iterable<S1>,
       pure: Effect.Effect<A, E, FixEnv<R, Evt, readonly S1[], readonly S2[]>>
     ): Effect.Effect<
       A,
-      InvalidStateError | OptimisticConcurrencyException | E,
+      PureStoreError | E,
       | RSchema
       | RPublish
       | Exclude<R, {
@@ -191,43 +201,45 @@ export const extendRepo = <
       batch: "batched" | number
     ): Effect.Effect<
       A[],
-      InvalidStateError | OptimisticConcurrencyException | E,
+      PureStoreError | E,
       | RSchema
       | RPublish
       | Exclude<R, {
         env: PureEnv<Evt, readonly S1[], readonly S2[]>
       }>
     >
-  } = (items, pure, batch?: "batched" | number) =>
+  } = ((items: Iterable<T>, pure: any, batch?: "batched" | number) =>
     batch
       ? Effect.forEach(
-        Array.chunksOf(items, batch === "batched" ? 100 : batch),
-        (batch) =>
+        Array.chunksOf([...items], batch === "batched" ? 100 : batch),
+        (batchItems) =>
           saveAllWithEffectInt(
-            runTerm(pure, batch)
+            runTerm(pure, batchItems as any)
           )
       )
       : saveAllWithEffectInt(
-        runTerm(pure, [...items])
-      )
+        runTerm(pure, [...items] as any)
+      )) as typeof saveManyWithPure
 
   const byIdAndSaveWithPure: {
     <R, A, E, S2 extends T>(
       id: T[IdKey],
       pure: Effect.Effect<A, E, FixEnv<R, Evt, T, S2>>
     ): Effect.Effect<
       A,
-      InvalidStateError | OptimisticConcurrencyException | NotFoundError<ItemType> | E,
+      PureStoreError | NotFoundError<ItemType> | E,
       | RSchema
       | RPublish
       | Exclude<R, {
         env: PureEnv<Evt, T, S2>
       }>
     >
-  } = (id, pure): any => get(id).pipe(Effect.flatMap((item) => saveWithPure_(item, pure)))
+  } = (id, pure) => get(id).pipe(Effect.flatMap((item) => saveWithPure_(item, pure)))
 
+  // query can raise DatabaseError; NotFound is completed per-entry. Dual package
+  // round-trips may still surface other store errors via failCause — keep channel honest.
   type Req =
-    & Request.Request<T, NotFoundError<ItemType>>
+    & Request.Request<T, NotFoundError<ItemType> | DatabaseError>
     & { _tag: `Get${ItemType}`; id: T[IdKey] }
   const _request = Request.tagged<Req>(`Get${repo.itemType}`)
 
@@ -237,9 +249,9 @@ export const extendRepo = <
       _key: unknown
     ) =>
       (repo.query(Q.where(repo.idKey as any, "in" as any, entries.map((_) => _.request.id)) as any) as Effect.Effect<
-        readonly T[]
+        readonly T[],
+        DatabaseError
       >)
-        // TODO
         .pipe(
           Effect.andThen((items) =>
             Effect.forEach(entries, (entry) =>
@@ -270,7 +282,7 @@ export const extendRepo = <
      * Enables chunked writes for large batches via `options.batch`.
      * Note: batching breaks transactional properties because chunks are saved independently.
      */
-    save: ((itemOrItems: T | ReadonlyArray<T>, options?: BatchOptions) => {
+    save: (itemOrItems: T | ReadonlyArray<T>, options?: BatchOptions) => {
       const items = asReadonlyArray(itemOrItems)
       if (!Array.isReadonlyArrayNonEmpty(items)) {
         return Effect.void
@@ -284,23 +296,16 @@ export const extendRepo = <
         (batch) => repo.saveAndPublish(batch),
         { discard: true }
       )
-    }) as (
-      itemOrItems: T | ReadonlyArray<T>,
-      options?: BatchOptions
-    ) => Effect.Effect<
-      void,
-      InvalidStateError | OptimisticConcurrencyException,
-      RSchema | RPublish
-    >,
+    },
     saveWithEvents: (events: Iterable<Evt>) => (...items: NonEmptyArray<T>) => repo.saveAndPublish(items, events),
     /**
      * Enables chunked deletes for large batches via `options.batch`.
      * Note: batching breaks transactional properties because chunks are removed independently.
      */
-    remove: ((itemOrItems: T | ReadonlyArray<T>, options?: BatchOptions) => {
+    remove: (itemOrItems: T | ReadonlyArray<T>, options?: BatchOptions) => {
       const items = asReadonlyArray(itemOrItems)
       if (!Array.isReadonlyArrayNonEmpty(items)) {
-        return Effect.void
+        return Effect.void as Effect.Effect<void, DatabaseError, RSchema | RPublish>
       }
       const batchSize = getBatchSize(options?.batch)
       if (batchSize === undefined) {
@@ -311,18 +316,15 @@ export const extendRepo = <
         (batch) => repo.removeAndPublish(batch),
         { discard: true }
       )
-    }) as (
-      itemOrItems: T | ReadonlyArray<T>,
-      options?: BatchOptions
-    ) => Effect.Effect<void, never, RSchema | RPublish>,
+    },
     /**
      * Enables chunked deletes for large batches via `options.batch`.
      * Note: batching breaks transactional properties because chunks are removed independently.
      */
-    removeById: ((idOrIds: T[IdKey] | ReadonlyArray<T[IdKey]>, options?: BatchOptions) => {
+    removeById: (idOrIds: T[IdKey] | ReadonlyArray<T[IdKey]>, options?: BatchOptions) => {
       const ids = asReadonlyArray(idOrIds)
       if (!Array.isReadonlyArrayNonEmpty(ids)) {
-        return Effect.void
+        return Effect.void as Effect.Effect<void, DatabaseError, RSchema>
       }
       const batchSize = getBatchSize(options?.batch)
       if (batchSize === undefined) {
@@ -333,10 +335,7 @@ export const extendRepo = <
         (batch) => repo.removeById(batch),
         { discard: true }
       )
-    }) as (
-      idOrIds: T[IdKey] | ReadonlyArray<T[IdKey]>,
-      options?: BatchOptions
-    ) => Effect.Effect<void, never, RSchema>,
+    },
     queryAndSavePure,
     saveManyWithPure,
     byIdAndSaveWithPure,

packages/infra/src/Store/SQL.ts

@@ -25,6 +25,16 @@ const sqlIsTransient = (e: unknown) =>
 // which would turn it into an opaque, non-serializable defect).
 const toDatabaseError = (e: unknown) =>
   new DatabaseError({ message: `SQL request failed: ${sqlErrorMessage(e)}`, transient: sqlIsTransient(e), cause: e })
+// withTransaction may re-raise setInternal's typed errors or add SqlError on begin/commit.
+// Preserve DatabaseError / OCC by `_tag` (not instanceof — dual package instances break it);
+// map residual SQL failures to DatabaseError.
+const preserveStoreError = (e: unknown): DatabaseError | OptimisticConcurrencyException => {
+  if (e !== null && typeof e === "object" && "_tag" in e) {
+    if (e._tag === "DatabaseError") return e as DatabaseError
+    if (e._tag === "OptimisticConcurrencyException") return e as OptimisticConcurrencyException
+  }
+  return toDatabaseError(e)
+}
 
 export type WithNsTransactionFn = <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
 
@@ -155,7 +165,7 @@ function makeSQLStoreInt(system: DbSystem, dialect: SQLDialect, jsonColumnType:
           sql
             .withTransaction(Effect.forEach(items, (e) => setInternal(e, ns)))
             .pipe(
-              Effect.orDie,
+              Effect.mapError(preserveStoreError),
               Effect.map((_) => _ as unknown as NonEmptyReadonlyArray<PM>)
             )
 
@@ -172,7 +182,12 @@ function makeSQLStoreInt(system: DbSystem, dialect: SQLDialect, jsonColumnType:
           yield* InfraLogger.logInfo(`Seeding data for ${name} (namespace: ${ns})`)
           const items = yield* seed.pipe(Effect.provide(ctx), Effect.orDie)
           const ne = toNonEmptyArray([...items])
-          if (Option.isSome(ne)) yield* bulkSetInternal(ne.value, ns)
+          // Seed inserts are not concurrent; OCC here is a programming defect.
+          if (Option.isSome(ne)) {
+            yield* bulkSetInternal(ne.value, ns).pipe(
+              Effect.catchTag("OptimisticConcurrencyException", Effect.die)
+            )
+          }
           yield* exec(
             `INSERT INTO "_migrations" (id, version) VALUES (?, ?)`,
             [`${tableName}::${ns}`, tableName]
@@ -488,7 +503,7 @@ function makeSQLiteStorePerNs(
           sql
             .withTransaction(Effect.forEach(items, (e) => setInternal(e, ns)))
             .pipe(
-              Effect.orDie,
+              Effect.mapError(preserveStoreError),
               Effect.map((_) => _ as unknown as NonEmptyReadonlyArray<PM>)
             ))
 
@@ -506,7 +521,12 @@ function makeSQLiteStorePerNs(
         yield* InfraLogger.logInfo(`Seeding data for ${name} (namespace: ${ns})`)
         const items = yield* seed.pipe(Effect.provide(ctx), Effect.orDie)
         const ne = toNonEmptyArray([...items])
-        if (Option.isSome(ne)) yield* bulkSetInternal(ne.value, ns)
+        // Seed inserts are not concurrent; OCC here is a programming defect.
+        if (Option.isSome(ne)) {
+          yield* bulkSetInternal(ne.value, ns).pipe(
+            Effect.catchTag("OptimisticConcurrencyException", Effect.die)
+          )
+        }
         yield* exec(
           ns,
           `INSERT INTO "_migrations" (id, version) VALUES (?, ?)`,

packages/infra/src/Store/SQL/Pg.ts

@@ -9,12 +9,30 @@ import * as Option from "effect-app/Option"
 import { type FilterArgs, type PersistenceModelType, type StorageConfig, type Store, type StoreConfig, storeId, StoreMaker } from "effect-app/Store"
 import * as Struct from "effect/Struct"
 import { SqlClient } from "effect/unstable/sql"
-import { OptimisticConcurrencyException } from "../../errors.ts"
+import { DatabaseError, OptimisticConcurrencyException } from "../../errors.ts"
 import { InfraLogger } from "../../logger.ts"
 import { annotateDb } from "../../otel.ts"
 import { makeETag } from "../utils.ts"
 import { buildWhereSQLQuery, logQuery, pgDialect } from "./query.ts"
 
+const sqlErrorMessage = (e: unknown) => (e as any)?.message ? String((e as any).message) : String(e)
+const sqlIsTransient = (e: unknown) =>
+  /timeout|ETIMEDOUT|ECONNRESET|ECONNREFUSED|connection|deadlock|too many connections/i.test(sqlErrorMessage(e))
+// Map a SqlError into a typed, serializable DatabaseError (instead of `.orDie`,
+// which would turn it into an opaque, non-serializable defect).
+const toDatabaseError = (e: unknown) =>
+  new DatabaseError({ message: `SQL request failed: ${sqlErrorMessage(e)}`, transient: sqlIsTransient(e), cause: e })
+// withTransaction may re-raise setInternal's typed errors or add SqlError on begin/commit.
+// Preserve DatabaseError / OCC by `_tag` (not instanceof — dual package instances break it);
+// map residual SQL failures to DatabaseError.
+const preserveStoreError = (e: unknown): DatabaseError | OptimisticConcurrencyException => {
+  if (e !== null && typeof e === "object" && "_tag" in e) {
+    if (e._tag === "DatabaseError") return e as DatabaseError
+    if (e._tag === "OptimisticConcurrencyException") return e as OptimisticConcurrencyException
+  }
+  return toDatabaseError(e)
+}
+
 const parseRow = <Encoded extends FieldValues>(
   row: { id: string; _etag: string | null; data: unknown },
   idKey: PropertyKey,
@@ -85,7 +103,8 @@ const makePgStore = Effect.fnUntraced(function*({ prefix }: StorageConfig) {
         return { id, _etag: newE._etag!, data, item: newE }
       }
 
-      const exec = (query: string, params?: readonly unknown[]) => sql.unsafe(query, params as any).pipe(Effect.orDie)
+      const exec = (query: string, params?: readonly unknown[]) =>
+        sql.unsafe(query, params as any).pipe(Effect.mapError(toDatabaseError))
 
       const setInternal = Effect.fnUntraced(function*(e: PM, ns: string) {
         const row = toRow(e)
@@ -130,12 +149,12 @@ const makePgStore = Effect.fnUntraced(function*({ prefix }: StorageConfig) {
         sql
           .withTransaction(Effect.forEach(items, (e) => setInternal(e, ns)))
           .pipe(
-            Effect.orDie,
+            Effect.mapError(preserveStoreError),
             Effect.map((_) => _ as unknown as NonEmptyReadonlyArray<PM>)
           )
 
       const ctx = yield* Effect.context<R>()
-      const seedCache = new Map<string, Effect.Effect<void>>()
+      const seedCache = new Map<string, Effect.Effect<void, DatabaseError>>()
       const makeSeedEffect = Effect.fnUntraced(function*(ns: string) {
         yield* ensureTable
         if (!seed) return
@@ -147,7 +166,12 @@ const makePgStore = Effect.fnUntraced(function*({ prefix }: StorageConfig) {
         yield* InfraLogger.logInfo(`Seeding data for ${name} (namespace: ${ns})`)
         const items = yield* seed.pipe(Effect.provide(ctx), Effect.orDie)
         const ne = toNonEmptyArray([...items])
-        if (Option.isSome(ne)) yield* bulkSetInternal(ne.value, ns)
+        // Seed inserts are not concurrent; OCC here is a programming defect.
+        if (Option.isSome(ne)) {
+          yield* bulkSetInternal(ne.value, ns).pipe(
+            Effect.catchTag("OptimisticConcurrencyException", Effect.die)
+          )
+        }
         yield* exec(
           `INSERT INTO "_migrations" (id, version) VALUES ($1, $2)`,
           [`${tableName}::${ns}`, tableName]
@@ -349,8 +373,10 @@ const makePgStore = Effect.fnUntraced(function*({ prefix }: StorageConfig) {
           )
       }
 
-      // Eagerly seed primary namespace on initialization
-      yield* seedNamespace("primary")
+      // Eagerly seed primary namespace on initialization. A seed failure at
+      // construction is fatal (orDie); the per-call `seedNamespace` still
+      // surfaces DatabaseError to callers.
+      yield* seedNamespace("primary").pipe(Effect.orDie)
 
       return s
     })