1import {
2 Credentials,
3 formatHeaders,
4} from "@distilled.cloud/cloudflare/Credentials";
5import * as d1 from "@distilled.cloud/cloudflare/d1";
6import * as Effect from "effect/Effect";
7import * as HttpClient from "effect/http/HttpClient";
8import * as HttpClientRequest from "effect/http/HttpClientRequest";
9import { md5 } from "../../Util/md5.ts";
10
11export interface ImportDatabaseOptions {
12 accountId: string;
13 databaseId: string;
14 sqlData: string | Uint8Array;
15 filename?: string;
16}
17
18export interface ImportDatabaseResult {
19 filename: string;
20 numQueries: number;
21}
22
23interface ImportPollingResponse {
24 at_bookmark?: string | null;
25 error?: string | null;
26 filename?: string | null;
27 messages?: string[] | null;
28 result?: {
29 final_bookmark?: string | null;
30 num_queries?: number | null;
31 } | null;
32 status?: "complete" | "error" | null;
33 success?: boolean | null;
34 upload_url?: string | null;
35}
36
37/**
38 * Import SQL into a D1 database via the multi-step Cloudflare flow:
39 * 1. POST `action: "init"` -> returns presigned `upload_url`
40 * 2. PUT raw SQL to `upload_url`
41 * 3. POST `action: "ingest"` -> returns `at_bookmark`
42 * 4. POST `action: "poll"` until `status === "complete"`
43 */
44export const importD1Database = (options: ImportDatabaseOptions) =>
45 Effect.gen(function* () {
46 const credentialsEff = yield* Credentials;
47 const credentials = yield* credentialsEff;
48 const authHeaders = formatHeaders(credentials);
49 const url = importEndpoint(
50 credentials.apiBaseUrl,
51 options.accountId,
52 options.databaseId,
53 );
54 const client = yield* HttpClient.HttpClient;
55
56 const postJson = (
57 body: unknown,
58 ): Effect.Effect<ImportPollingResponse, never, never> =>
59 Effect.gen(function* () {
60 const req = HttpClientRequest.post(url).pipe(
61 HttpClientRequest.setHeaders(authHeaders),
62 HttpClientRequest.bodyJsonUnsafe(body),
63 );
64 const res = yield* client.execute(req).pipe(Effect.orDie);
65 if (res.status < 200 || res.status >= 300) {
66 const text = yield* res.text.pipe(Effect.orElseSucceed(() => ""));
67 return yield* Effect.die(
68 `D1 import request failed (${res.status}): ${text}`,
69 );
70 }
71 const text = yield* res.text.pipe(Effect.orDie);
72 const json = JSON.parse(text) as { result: ImportPollingResponse };
73 return json.result;
74 });
75
76 const etag = yield* md5(options.sqlData);
77
78 // Step 1: init via the typed d1 client
79 const initDb = yield* d1.importDatabase;
80 const init = yield* initDb({
81 accountId: options.accountId,
82 databaseId: options.databaseId,
83 action: "init",
84 etag,
85 });
86
87 if (!init.uploadUrl) {
88 return yield* Effect.die(
89 init.error ?? "Failed to get upload URL for D1 import",
90 );
91 }
92 const uploadFilename = options.filename ?? init.filename ?? "import.sql";
93
94 // Step 2: PUT to the presigned upload URL
95 const bytes =
96 typeof options.sqlData === "string"
97 ? new TextEncoder().encode(options.sqlData)
98 : options.sqlData;
99 const putReq = HttpClientRequest.put(init.uploadUrl).pipe(
100 HttpClientRequest.bodyUint8Array(bytes, "application/sql"),
101 );
102 const putRes = yield* client.execute(putReq).pipe(Effect.orDie);
103 if (putRes.status < 200 || putRes.status >= 300) {
104 const text = yield* putRes.text.pipe(Effect.orElseSucceed(() => ""));
105 return yield* Effect.die(
106 `Failed to upload SQL file to D1 (${putRes.status}): ${text}`,
107 );
108 }
109
110 // Step 3: ingest
111 const ingest = yield* postJson({
112 action: "ingest",
113 etag,
114 filename: init.filename,
115 });
116 if (!ingest.at_bookmark) {
117 return yield* Effect.die(
118 ingest.error ?? "Ingest response missing bookmark",
119 );
120 }
121
122 // Step 4: poll until complete
123 const poll = (
124 bookmark: string,
125 ): Effect.Effect<ImportDatabaseResult, never, never> =>
126 Effect.gen(function* () {
127 const data = yield* postJson({
128 action: "poll",
129 current_bookmark: bookmark,
130 });
131 if (data.status === "complete" && data.result) {
132 return {
133 filename: data.filename ?? uploadFilename,
134 numQueries: data.result.num_queries ?? 0,
135 };
136 }
137 if (data.status === "error") {
138 return yield* Effect.die(data.error ?? "Error during D1 import");
139 }
140 if (!data.at_bookmark) {
141 return yield* Effect.die("D1 import poll missing bookmark");
142 }
143 return yield* poll(data.at_bookmark);
144 });
145
146 return yield* poll(ingest.at_bookmark);
147 });
148
149const importEndpoint = (
150 apiBaseUrl: string,
151 accountId: string,
152 databaseId: string,
153) => `${apiBaseUrl}/accounts/${accountId}/d1/database/${databaseId}/import`;