Fix Kysely result plugins with Effect SQL clients

.changeset/quiet-kysely-results.md

@@ -0,0 +1,5 @@
+---
+"@effect/sql-kysely": patch
+---
+
+Apply Kysely result plugins to rows returned through Effect SQL clients. Other `QueryResult` metadata is unavailable because Effect SQL clients expose rows only.

packages/sql-kysely/src/internal/kysely.ts

@@ -63,7 +63,7 @@ export const makeWithSql = <DB>(config: KyselyConfig) =>
     const selectPrototype = Object.getPrototypeOf(db.selectFrom("" as any))
     patch(selectPrototype)
 
-    return effectifyWithSql(db, client, ["withTransaction", "compile"])
+    return effectifyWithSql(db, client, ["withTransaction", "compile"], config.plugins)
   })
 
 /**

packages/sql-kysely/src/internal/patch.ts

@@ -2,7 +2,7 @@ import type * as Client from "@effect/sql/SqlClient"
 import { SqlError } from "@effect/sql/SqlError"
 import * as Effect from "effect/Effect"
 import * as Effectable from "effect/Effectable"
-import type { Compilable } from "kysely"
+import type { Compilable, KyselyPlugin, QueryResult } from "kysely"
 
 const ATTR_DB_QUERY_TEXT = "db.query.text"
 
@@ -26,47 +26,66 @@ export const patch = (prototype: any) => {
   }
 }
 
-/**
- * @internal
- * replace at runtime the commit method on instances that have been patched by the provided one
- * this allows multiple client db instances to have different drivers (@effect/sql or kysely)
- */
 function effectifyWith(
   obj: any,
-  commit: () => Effect.Effect<ReadonlyArray<unknown>, SqlError>,
-  whitelist: Array<string>
+  commit: (plugins: ReadonlyArray<KyselyPlugin>) => Effect.Effect<ReadonlyArray<unknown>, SqlError>,
+  whitelist: Array<string>,
+  plugins: ReadonlyArray<KyselyPlugin> = []
 ) {
   if (typeof obj !== "object" || obj === null) {
     return obj
   }
   return new Proxy(obj, {
-    get(target, prop): any {
-      // Respect the proxy invariant: non-configurable, non-writable
-      // properties must return their actual value.
+    get(target, prop, receiver) {
+      // Proxy invariants require returning fixed properties unchanged.
       const desc = Object.getOwnPropertyDescriptor(target, prop)
       if (desc && !desc.configurable && !desc.writable) {
         return target[prop]
       }
       const prototype = Object.getPrototypeOf(target)
       if (Effect.EffectTypeId in prototype && prop === "commit") {
-        return commit.bind(target)
+        return commit.bind(target, plugins)
       }
       if (typeof (target[prop]) === "function") {
         if (typeof prop === "string" && whitelist.includes(prop)) {
           return target[prop].bind(target)
         }
-        return (...args: Array<any>) => effectifyWith(target[prop].call(target, ...args), commit, whitelist)
+        return (...args: Array<unknown>) => {
+          // Callback helpers need the proxy to retain plugin tracking.
+          if (prop === "$call" || (prop === "$if" && args[0])) {
+            return target[prop].call(receiver, ...args)
+          }
+          return effectifyWith(
+            target[prop].call(target, ...args),
+            commit,
+            whitelist,
+            prop === "withPlugin" ? [...plugins, args[0] as KyselyPlugin] : prop === "withoutPlugins" ? [] : plugins
+          )
+        }
       }
-      return effectifyWith(target[prop], commit, whitelist)
+      return effectifyWith(target[prop], commit, whitelist, plugins)
     }
   })
 }
 
 /** @internal */
 const makeSqlCommit = (client: Client.SqlClient) => {
-  return function(this: Compilable) {
-    const { parameters, sql } = this.compile()
-    return client.unsafe(sql, parameters as any)
+  return function(this: Compilable, plugins: ReadonlyArray<KyselyPlugin>) {
+    const { parameters, queryId, sql } = this.compile()
+    const execute = client.unsafe<Record<string, unknown>>(sql, parameters)
+    if (plugins.length === 0) return execute
+    return Effect.flatMap(execute, (rows) =>
+      Effect.map(
+        Effect.reduce(plugins, { rows: Array.from(rows) } as QueryResult<Record<string, unknown>>, (result, plugin) =>
+          Effect.tryPromise({
+            try: () =>
+              plugin.transformResult({ queryId, result }),
+            catch: (cause) =>
+              new SqlError({ cause })
+          })),
+        (result) =>
+          result.rows
+      ))
   }
 }
 
@@ -87,8 +106,12 @@ function executeCommit(this: Executable) {
 /**
  *  @internal
  */
-export const effectifyWithSql = <T>(obj: T, client: Client.SqlClient, whitelist: Array<string> = []): T =>
-  effectifyWith(obj, makeSqlCommit(client), whitelist)
+export const effectifyWithSql = <T>(
+  obj: T,
+  client: Client.SqlClient,
+  whitelist: Array<string> = [],
+  plugins: ReadonlyArray<KyselyPlugin> = []
+): T => effectifyWith(obj, makeSqlCommit(client), whitelist, plugins)
 
 /**
  *  @internal

packages/sql-kysely/test/Sqlite.test.ts

@@ -1,9 +1,9 @@
-import { SqlResolver } from "@effect/sql"
+import { SqlError, SqlResolver } from "@effect/sql"
 import * as SqliteKysely from "@effect/sql-kysely/Sqlite"
 import * as Sqlite from "@effect/sql-sqlite-node"
 import { assert, describe, it } from "@effect/vitest"
 import { Context, Effect, Exit, Layer, Option, Schema } from "effect"
-import type { Generated } from "kysely"
+import { CamelCasePlugin, type Generated, type KyselyPlugin, type QueryId } from "kysely"
 
 export interface User {
   id: Generated<number>
@@ -24,6 +24,78 @@ const SqliteLive = Sqlite.SqliteClient.layer({
 const KyselyLive = Layer.effect(SqliteDB, SqliteKysely.make<Database>()).pipe(Layer.provide(SqliteLive))
 
 describe("SqliteKysely", () => {
+  it.effect("result plugins", () =>
+    Effect.gen(function*() {
+      const db = yield* SqliteKysely.make<{ users: { userName: string } }>({
+        plugins: [new CamelCasePlugin()]
+      })
+      yield* db.schema.createTable("users").addColumn("userName", "text", (c) => c.notNull())
+      yield* db.insertInto("users").values({ userName: "Alice" })
+      assert.deepStrictEqual(yield* db.selectFrom("users").selectAll(), [{ userName: "Alice" }])
+      yield* db.withTransaction(
+        db.updateTable("users").set({ userName: "Bob" }).pipe(Effect.andThen(Effect.fail("rollback")))
+      ).pipe(Effect.flip)
+      assert.deepStrictEqual(yield* db.selectFrom("users").selectAll(), [{ userName: "Alice" }])
+    }).pipe(Effect.provide(SqliteLive)))
+
+  it.effect("scoped result plugins", () =>
+    Effect.gen(function*() {
+      const db = yield* SqliteKysely.make<{ users: { user_name: string } }>()
+      yield* db.schema.createTable("users").addColumn("user_name", "text")
+      yield* db.insertInto("users").values({ user_name: "Alice" })
+      const camel = db.withPlugin(new CamelCasePlugin())
+      assert.deepStrictEqual<unknown>(yield* camel.selectFrom("users").selectAll(), [{ userName: "Alice" }])
+      assert.deepStrictEqual(yield* camel.withoutPlugins().selectFrom("users").selectAll(), [{ user_name: "Alice" }])
+      assert.deepStrictEqual<unknown>(yield* db.selectFrom("users").selectAll().withPlugin(new CamelCasePlugin()), [
+        { userName: "Alice" }
+      ])
+      const query = db.selectFrom("users").selectAll()
+      assert.deepStrictEqual<unknown>(yield* query.$call((q) => q.withPlugin(new CamelCasePlugin())), [
+        { userName: "Alice" }
+      ])
+      assert.deepStrictEqual<unknown>(yield* query.$if(true, (q) => q.withPlugin(new CamelCasePlugin())), [
+        { userName: "Alice" }
+      ])
+      assert.deepStrictEqual(yield* query.$if(false, (q) => q.withPlugin(new CamelCasePlugin())), [
+        { user_name: "Alice" }
+      ])
+    }).pipe(Effect.provide(SqliteLive)))
+
+  it.effect("result plugin order and query identity", () =>
+    Effect.gen(function*() {
+      const queries = new WeakSet<QueryId>()
+      const plugin: KyselyPlugin = {
+        transformQuery: ({ node, queryId }) => {
+          queries.add(queryId)
+          return node
+        },
+        transformResult: ({ queryId, result }) => {
+          assert.isTrue(queries.has(queryId))
+          return Promise.resolve({
+            ...result,
+            rows: result.rows.map((row) => ({ ...row, userName: `${row.userName}!` }))
+          })
+        }
+      }
+      const db = yield* SqliteKysely.make<{ users: { userName: string } }>({
+        plugins: [new CamelCasePlugin(), plugin]
+      })
+      yield* db.schema.createTable("users").addColumn("userName", "text")
+      yield* db.insertInto("users").values({ userName: "Alice" })
+      assert.deepStrictEqual(yield* db.selectFrom("users").selectAll(), [{ userName: "Alice!" }])
+    }).pipe(Effect.provide(SqliteLive)))
+
+  it.effect("result plugin failures", () =>
+    Effect.gen(function*() {
+      const db = yield* SqliteKysely.make<{ users: { name: string } }>()
+      yield* db.schema.createTable("users").addColumn("name", "text")
+      const error = yield* db.selectFrom("users").selectAll().withPlugin({
+        transformQuery: ({ node }) => node,
+        transformResult: () => Promise.reject("boom")
+      }).pipe(Effect.flip)
+      assert(error instanceof SqlError.SqlError)
+    }).pipe(Effect.provide(SqliteLive)))
+
   it.effect("queries", () =>
     Effect.gen(function*() {
       const db = yield* SqliteDB