1import {
2 inlineSqlParams,
3 MigrationError,
4 migrationError,
5 runMigrations,
6 type MigrationDialect,
7 type NormalizedMigrationsInput,
8 type SqlExecutor,
9 type StampedMigrationsState,
10} from "@alchemy.run/sql/Migrations";
11import * as Lambda from "@distilled.cloud/aws/lambda";
12import * as rds from "@distilled.cloud/aws/rds";
13import * as rdsdata from "@distilled.cloud/aws/rds-data";
14import * as secretsmanager from "@distilled.cloud/aws/secrets-manager";
15import * as Effect from "effect/Effect";
16import * as Redacted from "effect/Redacted";
17import * as Schedule from "effect/Schedule";
18import * as Stream from "effect/Stream";
19import { withMigrationConnection, type MigrationConnectionTarget } from "./MigrationConnection.ts";
20import type { MigrationRunnerRequest, MigrationRunnerResponse } from "./MigrationRunnerHandler.ts";
21
22/**
23 * How migrations reach an RDS database:
24 *
25 * - `"data-api"` — the RDS Data API over HTTPS. Aurora clusters with the
26 * Data API enabled only; no network route needed.
27 * - `"direct"` — a `pg`/`mysql2` connection from the deploying machine to
28 * the database endpoint. Needs a network route (public endpoint, VPN,
29 * bastion).
30 * - an `AWS.RDS.MigrationRunner` (or any `{ functionArn }`) — a Lambda in
31 * the database's VPC opens the connection; the deploying machine only
32 * invokes it.
33 */
34export type MigrationsTransport = "data-api" | "direct" | { functionArn: string };
35
36const dialects: Record<string, "postgres" | "mysql" | undefined> = {
37 postgres: "postgres",
38 "aurora-postgresql": "postgres",
39 mysql: "mysql",
40 mariadb: "mysql",
41 "aurora-mysql": "mysql",
42};
43
44/** The writer endpoint, database and credential sources of an instance. */
45interface MigrationTarget {
46 dialect: "postgres" | "mysql";
47 host: string | undefined;
48 port: number | undefined;
49 database: string | undefined;
50 /** The cluster ARN when the Data API is enabled on the instance's cluster. */
51 dataApiArn: string | undefined;
52 managedSecretArn: string | undefined;
53}
54
55const describeTarget = Effect.fn(function* (instance: rds.DBInstance) {
56 const identifier = instance.DBInstanceIdentifier ?? "unknown";
57 if (instance.DBClusterIdentifier) {
58 const cluster = yield* rds
59 .describeDBClusters({ DBClusterIdentifier: instance.DBClusterIdentifier })
60 .pipe(
61 Effect.map((response) => response.DBClusters?.[0]),
62 Effect.catchTag("DBClusterNotFoundFault", () => Effect.undefined),
63 );
64 if (!cluster) {
65 return yield* new MigrationError({
66 message: `DB cluster '${instance.DBClusterIdentifier}' of instance '${identifier}' not found`,
67 });
68 }
69 const engine = cluster.Engine ?? "";
70 const dialect = dialects[engine];
71 if (!dialect) {
72 return yield* unsupportedEngine(instance.DBClusterIdentifier, engine);
73 }
74 return {
75 dialect,
76 host: cluster.Endpoint,
77 port: cluster.Port,
78 database: cluster.DatabaseName,
79 dataApiArn: cluster.HttpEndpointEnabled ? cluster.DBClusterArn : undefined,
80 managedSecretArn: cluster.MasterUserSecret?.SecretArn,
81 } satisfies MigrationTarget;
82 }
83 const engine = instance.Engine ?? "";
84 const dialect = dialects[engine];
85 if (!dialect) {
86 return yield* unsupportedEngine(identifier, engine);
87 }
88 return {
89 dialect,
90 host: instance.Endpoint?.Address,
91 port: instance.Endpoint?.Port,
92 database: instance.DBName,
93 dataApiArn: undefined,
94 managedSecretArn: instance.MasterUserSecret?.SecretArn,
95 } satisfies MigrationTarget;
96});
97
98const unsupportedEngine = (identifier: string, engine: string) =>
99 new MigrationError({
100 message: `'${identifier}' runs ${engine || "an unknown engine"}; migrations support PostgreSQL, MySQL and MariaDB engines (including Aurora)`,
101 });
102
103// An auto-paused Serverless v2 writer resumes on the first request.
104const resumeRetry = {
105 while: (e: { readonly _tag: string }) => e._tag === "DatabaseResumingException",
106 schedule: Schedule.spaced("5 seconds"),
107 times: 10,
108};
109
110/**
111 * The Data API as a {@link SqlExecutor}. Queries inline their params (the
112 * Data API binds `:name` placeholders only), and batches run inside a
113 * Data API transaction.
114 */
115const makeDataApiExecutor = Effect.fn(function* (options: {
116 dialect: MigrationDialect;
117 resourceArn: string;
118 secretArn: string;
119 database: string | undefined;
120}) {
121 const executeStatement = yield* rdsdata.executeStatement;
122 const beginTransaction = yield* rdsdata.beginTransaction;
123 const commitTransaction = yield* rdsdata.commitTransaction;
124 const rollbackTransaction = yield* rdsdata.rollbackTransaction;
125 const { dialect, resourceArn, secretArn, database } = options;
126
127 return {
128 dialect,
129 query: (sql, params) =>
130 executeStatement({
131 resourceArn,
132 secretArn,
133 database,
134 sql: inlineSqlParams(sql, params ?? [], dialect),
135 formatRecordsAs: "JSON",
136 }).pipe(
137 Effect.retry(resumeRetry),
138 Effect.mapError(migrationError("Data API query")),
139 Effect.flatMap((result) =>
140 Effect.try({
141 try: () =>
142 (result.formattedRecords ? JSON.parse(result.formattedRecords) : []) as Array<
143 Record<string, unknown>
144 >,
145 catch: (cause) =>
146 new MigrationError({
147 message: `Data API returned malformed records: ${String(cause)}`,
148 cause,
149 }),
150 }),
151 ),
152 ),
153 batch: (statements) =>
154 Effect.gen(function* () {
155 const { transactionId } = yield* beginTransaction({
156 resourceArn,
157 secretArn,
158 database,
159 }).pipe(
160 Effect.retry(resumeRetry),
161 Effect.mapError(migrationError("Data API begin transaction")),
162 );
163 if (!transactionId) {
164 return yield* new MigrationError({ message: "Data API returned no transaction id" });
165 }
166 yield* Effect.gen(function* () {
167 yield* Effect.forEach(
168 statements,
169 (sql) => executeStatement({ resourceArn, secretArn, database, sql, transactionId }),
170 { discard: true },
171 );
172 yield* commitTransaction({ resourceArn, secretArn, transactionId });
173 }).pipe(
174 Effect.onError(() =>
175 rollbackTransaction({ resourceArn, secretArn, transactionId }).pipe(Effect.ignore),
176 ),
177 Effect.mapError(migrationError("Data API migration batch")),
178 );
179 }),
180 } satisfies SqlExecutor;
181});
182
183/**
184 * An in-VPC `AWS.RDS.MigrationRunner` as a {@link SqlExecutor}: every
185 * query and every migration batch is one synchronous invocation that opens
186 * its own connection. Params are inlined so the payload is plain SQL.
187 */
188const makeRunnerExecutor = Effect.fn(function* (options: {
189 functionArn: string;
190 target: MigrationConnectionTarget;
191}) {
192 const invoke = yield* Lambda.invoke;
193 const { functionArn, target } = options;
194 const call = (request: MigrationRunnerRequest) =>
195 Effect.gen(function* () {
196 const response = yield* invoke({
197 FunctionName: functionArn,
198 Payload: JSON.stringify(request),
199 }).pipe(Effect.mapError(migrationError(`Invoking migration runner ${functionArn}`)));
200 const body = response.Payload
201 ? yield* response.Payload.pipe(
202 Stream.decodeText(),
203 Stream.mkString,
204 Effect.mapError(
205 (cause) =>
206 new MigrationError({
207 message: `Failed to read the migration runner's response: ${cause.message}`,
208 cause,
209 }),
210 ),
211 )
212 : "";
213 const parsed = yield* Effect.try({
214 try: () => JSON.parse(body) as MigrationRunnerResponse & { errorMessage?: string },
215 catch: (cause) =>
216 new MigrationError({
217 message: `Migration runner returned a malformed response: ${body}`,
218 cause,
219 }),
220 });
221 if (response.FunctionError) {
222 return yield* new MigrationError({
223 message: `Migration runner ${functionArn} failed: ${parsed.errorMessage ?? body}`,
224 });
225 }
226 if (!parsed.ok) {
227 return yield* new MigrationError({ message: parsed.message });
228 }
229 return parsed.rows;
230 });
231
232 return {
233 dialect: target.dialect,
234 query: (sql, params) =>
235 call({ op: "query", target, sql: inlineSqlParams(sql, params ?? [], target.dialect) }),
236 batch: (statements) => call({ op: "batch", target, statements }).pipe(Effect.asVoid),
237 } satisfies SqlExecutor;
238});
239
240const readSecretCredentials = Effect.fn(function* (secretArn: string) {
241 const value = yield* secretsmanager
242 .getSecretValue({ SecretId: secretArn })
243 .pipe(Effect.mapError(migrationError(`Reading secret ${secretArn}`)));
244 const secretString =
245 value.SecretString === undefined
246 ? "{}"
247 : typeof value.SecretString === "string"
248 ? value.SecretString
249 : Redacted.value(value.SecretString);
250 const secret = yield* Effect.try({
251 try: () => JSON.parse(secretString) as { username?: string; password?: string },
252 catch: (cause) => new MigrationError({ message: `Secret ${secretArn} is not JSON`, cause }),
253 });
254 if (!secret.username || !secret.password) {
255 return yield* new MigrationError({
256 message: `Secret ${secretArn} has no "username" and "password"`,
257 });
258 }
259 return { username: secret.username, password: secret.password };
260});
261
262/**
263 * Apply migrations to the database an RDS instance serves — its own, or its
264 * Aurora cluster's when it is a cluster member. Call once the instance is
265 * available. Credentials, in order: `secretArn`, then `credentials`
266 * (direct and Lambda transports only), then the RDS-managed master secret.
267 */
268export const runDBInstanceMigrations = Effect.fn(function* (options: {
269 instance: rds.DBInstance;
270 input: NormalizedMigrationsInput;
271 stamped: StampedMigrationsState;
272 transport: MigrationsTransport | undefined;
273 secretArn: string | undefined;
274 credentials: { username: string; password: Redacted.Redacted<string> } | undefined;
275}) {
276 const identifier = options.instance.DBInstanceIdentifier ?? "unknown";
277 const target = yield* describeTarget(options.instance);
278 const transport = options.transport ?? (target.dataApiArn ? "data-api" : "direct");
279
280 if (transport === "data-api") {
281 const resourceArn = target.dataApiArn;
282 if (!resourceArn) {
283 return yield* new MigrationError({
284 message: options.instance.DBClusterIdentifier
285 ? `DB cluster '${options.instance.DBClusterIdentifier}' does not have the Data API enabled (set enableHttpEndpoint: true)`
286 : `DB instance '${identifier}' is not an Aurora cluster member; the Data API transport needs an Aurora cluster`,
287 });
288 }
289 const secretArn = options.secretArn ?? target.managedSecretArn;
290 if (!secretArn) {
291 return yield* new MigrationError({
292 message: `'${identifier}' has no RDS-managed master secret; pass migrationsSecretArn`,
293 });
294 }
295 return yield* runMigrations({
296 input: options.input,
297 stamped: options.stamped,
298 withExecutor: (apply) =>
299 makeDataApiExecutor({
300 dialect: target.dialect,
301 resourceArn,
302 secretArn,
303 database: target.database,
304 }).pipe(Effect.flatMap(apply)),
305 });
306 }
307
308 if (!target.host || !target.port) {
309 return yield* new MigrationError({ message: `'${identifier}' has no endpoint yet` });
310 }
311 if (target.dialect === "mysql" && !target.database) {
312 return yield* new MigrationError({
313 message: `'${identifier}' has no database name; set dbName (or databaseName on the cluster) to apply migrations`,
314 });
315 }
316 const secretArn =
317 options.secretArn ?? (options.credentials ? undefined : target.managedSecretArn);
318 const credentials = secretArn
319 ? yield* readSecretCredentials(secretArn)
320 : options.credentials
321 ? {
322 username: options.credentials.username,
323 password: Redacted.value(options.credentials.password),
324 }
325 : undefined;
326 if (!credentials) {
327 return yield* new MigrationError({
328 message: `'${identifier}' has no credentials for migrations; set manageMasterUserPassword, masterUserPassword, or migrationsSecretArn`,
329 });
330 }
331 const connection: MigrationConnectionTarget = {
332 dialect: target.dialect,
333 host: target.host,
334 port: target.port,
335 // PostgreSQL always has the `postgres` database.
336 database: target.database ?? (target.dialect === "postgres" ? "postgres" : undefined),
337 ...credentials,
338 };
339
340 return yield* runMigrations({
341 input: options.input,
342 stamped: options.stamped,
343 withExecutor: (apply) =>
344 transport === "direct"
345 ? withMigrationConnection(connection, apply)
346 : makeRunnerExecutor({ functionArn: transport.functionArn, target: connection }).pipe(
347 Effect.flatMap(apply),
348 ),
349 });
350});