1import * as Data from "effect/Data";
2import * as Effect from "effect/Effect";
3import * as Redacted from "effect/Redacted";
4import type { Client } from "pg";
5import {
6 makePgMigrationExecutor,
7 runMigrations,
8 type NormalizedMigrationsInput,
9 type StampedMigrationsState,
10} from "../SQL/Migrations/index.ts";
11import { importPg } from "../SQL/PostgresDriver.ts";
12
13export class PgError extends Data.TaggedError("PgError")<{
14 message: string;
15 cause?: unknown;
16}> {}
17
18/**
19 * Strip query-string SSL flags from a Neon connection URI. Neon's URIs
20 * include `sslmode=require` and `channel_binding=require`, which trigger
21 * a deprecation warning from `pg-connection-string`:
22 *
23 * SECURITY WARNING: The SSL modes 'prefer', 'require', and 'verify-ca'
24 * are treated as aliases for 'verify-full'.
25 *
26 * We control SSL programmatically via the `ssl` option below, so removing
27 * the conflicting query params silences the warning without changing the
28 * effective behavior.
29 */
30const stripSslQueryParams = (uri: string): string => {
31 try {
32 const url = new URL(uri);
33 url.searchParams.delete("sslmode");
34 url.searchParams.delete("channel_binding");
35 return url.toString();
36 } catch {
37 return uri;
38 }
39};
40
41const toPgError = (cause: unknown) =>
42 new PgError({
43 message: cause instanceof Error ? cause.message : String(cause),
44 cause,
45 });
46
47/** Open a pg client for the scope of `use`, closing it afterwards. */
48export const withPgClient = <A, E, R>(
49 connectionUri: Redacted.Redacted<string>,
50 use: (client: Client) => Effect.Effect<A, E, R>,
51): Effect.Effect<A, PgError | E, R> =>
52 Effect.acquireUseRelease(
53 Effect.tryPromise({
54 try: async () => {
55 const { Client } = await importPg();
56 const client = new Client({
57 connectionString: stripSslQueryParams(Redacted.value(connectionUri)),
58 ssl: { rejectUnauthorized: false },
59 });
60 await client.connect();
61 return client;
62 },
63 catch: toPgError,
64 }),
65 use,
66 (client) => Effect.promise(() => client.end().catch(() => {})),
67 );
68
69/**
70 * Neon's migration adaptation is exactly this: the shared pipeline with a
71 * connection-URI-scoped pg client as its executor.
72 */
73export const runPgMigrations = (options: {
74 connectionUri: Redacted.Redacted<string>;
75 input: NormalizedMigrationsInput;
76 stamped: StampedMigrationsState;
77}) =>
78 runMigrations({
79 ...options,
80 withExecutor: (apply) =>
81 withPgClient(options.connectionUri, (client) =>
82 apply(makePgMigrationExecutor(client)),
83 ),
84 });
85
86/**
87 * Run a single SQL script against the database (used for `importFiles`).
88 */
89export const runSql = (connectionUri: Redacted.Redacted<string>, sql: string) =>
90 withPgClient(connectionUri, (client) =>
91 Effect.tryPromise({
92 try: () => client.query(sql),
93 catch: toPgError,
94 }),
95 ).pipe(Effect.asVoid);