1import * as ps from "@distilled.cloud/planetscale";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Redacted from "effect/Redacted";
5import * as Schedule from "effect/Schedule";
6import type { Client } from "pg";
7import {
8 makePgMigrationExecutor,
9 runMigrations,
10 type NormalizedMigrationsInput,
11 type StampedMigrationsState,
12} from "../../SQL/Migrations/index.ts";
13import { readSqlFile } from "../../SQL/SqlFile.ts";
14
15// `pg` is an optional peer dependency — loaded lazily so importing the
16// Planetscale provider never requires the driver unless migrations run.
17const importPg = () =>
18 import("pg").catch((cause) => {
19 throw new Error(
20 "Failed to load the 'pg' driver. Install the optional peer dependency 'pg' to run Planetscale Postgres migrations.",
21 { cause },
22 );
23 });
24
25const MIGRATION_ROLE_TTL_SECONDS = 600;
26
27export class PostgresMigrationError extends Data.TaggedError(
28 "Planetscale::PostgresMigrationError",
29)<{
30 message: string;
31 cause?: unknown;
32}> {}
33
34export interface PostgresMigrationTarget {
35 organization: string;
36 database: string;
37 branch: string;
38}
39
40/**
41 * PlanetScale Postgres's migration adaptation is exactly this: the shared
42 * pipeline with a temp-role-scoped pg client as its executor.
43 */
44export const runPostgresMigrations = (
45 target: PostgresMigrationTarget,
46 input: NormalizedMigrationsInput,
47 stamped: StampedMigrationsState,
48) =>
49 runMigrations({
50 input,
51 stamped,
52 withExecutor: (apply) =>
53 withPostgresClient(target, (client) =>
54 apply(makePgMigrationExecutor(client)),
55 ),
56 });
57
58export const runPostgresImports = (
59 target: PostgresMigrationTarget,
60 importFiles: ReadonlyArray<string>,
61 rootDir: string,
62 previous: Record<string, string>,
63) =>
64 Effect.gen(function* () {
65 const hashes: Record<string, string> = { ...previous };
66 for (const filePath of importFiles) {
67 const file = yield* readSqlFile(rootDir, filePath);
68 if (previous[filePath] === file.hash) {
69 hashes[filePath] = file.hash;
70 continue;
71 }
72 yield* runPostgresSql(target, file.sql);
73 hashes[filePath] = file.hash;
74 }
75 const tracked = new Set(importFiles);
76 for (const key of Object.keys(hashes)) {
77 if (!tracked.has(key)) delete hashes[key];
78 }
79 return hashes;
80 });
81
82const runPostgresSql = (target: PostgresMigrationTarget, sql: string) =>
83 withPostgresClient(target, (client) => pgExec(client, sql));
84
85const withPostgresClient = <A, E, R>(
86 target: PostgresMigrationTarget,
87 use: (client: Client) => Effect.Effect<A, E, R>,
88) =>
89 withTemporaryPostgresRole(target, (role) =>
90 Effect.acquireUseRelease(
91 Effect.gen(function* () {
92 const { Client } = yield* Effect.tryPromise({
93 try: importPg,
94 catch: toMigrationError,
95 });
96 const client = yield* Effect.sync(
97 () =>
98 new Client({
99 connectionString: stripPgSslQueryParams(
100 Redacted.value(role.connectionUrl),
101 ),
102 ssl: { rejectUnauthorized: true },
103 }),
104 );
105 yield* Effect.tryPromise({
106 try: () => client.connect(),
107 catch: toMigrationError,
108 });
109 return client;
110 }),
111 use,
112 (client) =>
113 Effect.tryPromise({
114 try: () => client.end(),
115 catch: toMigrationError,
116 }).pipe(Effect.catch(() => Effect.void)),
117 ),
118 );
119
120const withTemporaryPostgresRole = <A, E, R>(
121 target: PostgresMigrationTarget,
122 use: (role: {
123 id: string;
124 connectionUrl: Redacted.Redacted<string>;
125 }) => Effect.Effect<A, E, R>,
126) =>
127 Effect.acquireUseRelease(
128 Effect.gen(function* () {
129 const created = yield* ps.createRole({
130 organization: target.organization,
131 database: target.database,
132 branch: target.branch,
133 ttl: MIGRATION_ROLE_TTL_SECONDS,
134 inherited_roles: ["postgres"],
135 });
136
137 const password = created.password!;
138 const value = Redacted.value(password);
139 const connectionUrl = yield* Effect.sync(
140 () =>
141 `postgresql://${encodeURIComponent(created.username)}:${encodeURIComponent(value)}@${created.access_host_url}:5432/${created.database_name}?sslmode=verify-full`,
142 );
143 return {
144 id: created.id,
145 connectionUrl: Redacted.make(connectionUrl),
146 };
147 }),
148 use,
149 (role) =>
150 ps
151 .deleteRole({
152 organization: target.organization,
153 database: target.database,
154 branch: target.branch,
155 id: role.id,
156 successor: "postgres",
157 })
158 .pipe(
159 // Already-deleted roles are a success: nothing to clean up.
160 Effect.catchTag("NotFound", () => Effect.void),
161 Effect.retry({
162 schedule: Schedule.max([
163 Schedule.exponential("500 millis"),
164 Schedule.recurs(5),
165 ]),
166 }),
167 // Migrations succeeded; don't fail the parent over a release-step
168 // hiccup. The role's TTL bounds the orphan window; log loudly so
169 // an operator can clean it up manually if needed.
170 Effect.catch((cause: unknown) =>
171 Effect.logWarning(
172 `Failed to delete temporary Planetscale migration role after retries. ` +
173 `It will expire via TTL (~${MIGRATION_ROLE_TTL_SECONDS}s). ` +
174 `organization=${target.organization} database=${target.database} ` +
175 `branch=${target.branch} id=${role.id}`,
176 cause,
177 ),
178 ),
179 ),
180 );
181
182const pgExec = (client: Client, sql: string, values?: ReadonlyArray<unknown>) =>
183 Effect.tryPromise({
184 try: () =>
185 client.query(sql, values as Array<unknown>).then(() => undefined),
186 catch: toMigrationError,
187 });
188
189const toMigrationError = (cause: unknown) =>
190 new PostgresMigrationError({
191 message: cause instanceof Error ? cause.message : String(cause),
192 cause,
193 });
194
195const stripPgSslQueryParams = (uri: string): string => {
196 try {
197 const url = new URL(uri);
198 url.searchParams.delete("sslmode");
199 url.searchParams.delete("channel_binding");
200 return url.toString();
201 } catch {
202 return uri;
203 }
204};