1import * as spanner from "@distilled.cloud/gcp/spanner_v1";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Semaphore from "effect/Semaphore";
5import type { Database } from "./Database.ts";
6import { ExecuteSql, type ExecuteSqlRequest } from "./ExecuteSql.ts";
7import { bindGcpHost } from "../Host.ts";
8import { grantFor } from "../HttpBinding.ts";
9
10/**
11 * HTTP implementation of {@link ExecuteSql}.
12 *
13 * Reuses one multiplexed session per database for the life of the runtime
14 * instance (multiplexed sessions serve concurrent requests and are never
15 * deleted by the client). A session the server has dropped
16 * (`SessionNotFound`) is replaced once and the statement retried.
17 *
18 * @layer
19 * @provides GCP.Spanner.ExecuteSql
20 */
21export const ExecuteSqlHttp = Layer.effect(
22 ExecuteSql,
23 Effect.gen(function* () {
24 const createSession =
25 yield* spanner.createProjectsInstancesDatabasesSessions;
26 const executeSql =
27 yield* spanner.executeSqlProjectsInstancesDatabasesSessions;
28 // Session names by database name; plain values, safe to keep per instance.
29 const sessions = new Map<string, string>();
30 const lock = yield* Semaphore.make(1);
31
32 const sessionFor = (database: string, stale?: string) =>
33 lock.withPermits(1)(
34 Effect.gen(function* () {
35 const cached = sessions.get(database);
36 if (cached !== undefined && cached !== stale) return cached;
37 const session = yield* createSession({
38 database,
39 body: { session: { multiplexed: true } },
40 });
41 const name = session.name ?? "";
42 sessions.set(database, name);
43 return name;
44 }),
45 );
46
47 return Effect.fn(function* (database: Database) {
48 yield* bindGcpHost({
49 tag: "GCP.Spanner.ExecuteSql",
50 resource: database,
51 iam: [
52 grantFor(
53 { role: "roles/spanner.databaseUser", on: "spanner.database" },
54 database.name,
55 ),
56 ],
57 });
58 const name = yield* database.name;
59 return Effect.fn(`GCP.Spanner.ExecuteSql(${database.LogicalId})`)(
60 function* (request: ExecuteSqlRequest) {
61 const databaseName = yield* name;
62 const session = yield* sessionFor(databaseName);
63 return yield* executeSql({ session, body: request }).pipe(
64 Effect.catchTag("SessionNotFound", () =>
65 Effect.gen(function* () {
66 const fresh = yield* sessionFor(databaseName, session);
67 return yield* executeSql({ session: fresh, body: request });
68 }),
69 ),
70 );
71 },
72 );
73 });
74 }),
75);