1
import { execFile as execFileCallback, spawn, type ChildProcess } from "node:child_process";
2
import { createWriteStream } from "node:fs";
3
import { mkdtemp, rm, writeFile } from "node:fs/promises";
4
import net from "node:net";
5
import os from "node:os";
6
import path from "node:path";
7
import { setTimeout as sleep } from "node:timers/promises";
8
import { promisify } from "node:util";
9
 
10
const execFile = promisify(execFileCallback);
11
 
12
export const repoRoot = path.resolve(import.meta.dirname, "../..");
13
 
14
const viteBinPath = path.join(repoRoot, "node_modules", "vite", "bin", "vite.js");
15
const wranglerBinPath = path.join(repoRoot, "node_modules", "wrangler", "bin", "wrangler.js");
16
const d1DatabaseName = "dab-control-plane";
17
const startAttempts = 5;
18
const readyTimeoutMs = 30_000;
19
 
20
export interface IsolatedWorkerOptions {
21
  name: string;
22
  tempPrefix: string;
23
  preserveEnv: string;
24
  oidcIssuer: string;
25
  d1DatabaseId: string;
26
  r2BucketName: string;
27
  resourceSuffix: string;
28
  devVars: Record<string, string>;
29
}
30
 
31
export interface IsolatedWorker {
32
  port: number;
33
  tempDir: string;
34
}
35
 
36
export async function runIsolatedWorker(
37
  options: IsolatedWorkerOptions,
38
  callback: (worker: IsolatedWorker) => Promise<void>,
39
): Promise<void> {
40
  const tempDir = await mkdtemp(path.join(os.tmpdir(), options.tempPrefix));
41
  const wranglerConfigPath = path.join(tempDir, "wrangler.json");
42
  const devVarsPath = path.join(tempDir, ".dev.vars");
43
  const preserve = process.env[options.preserveEnv] === "1";
44
  let started: { child: ChildProcess; port: number } | undefined;
45
 
46
  await writeFile(wranglerConfigPath, `${localWranglerConfig(options)}\n`);
47
  await writeFile(devVarsPath, `${devVars(options.devVars)}\n`);
48
  await applyD1Migrations(tempDir, wranglerConfigPath);
49
 
50
  try {
51
    started = await startWorker(tempDir, wranglerConfigPath);
52
    await callback({ port: started.port, tempDir });
53
  } finally {
54
    if (started) await stop(started.child);
55
    if (!preserve) {
56
      await rm(devVarsPath, { force: true });
57
      await rm(tempDir, { recursive: true, force: true });
58
    } else {
59
      console.log(`preserved ${tempDir}`);
60
      console.log(`preserved ${devVarsPath}`);
61
      console.log(`preserved ${wranglerConfigPath}`);
62
    }
63
  }
64
}
65
 
66
function devVars(values: Record<string, string>): string {
67
  return Object.entries(values)
68
    .map(([key, value]) => `${key}=${value}`)
69
    .join("\n");
70
}
71
 
72
async function startWorker(
73
  tempDir: string,
74
  wranglerConfigPath: string,
75
): Promise<{ child: ChildProcess; port: number }> {
76
  let lastError: unknown;
77
 
78
  for (let attempt = 1; attempt <= startAttempts; attempt += 1) {
79
    const port = await allocatePort();
80
    try {
81
      const child = await startViteDev(tempDir, port, wranglerConfigPath);
82
      return { child, port };
83
    } catch (error) {
84
      lastError = error;
85
      if (!isPortInUse(error) || attempt === startAttempts) throw error;
86
    }
87
  }
88
 
89
  throw lastError instanceof Error ? lastError : new Error("Unable to start Vite dev server");
90
}
91
 
92
function allocatePort(): Promise<number> {
93
  return new Promise((resolve, reject) => {
94
    const server = net.createServer();
95
    server.on("error", reject);
96
    server.listen(0, "127.0.0.1", () => {
97
      const address = server.address();
98
      if (!address || typeof address === "string") {
99
        reject(new Error("Unable to allocate loopback port"));
100
        return;
101
      }
102
      const port = address.port;
103
      server.close(() => resolve(port));
104
    });
105
  });
106
}
107
 
108
async function startViteDev(tempDir: string, port: number, wranglerConfigPath: string): Promise<ChildProcess> {
109
  const stdout = createWriteStream(path.join(tempDir, "vite.stdout.log"));
110
  const stderr = createWriteStream(path.join(tempDir, "vite.stderr.log"));
111
  const child = spawn(
112
    process.execPath,
113
    [viteBinPath, "dev", "--host", "127.0.0.1", "--port", String(port), "--strictPort"],
114
    {
115
      cwd: repoRoot,
116
      env: {
117
        ...process.env,
118
        DAB_PERSIST_STATE_PATH: tempDir,
119
        DAB_WRANGLER_CONFIG_PATH: wranglerConfigPath,
120
      },
121
      stdio: ["ignore", "pipe", "pipe"],
122
      detached: process.platform !== "win32",
123
    },
124
  );
125
 
126
  child.stdout?.pipe(stdout);
127
  child.stderr?.pipe(stderr);
128
  child.once("close", () => {
129
    stdout.close();
130
    stderr.close();
131
  });
132
 
133
  try {
134
    await waitForHealth(port, child);
135
    return child;
136
  } catch (error) {
137
    await stop(child);
138
    throw error;
139
  }
140
}
141
 
