1import * as Data from "effect/Data";
2import * as Effect from "effect/Effect";
3import * as Redacted from "effect/Redacted";
4import * as Schedule from "effect/Schedule";
5import type { Client } from "pg";
6import {
7 makePgMigrationExecutor,
8 runMigrations,
9 type NormalizedMigrationsInput,
10 type StampedMigrationsState,
11} from "../SQL/Migrations/index.ts";
12import { importPg } from "../SQL/PostgresDriver.ts";
13import { readSqlFile } from "../SQL/SqlFile.ts";
14
15export class PostgresMigrationError extends Data.TaggedError(
16 "Fly.PostgresMigrationError",
17)<{
18 message: string;
19 cause?: unknown;
20}> {}
21
22/**
23 * Strip query-string SSL flags so `pg-connection-string` does not treat
24 * `sslmode=require` as `verify-full`. TLS and certificate verification
25 * are set on the client (`ssl.rejectUnauthorized: true`).
26 */
27export const stripSslQueryParams = (uri: string): string => {
28 try {
29 const url = new URL(uri);
30 url.searchParams.delete("sslmode");
31 url.searchParams.delete("channel_binding");
32 return url.toString();
33 } catch {
34 return uri;
35 }
36};
37
38const toMigrationError = (cause: unknown) =>
39 new PostgresMigrationError({
40 message: cause instanceof Error ? cause.message : String(cause),
41 cause,
42 });
43
44/**
45 * A freshly-provisioned Fly Managed Postgres cluster reports `ready` on the
46 * API before its hostname is resolvable and its endpoint accepts
47 * connections — reconcile's first migration connect can fail with
48 * `ENOTFOUND` (DNS still propagating) or a refused/reset socket. These are
49 * eventual consistency, not failures: retry the connect on a bounded
50 * schedule (~45s) before surfacing the error.
51 */
52const TRANSIENT_CONNECT_CODES = new Set([
53 "ENOTFOUND",
54 "EAI_AGAIN",
55 "ECONNREFUSED",
56 "ECONNRESET",
57 "ETIMEDOUT",
58]);
59
60const isTransientConnectError = (error: PostgresMigrationError): boolean => {
61 const code = (error.cause as { code?: unknown } | null | undefined)?.code;
62 return typeof code === "string" && TRANSIENT_CONNECT_CODES.has(code);
63};
64
65/** Open a pg client for the scope of `use`, closing it afterwards. */
66export const withPgClient = <A, E, R>(
67 connectionUri: Redacted.Redacted<string>,
68 use: (client: Client) => Effect.Effect<A, E, R>,
69): Effect.Effect<A, PostgresMigrationError | E, R> =>
70 Effect.acquireUseRelease(
71 Effect.tryPromise({
72 try: async () => {
73 const { Client } = await importPg();
74 const client = new Client({
75 connectionString: stripSslQueryParams(Redacted.value(connectionUri)),
76 ssl: { rejectUnauthorized: true },
77 });
78 await client.connect();
79 return client;
80 },
81 catch: toMigrationError,
82 }).pipe(
83 Effect.retry({
84 while: isTransientConnectError,
85 schedule: Schedule.spaced("3 seconds"),
86 times: 15,
87 }),
88 // Still unreachable after the bounded retry: almost always a missing
89 // route rather than propagation — MPG hostnames only exist on the org
90 // private network. Say so instead of a bare `getaddrinfo ENOTFOUND`.
91 Effect.mapError((error) =>
92 isTransientConnectError(error)
93 ? new PostgresMigrationError({
94 message:
95 `${error.message} — Fly Managed Postgres is only reachable ` +
96 "on the org's private network. Run the deploy behind a " +
97 "WireGuard peer or `fly mpg proxy`, or from a machine on " +
98 "6PN. See https://fly.io/docs/mpg/create-and-connect/",
99 cause: error.cause,
100 })
101 : error,
102 ),
103 ),
104 use,
105 (client) => Effect.promise(() => client.end().catch(() => {})),
106 );
107
108/**
109 * Fly Managed Postgres's migration adaptation is the shared pipeline
110 * with a connection-URI-scoped pg client as its executor. Use the
111 * direct (non-PgBouncer) URI so DDL and advisory locks work.
112 */
113export const runPgMigrations = (options: {
114 connectionUri: Redacted.Redacted<string>;
115 input: NormalizedMigrationsInput;
116 stamped: StampedMigrationsState;
117}) =>
118 runMigrations({
119 ...options,
120 withExecutor: (apply) =>
121 withPgClient(options.connectionUri, (client) =>
122 apply(makePgMigrationExecutor(client)),
123 ),
124 });
125
126/** Run a single SQL script against the database (used for `importFiles`). */
127export const runSql = (connectionUri: Redacted.Redacted<string>, sql: string) =>
128 withPgClient(connectionUri, (client) =>
129 Effect.tryPromise({
130 try: () => client.query(sql),
131 catch: toMigrationError,
132 }),
133 ).pipe(Effect.asVoid);
134
135export const runImports = (
136 connectionUri: Redacted.Redacted<string>,
137 importFiles: ReadonlyArray<string>,
138 rootDir: string,
139 previous: Record<string, string>,
140) =>
141 Effect.gen(function* () {
142 const hashes: Record<string, string> = { ...previous };
143 for (const filePath of importFiles) {
144 const file = yield* readSqlFile(rootDir, filePath);
145 if (previous[filePath] === file.hash) {
146 hashes[filePath] = file.hash;
147 continue;
148 }
149 yield* runSql(connectionUri, file.sql);
150 hashes[filePath] = file.hash;
151 }
152 const tracked = new Set(importFiles);
153 for (const key of Object.keys(hashes)) {
154 if (!tracked.has(key)) delete hashes[key];
155 }
156 return hashes;
157 });