1import * as Effect from "effect/Effect";
2import {
3 convertedRowInsertSql,
4 findForeignHistory,
5 matchForeignRows,
6 toTimestampString,
7} from "./Convert.ts";
8import {
9 MigrationError,
10 MigrationHistoryConflictError,
11 type MigrationDialect,
12 type MigrationRecord,
13 type SqlExecutor,
14} from "./Format.ts";
15import { classifyTable, tableColumns } from "./Introspect.ts";
16import { quoteIdentifier, sqlLiteral } from "./Utils.ts";
17
18export const ALCHEMY_DEFAULT_TABLE = "__alchemy_migrations";
19
20/**
21 * THE Alchemy applied-migrations table: `id, hash, created_at, name,
22 * applied_at`, name-keyed detection. This is deliberately drizzle's column
23 * shape — a database whose history lives in `__drizzle_migrations` adopts
24 * with a verbatim row copy — but Alchemy owns the table and it is the only
25 * format Alchemy ever writes. Migrating from drizzle/prisma/wrangler
26 * bookkeeping is a one-way conversion performed once (see `Convert.ts`).
27 */
28const createTableSql = (table: string, dialect: MigrationDialect): string => {
29 const quoted = quoteIdentifier(table, dialect);
30 switch (dialect) {
31 case "sqlite":
32 return `CREATE TABLE IF NOT EXISTS ${quoted} (
33 id INTEGER PRIMARY KEY,
34 hash text NOT NULL,
35 created_at numeric,
36 name text,
37 applied_at TEXT
38);`;
39 case "postgres":
40 return `CREATE TABLE IF NOT EXISTS ${quoted} (
41 id SERIAL PRIMARY KEY,
42 hash text NOT NULL,
43 created_at bigint,
44 name text,
45 applied_at timestamp with time zone DEFAULT now()
46);`;
47 case "mysql":
48 return `CREATE TABLE IF NOT EXISTS ${quoted} (
49 id SERIAL PRIMARY KEY,
50 hash TEXT NOT NULL,
51 created_at BIGINT,
52 name TEXT,
53 applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
54);`;
55 }
56};
57
58const insertSql = (
59 table: string,
60 dialect: MigrationDialect,
61 record: Pick<MigrationRecord, "name" | "hash" | "createdAtMillis">,
62): string => {
63 const quoted = quoteIdentifier(table, dialect);
64 const applied = dialect === "sqlite" ? ", datetime('now')" : "";
65 const appliedColumn = dialect === "sqlite" ? ", applied_at" : "";
66 return `INSERT INTO ${quoted} (hash, created_at, name${appliedColumn}) VALUES (${sqlLiteral(record.hash)}, ${sqlLiteral(record.createdAtMillis ?? null)}, ${sqlLiteral(record.name)}${applied});`;
67};
68
69const renameSql = (
70 from: string,
71 to: string,
72 dialect: MigrationDialect,
73): string =>
74 dialect === "mysql"
75 ? `RENAME TABLE ${quoteIdentifier(from, dialect)} TO ${quoteIdentifier(to, dialect)};`
76 : `ALTER TABLE ${quoteIdentifier(from, dialect)} RENAME TO ${quoteIdentifier(to, dialect)};`;
77
78/**
79 * Rebuild an in-place table (legacy Alchemy 3-column / oldest 2-column /
80 * wrangler-shaped) into the Alchemy shape, backfilling `hash` and
81 * `created_at` from local records matched by name. A recorded row with no
82 * matching local file is a hard error — mirrors the same rule conversion
83 * applies (see `matchForeignRows`).
84 */
85const rebuildInPlace = (options: {
86 executor: SqlExecutor;
87 table: string;
88 records: ReadonlyArray<MigrationRecord>;
89 /** SQL expression yielding the migration name from the old table. */
90 nameExpr: string;
91 tool: "legacy-alchemy" | "wrangler";
92}) =>
93 Effect.gen(function* () {
94 const { executor, table, records, nameExpr } = options;
95 const dialect = executor.dialect;
96 const quoted = quoteIdentifier(table, dialect);
97 const rows = yield* executor.query(
98 `SELECT ${nameExpr} AS name, applied_at FROM ${quoted} ORDER BY id;`,
99 );
100 const matched = yield* matchForeignRows({
101 history: {
102 tool: options.tool,
103 source: table,
104 rows: rows
105 .filter((row) => row.name !== null && row.name !== undefined)
106 .map((row) => ({
107 name: String(row.name),
108 hash: undefined,
109 createdAtMillis: undefined,
110 appliedAt: toTimestampString(row.applied_at),
111 })),
112 },
113 records,
114 });
115
116 const temp = `${table}_alchemy_upgrade`;
117 yield* executor.batch([
118 `DROP TABLE IF EXISTS ${quoteIdentifier(temp, dialect)};`,
119 createTableSql(temp, dialect).replace(
120 "CREATE TABLE IF NOT EXISTS",
121 "CREATE TABLE",
122 ),
123 ...matched.map((row) => convertedRowInsertSql(temp, dialect, row)),
124 `DROP TABLE ${quoted};`,
125 renameSql(temp, table, dialect),
126 ]);
127 });
128
129const ensureTable = (options: {
130 executor: SqlExecutor;
131 table: string;
132 records: ReadonlyArray<MigrationRecord>;
133}) =>
134 Effect.gen(function* () {
135 const { executor, table, records } = options;
136 const shape = classifyTable(yield* tableColumns(executor, table));
137 switch (shape) {
138 case "absent": {
139 // Greenfield for us — but possibly not for the database. Adopt any
140 // history the previous tool (drizzle-kit / prisma / wrangler) left
141 // behind: copy it into our table ONCE and freeze theirs. One-way.
142 const history = yield* findForeignHistory({ executor, table });
143 const converted = history
144 ? yield* matchForeignRows({ history, records })
145 : [];
146 yield* executor.batch([
147 createTableSql(table, executor.dialect),
148 ...converted.map((row) =>
149 convertedRowInsertSql(table, executor.dialect, row),
150 ),
151 ]);
152 return;
153 }
154 case "drizzle-shaped":
155 // Already our column shape (shared with drizzle v1) — including
156 // the case where the user pointed `table` straight at an existing
157 // `__drizzle_migrations`. Adopt in place.
158 return;
159 case "legacy-alchemy":
160 yield* rebuildInPlace({
161 executor,
162 table,
163 records,
164 nameExpr: "name",
165 tool: "legacy-alchemy",
166 });
167 return;
168 case "legacy-2col":
169 yield* rebuildInPlace({
170 executor,
171 table,
172 records,
173 nameExpr: "id",
174 tool: "legacy-alchemy",
175 });
176 return;
177 case "wrangler":
178 // A wrangler table at OUR resolved table name (e.g. legacy D1
179 // state pinned to `d1_migrations`, or a wrangler user's table
180 // adopted under an explicit `table:`): convert it in place.
181 yield* rebuildInPlace({
182 executor,
183 table,
184 records,
185 nameExpr: "name",
186 tool: "wrangler",
187 });
188 return;
189 case "unknown":
190 return yield* new MigrationError({
191 message:
192 `Migrations table "${table}" has an unrecognized column layout; ` +
193 `refusing to write bookkeeping into it.`,
194 });
195 }
196 });
197
198const appliedNames = (executor: SqlExecutor, table: string) =>
199 executor
200 .query(`SELECT name FROM ${quoteIdentifier(table, executor.dialect)};`)
201 .pipe(
202 Effect.map(
203 (rows) =>
204 new Set(
205 rows
206 .map((row) => row.name)
207 .filter((name) => name !== null && name !== undefined)
208 .map(String),
209 ),
210 ),
211 );
212
213/**
214 * Apply pending migrations with Alchemy's bookkeeping. Idempotent: each
215 * migration's statements and its bookkeeping INSERT go through
216 * `executor.batch` as one unit (a transaction on pg/mysql, one batched
217 * query on D1, which has no transactions over HTTP).
218 *
219 * Applied-detection is name-keyed with layout aliasing: pre-registry
220 * Alchemy recorded drizzle-layout migrations under `<dir>/migration.sql`
221 * while current records key them by `<dir>`, so both keys are honored.
222 */
223export const applyAlchemyFormat = (options: {
224 executor: SqlExecutor;
225 table: string;
226 records: ReadonlyArray<MigrationRecord>;
227}): Effect.Effect<void, MigrationError | MigrationHistoryConflictError> =>
228 Effect.gen(function* () {
229 const { executor, table, records } = options;
230 yield* ensureTable({ executor, table, records });
231 const applied = yield* appliedNames(executor, table);
232 for (const record of records) {
233 if (
234 applied.has(record.name) ||
235 applied.has(`${record.name}/migration.sql`) ||
236 applied.has(record.name.replace(/\/migration\.sql$/, ""))
237 ) {
238 continue;
239 }
240 yield* executor.batch([
241 ...record.statements,
242 insertSql(table, executor.dialect, record),
243 ]);
244 }
245 });