142
async function waitForHealth(port: number, child: ChildProcess): Promise<void> {
143
  const deadline = Date.now() + readyTimeoutMs;
144
  let lastError: unknown;
145
 
146
  while (Date.now() < deadline) {
147
    if (child.exitCode !== null) throw new Error(`Vite exited before readiness with code ${child.exitCode}`);
148
 
149
    try {
150
      const response = await fetch(`http://dab.localhost:${port}/healthz`);
151
      if (response.status === 200) return;
152
      lastError = new Error(`healthz returned ${response.status}`);
153
    } catch (error) {
154
      lastError = error;
155
    }
156
 
157
    await sleep(250);
158
  }
159
 
160
  throw lastError instanceof Error ? lastError : new Error("Timed out waiting for /healthz");
161
}
162
 
163
async function applyD1Migrations(persistTo: string, configPath: string): Promise<void> {
164
  await execFile(
165
    process.execPath,
166
    [
167
      wranglerBinPath,
168
      "d1",
169
      "migrations",
170
      "apply",
171
      d1DatabaseName,
172
      "--local",
173
      "--persist-to",
174
      persistTo,
175
      "--config",
176
      configPath,
177
    ],
178
    {
179
      cwd: repoRoot,
180
      env: { ...process.env, CI: "1", NO_D1_WARNING: "true" },
181
      timeout: 180_000,
182
      maxBuffer: 10 * 1024 * 1024,
183
    },
184
  );
185
}
186
 
187
async function stop(child: ChildProcess): Promise<void> {
188
  if (child.exitCode !== null) return;
189
 
190
  const kill = (signal: NodeJS.Signals): void => {
191
    if (child.pid === undefined) return;
192
    if (process.platform === "win32") {
193
      child.kill(signal);
194
      return;
195
    }
196
    process.kill(-child.pid, signal);
197
  };
198
 
199
  kill("SIGTERM");
200
  await sleep(1_000);
201
  if (child.exitCode === null) kill("SIGKILL");
202
}
203
 
204
function isPortInUse(error: unknown): boolean {
205
  const message = error instanceof Error ? error.message : String(error);
206
  return message.includes("is already in use") || message.includes("EADDRINUSE");
207
}
208
 
209
function localWranglerConfig(options: IsolatedWorkerOptions): string {
210
  return JSON.stringify(
211
    {
212
      name: options.name,
213
      main: path.join(repoRoot, "src/worker/index.ts"),
214
      compatibility_date: "2026-05-20",
215
      compatibility_flags: ["nodejs_als"],
216
      vars: {
217
        LOG_LEVEL: "warn",
218
        CONTROL_PLANE_HOST: "dab.localhost",
219
        SUBJECT_HOST_SUFFIX: ".dab.localhost",
220
        TESSERA_OIDC_ISSUER: options.oidcIssuer,
221
        MAX_FILE_BYTES: "100000000",
222
        MAX_XML_BODY_BYTES: "1048576",
223
        MAX_CALENDAR_OBJECT_BYTES: "1048576",
224
        MAX_ADDRESS_OBJECT_BYTES: "1048576",
225
        MAX_DAV_DEPTH_INFINITY_NODES: "10000",
226
        MAX_REPORT_RESULTS: "5000",
227
        MAX_RECURRENCE_YEARS: "5",
228
        MAX_RECURRENCE_INSTANCES: "10000",
229
        DAB_ENABLE_TEST_FIXTURES: "1",
230
      },
231
      secrets: {
232
        required: ["TESSERA_OIDC_CLIENT_ID", "TESSERA_OIDC_CLIENT_SECRET", "DAB_SESSION_SECRET"],
233
      },
234
      d1_databases: [
235
        {
236
          binding: "DAV_CONTROL_PLANE",
237
          database_name: d1DatabaseName,
238
          database_id: options.d1DatabaseId,
239
          migrations_dir: path.join(repoRoot, "drizzle/d1"),
240
        },
241
      ],
242
      r2_buckets: [{ binding: "FILE_BLOBS", bucket_name: options.r2BucketName }],
243
      durable_objects: {
244
        bindings: [
245
          { name: "AUTH", class_name: "AuthObject" },
246
          { name: "FILE_DAV", class_name: "FileDavObject" },
247
          { name: "CAL_DAV", class_name: "CalDavObject" },
248
          { name: "CARD_DAV", class_name: "CardDavObject" },
249
        ],
250
      },
251
      migrations: [
252
        {
253
          tag: "v1",
254
          new_sqlite_classes: ["AuthObject", "FileDavObject", "CalDavObject", "CardDavObject"],
255
        },
256
      ],
257
      queues: {
258
        producers: [
259
          { binding: "BLOB_GC", queue: `dab-blob-gc-${options.resourceSuffix}` },
260
          { binding: "REPAIR_JOBS", queue: `dab-repair-jobs-${options.resourceSuffix}` },
261
        ],
262
        consumers: [
263
          { queue: `dab-blob-gc-${options.resourceSuffix}`, max_batch_size: 10, max_batch_timeout: 5 },
264
          { queue: `dab-repair-jobs-${options.resourceSuffix}`, max_batch_size: 5, max_batch_timeout: 5 },
265
        ],
266
      },
267
      ratelimits: [
268
        { name: "RL_AUTH", namespace_id: "83800", simple: { limit: 10, period: 60 } },
269
        { name: "RL_DAV_AUTH", namespace_id: "83801", simple: { limit: 30, period: 60 } },
270
        { name: "RL_REPORT", namespace_id: "83802", simple: { limit: 60, period: 60 } },
271
      ],
272
    },
273
    null,
274
    2,
275
  );
276
}