History

Move Google and Microsoft onto the OpenAPI plugin (#1366)

* Add openapi provider catalog design notes * Add OpenAPI spec format adapter support * Add Google and Microsoft spec format adapters * Preserve streaming spec format refresh * Move provider add surfaces into catalog presets * Fix provider catalog health checks * Fix Google Directory health check args * Format design notes * Add openapi provider service split migration * Fix service split blob migration * Read all provider operation rows for the migration plan * Assign provider operations by fresh-add spec ownership * Fix Microsoft service split ownership * Map singular mail folder and users profile segments * Seed add form identity from the catalog preset name * Use OIDC id token for OAuth identity labels * Handle OAuth health without probes * Canonicalize Microsoft scopes and decode JWT claims as UTF-8 * Update lockfile for provider package dependency changes * Package provider service split migration * Retarget orphan guardrail policies as dormant rows in the cloud runner * Retarget orphan approve policies too * Fold provider catalogs into openapi

Rhys Sullivan committed 4 months ago

Commit 55a24631 parent 93b000a148 files · +8259 −6034Browse files

Showing the changes to scripts/migration/service-split-postgres-cli.ts (1 of 148 files). Show all

Unified · Split

scripts/migration/service-split-postgres-cli.ts+639−0
@@ -0,0 +1,639 @@
1/* oxlint-disable executor/no-error-constructor, executor/no-try-catch-or-throw -- boundary: one-shot migration CLI */
2
3import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
4import { resolve } from "node:path";
5import { pathToFileURL } from "node:url";
6
7import postgres from "../../apps/cloud/node_modules/postgres/src/index.js";
8
9import {
10 operationStorageKey,
11 planMigration,
12 renderOrgDiff,
13 renderSummary,
14 storageDataRecord,
15 tenantHash,
16 verifyPolicyRewriteNeverWidens,
17 type BlobRow,
18 type ConnectionRow,
19 type IntegrationRow,
20 type MigrationInput,
21 type OrgPlan,
22 type PluginStorageRow,
23 type ToolPolicyRow,
24 type ToolRow,
25} from "@executor-js/plugin-provider-service-split/planner";
26
27export interface SqlClient {
28 readonly unsafe: <T extends readonly object[] = readonly Record<string, unknown>[]>(
29 query: string,
30 params?: readonly unknown[],
31 ) => Promise<T>;
32 readonly begin: <T>(fn: (sql: SqlClient) => Promise<T>) => Promise<T>;
33 readonly end: (options?: { readonly timeout?: number }) => Promise<void>;
34}
35
36const args = process.argv.slice(2);
37const hasArg = (name: string): boolean => args.includes(name);
38const argValue = (name: string): string | undefined => {
39 const index = args.indexOf(name);
40 return index >= 0 ? args[index + 1] : undefined;
41};
42
43const apply = hasArg("--apply");
44const inputJson = argValue("--input-json");
45const outputDir = resolve(
46 argValue("--output-dir") ?? "~/agent-workspace/scratch/service-split-migration-dryrun",
47).replace(/^~(?=\/)/, process.env.HOME ?? "~");
48const orgFilter = argValue("--org");
49const trafficLastTenant = argValue("--traffic-last-tenant");
50const databaseUrl = argValue("--database-url") ?? process.env.DATABASE_URL;
51const r2CopiesDone = hasArg("--r2-copies-done");
52export const LEDGER_TABLE = "provider_service_split_org_migration";
53
54const parseInputJson = (path: string): MigrationInput =>
55 JSON.parse(readFileSync(path, "utf8")) as MigrationInput;
56
57const readRows = async <T extends readonly object[]>(sql: SqlClient, query: string): Promise<T> =>
58 sql.unsafe<T>(query);
59
60const readCompletedTenants = async (
61 sql: SqlClient,
62 dryRun: boolean,
63): Promise<readonly string[]> => {
64 const [table] = await sql.unsafe<{ ledger_table: string | null }[]>(
65 `SELECT to_regclass('public.${LEDGER_TABLE}')::text AS ledger_table`,
66 );
67 if (!table?.ledger_table) {
68 if (dryRun) return [];
69 await sql.unsafe(`
70 CREATE TABLE IF NOT EXISTS ${LEDGER_TABLE} (
71 tenant text PRIMARY KEY,
72 time_completed bigint NOT NULL
73 )
74 `);
75 return [];
76 }
77 const rows = await sql.unsafe<{ tenant: string }[]>(
78 `SELECT tenant FROM ${LEDGER_TABLE} ORDER BY tenant`,
79 );
80 return rows.map((row) => row.tenant);
81};
82
83const filterInputByOrg = (input: MigrationInput, filter: string | undefined): MigrationInput => {
84 if (!filter) return input;
85 const tenantMatches = new Set(
86 input.integrations
87 .map((row) => row.tenant)
88 .filter((tenant) => tenant === filter || tenantHash(tenant) === filter),
89 );
90 return {
91 ...input,
92 integrations: input.integrations.filter((row) => tenantMatches.has(row.tenant)),
93 connections: input.connections.filter((row) => tenantMatches.has(row.tenant)),
94 tools: input.tools.filter((row) => tenantMatches.has(row.tenant)),
95 pluginStorage: input.pluginStorage?.filter((row) => tenantMatches.has(row.tenant)),
96 blobs: input.blobs?.filter((row) => {
97 for (const tenant of tenantMatches) {
98 if (row.namespace.startsWith(`o:${tenant}/`)) return true;
99 }
100 return false;
101 }),
102 policies: input.policies.filter((row) => tenantMatches.has(row.tenant)),
103 completedTenants: input.completedTenants?.filter((tenant) => tenantMatches.has(tenant)),
104 };
105};
106
107export const readDatabaseInput = async (
108 sql: SqlClient,
109 dryRun: boolean,
110 blobBackend: MigrationInput["blobBackend"],
111): Promise<MigrationInput> => {
112 const completedTenants = await readCompletedTenants(sql, dryRun);
113 const integrations = await readRows<IntegrationRow[]>(
114 sql,
115 `
116 SELECT tenant, slug, plugin_id, name, description, config, health_check, config_revised_at,
117 can_remove, can_refresh, created_at::text, updated_at::text, row_id
118 FROM integration
119 WHERE plugin_id IN ('google', 'microsoft')
120 OR slug IN ('google', 'microsoft')
121 OR tenant IN (
122 SELECT tenant FROM integration
123 WHERE plugin_id IN ('google', 'microsoft')
124 )
125 ORDER BY tenant, slug
126 `,
127 );
128 const connections = await readRows<ConnectionRow[]>(
129 sql,
130 `
131 SELECT tenant, owner, subject, integration, name, template, provider, item_ids,
132 identity_label, description, last_health, tools_synced_at, oauth_client,
133 oauth_client_owner, refresh_item_id, expires_at, oauth_scope, oauth_token_url,
134 provider_state, created_at::text, updated_at::text, row_id
135 FROM connection
136 WHERE tenant IN (
137 SELECT tenant FROM integration
138 WHERE plugin_id IN ('google', 'microsoft')
139 )
140 ORDER BY tenant, integration, owner, subject, name
141 `,
142 );
143 const tools = await readRows<ToolRow[]>(
144 sql,
145 `
146 SELECT tenant, owner, subject, integration, connection, plugin_id, name, description,
147 input_schema, output_schema, annotations, created_at::text, updated_at::text, row_id
148 FROM tool
149 WHERE tenant IN (
150 SELECT tenant FROM integration
151 WHERE plugin_id IN ('google', 'microsoft')
152 )
153 ORDER BY tenant, integration, connection, name
154 `,
155 );
156 const pluginStorage = await sql.unsafe<PluginStorageRow[]>(
157 `
158 SELECT tenant, owner, subject, plugin_id, collection, key, data,
159 created_at::text, updated_at::text, row_id
160 FROM plugin_storage
161 WHERE tenant IN (
162 SELECT tenant FROM integration
163 WHERE plugin_id IN ('google', 'microsoft')
164 )
165 AND plugin_id IN ('google', 'microsoft')
166 AND collection = 'operation'
167 ORDER BY tenant, plugin_id, collection, key
168 `,
169 );
170 const blobs = await readRows<BlobRow[]>(
171 sql,
172 `
173 SELECT id, namespace, key
174 FROM blob
175 WHERE namespace IN (
176 SELECT 'o:' || tenant || '/' || plugin_id
177 FROM integration
178 WHERE plugin_id IN ('google', 'microsoft')
179 )
180 AND (key LIKE 'spec/%' OR key LIKE 'defs/%')
181 ORDER BY namespace, key
182 `,
183 );
184 const policies = await readRows<ToolPolicyRow[]>(
185 sql,
186 `
187 SELECT tenant, owner, subject, id, pattern, action, position, created_at::text,
188 updated_at::text, row_id
189 FROM tool_policy
190 WHERE tenant IN (
191 SELECT tenant FROM integration
192 WHERE plugin_id IN ('google', 'microsoft')
193 )
194 ORDER BY tenant, owner, subject, position, id
195 `,
196 );
197 return {
198 integrations,
199 connections,
200 tools,
201 pluginStorage,
202 blobs,
203 policies,
204 completedTenants,
205 trafficLastTenant,
206 blobBackend,
207 assumeExternalBlobSourcePresent: blobBackend === "external",
208 };
209};
210
211const scrubJson = (value: unknown): unknown => {
212 if (Array.isArray(value)) return value.map(scrubJson);
213 if (value && typeof value === "object") {
214 return Object.fromEntries(
215 Object.entries(value)
216 .filter(([, inner]) => inner !== undefined)
217 .map(([key, inner]) => [key, scrubJson(inner)]),
218 );
219 }
220 return value;
221};
222
223const stableId = (...parts: readonly string[]): string => {
224 const text = parts.join("\u0000");
225 let hash = 0;
226 for (let index = 0; index < text.length; index += 1) {
227 hash = (hash * 31 + text.charCodeAt(index)) >>> 0;
228 }
229 return `service_split_${hash.toString(36)}`;
230};
231
232const operationKeyPrefix = (integration: string): string =>
233 `${operationStorageKey(integration, "").split(".").slice(0, 2).join(".")}.%`;
234
235export const applyOrg = async (sql: SqlClient, org: OrgPlan): Promise<void> => {
236 if (org.completed) return;
237 if (org.hardErrors.length > 0) {
238 throw new Error(`org ${org.tenantHash} has unresolved hard errors; apply is blocked`);
239 }
240 await sql.begin(async (tx) => {
241 const now = new Date();
242 for (const copy of org.blobCopies.filter((item) => item.backend === "database")) {
243 const [source] = await tx.unsafe<{ readonly value: string }[]>(
244 "SELECT value FROM blob WHERE namespace = $1 AND key = $2 LIMIT 1",
245 [copy.sourceNamespace, copy.key],
246 );
247 if (!source) {
248 throw new Error(`Missing source blob ${copy.sourceNamespace}/${copy.key}`);
249 }
250 await tx.unsafe(
251 `
252 INSERT INTO blob (namespace, key, value, row_id, id)
253 VALUES ($1, $2, $3, $4, $5)
254 ON CONFLICT (id) DO NOTHING
255 `,
256 [
257 copy.targetNamespace,
258 copy.key,
259 source.value,
260 stableId("blob", org.tenant, copy.targetNamespace, copy.key),
261 JSON.stringify([copy.targetNamespace, copy.key]),
262 ],
263 );
264 }
265
266 for (const row of org.integrations.filter((item) => item.action === "create")) {
267 await tx.unsafe(
268 `
269 INSERT INTO integration (
270 slug, plugin_id, name, description, config, health_check, config_revised_at,
271 can_remove, can_refresh, created_at, updated_at, row_id, tenant
272 )
273 VALUES ($1, $2, $3, $4, $5::json, $6::json, NULL, true, true, $7, $7, $8, $9)
274 ON CONFLICT (tenant, slug) DO NOTHING
275 `,
276 [
277 row.target.slug,
278 row.target.pluginId,
279 row.target.name,
280 row.target.description,
281 JSON.stringify(scrubJson(row.config)),
282 row.healthCheck ? JSON.stringify(scrubJson(row.healthCheck)) : null,
283 now,
284 stableId("integration", org.tenant, row.target.slug),
285 org.tenant,
286 ],
287 );
288 }
289
290 for (const row of org.connections.filter((item) => item.action === "clone")) {
291 await tx.unsafe(
292 `
293 INSERT INTO connection (
294 integration, name, template, provider, item_ids, identity_label, description,
295 last_health, tools_synced_at, oauth_client, oauth_client_owner, refresh_item_id,
296 expires_at, oauth_scope, oauth_token_url, provider_state, created_at, updated_at,
297 row_id, tenant, owner, subject
298 )
299 SELECT $1, name, template, provider, item_ids, identity_label, description,
300 NULL, NULL, oauth_client, oauth_client_owner, refresh_item_id, expires_at,
301 oauth_scope, oauth_token_url, provider_state, created_at, $2, $3, tenant, owner, subject
302 FROM connection
303 WHERE tenant = $4 AND owner = $5 AND subject = $6 AND integration = $7 AND name = $8
304 ON CONFLICT (tenant, owner, subject, integration, name) DO NOTHING
305 `,
306 [
307 row.targetIntegration,
308 now,
309 stableId(
310 "connection",
311 org.tenant,
312 row.source.owner,
313 row.source.subject,
314 row.targetIntegration,
315 row.source.name,
316 ),
317 org.tenant,
318 row.source.owner,
319 row.source.subject,
320 row.source.integration,
321 row.source.name,
322 ],
323 );
324 }
325
326 for (const integration of org.integrations) {
327 for (const contribution of integration.sourceContributions) {
328 for (const toolName of contribution.operationToolNames) {
329 const [operation] = await tx.unsafe<
330 {
331 readonly data: unknown;
332 readonly created_at: string;
333 readonly updated_at: string;
334 }[]
335 >(
336 `
337 SELECT data, created_at::text, updated_at::text
338 FROM plugin_storage
339 WHERE tenant = $1
340 AND owner = 'org'
341 AND subject = ''
342 AND plugin_id = $2
343 AND collection = 'operation'
344 AND (
345 key = $3
346 OR (
347 data::jsonb ->> 'integration' = $4
348 AND data::jsonb ->> 'toolName' = $5
349 )
350 )
351 LIMIT 1
352 `,
353 [
354 org.tenant,
355 contribution.source.plugin_id,
356 operationStorageKey(contribution.source.slug, toolName),
357 contribution.source.slug,
358 toolName,
359 ],
360 );
361 if (!operation) {
362 throw new Error(
363 `Missing operation row for ${tenantHash(org.tenant)}/${contribution.source.slug}/${toolName}`,
364 );
365 }
366 await tx.unsafe(
367 `
368 INSERT INTO plugin_storage (
369 plugin_id, collection, key, data, created_at, updated_at, row_id,
370 tenant, owner, subject
371 )
372 VALUES ($1, 'operation', $2, $3::json, $4, $5, $6, $7, 'org', '')
373 ON CONFLICT (tenant, owner, subject, plugin_id, collection, key)
374 DO UPDATE SET data = EXCLUDED.data, updated_at = EXCLUDED.updated_at
375 `,
376 [
377 integration.target.pluginId,
378 operationStorageKey(integration.target.slug, toolName),
379 JSON.stringify(
380 scrubJson({
381 ...storageDataRecord(operation),
382 integration: integration.target.slug,
383 toolName,
384 }),
385 ),
386 operation.created_at,
387 now,
388 stableId("operation", org.tenant, integration.target.slug, toolName),
389 org.tenant,
390 ],
391 );
392 await tx.unsafe(
393 `
394 INSERT INTO tool (
395 integration, connection, plugin_id, name, description, input_schema, output_schema,
396 annotations, created_at, updated_at, row_id, tenant, owner, subject
397 )
398 SELECT $1, connection, $2, name, description, input_schema, output_schema,
399 annotations, created_at, $3, concat($4::text, '_', row_id), tenant, owner, subject
400 FROM tool
401 WHERE tenant = $5 AND integration = $6 AND name = $7
402 ON CONFLICT (tenant, owner, subject, integration, connection, name) DO NOTHING
403 `,
404 [
405 integration.target.slug,
406 integration.target.pluginId,
407 now,
408 stableId("tool", org.tenant, integration.target.slug),
409 org.tenant,
410 contribution.source.slug,
411 toolName,
412 ],
413 );
414 }
415 }
416 }
417
418 for (const policy of org.policies.filter((item) => item.action === "rewrite")) {
419 await tx.unsafe(
420 "DELETE FROM tool_policy WHERE tenant = $1 AND owner = $2 AND subject = $3 AND id = $4",
421 [org.tenant, policy.policy.owner, policy.policy.subject, policy.policy.id],
422 );
423 for (const [index, pattern] of policy.afterPatterns.entries()) {
424 await tx.unsafe(
425 `
426 INSERT INTO tool_policy (
427 id, pattern, action, position, created_at, updated_at, row_id, tenant, owner, subject
428 )
429 VALUES ($1, $2, $3, $4, $5, $5, $6, $7, $8, $9)
430 ON CONFLICT (tenant, owner, subject, id) DO NOTHING
431 `,
432 [
433 policy.afterPatterns.length === 1
434 ? policy.policy.id
435 : `${policy.policy.id}_${index + 1}`,
436 pattern,
437 policy.policy.action,
438 policy.policy.position,
439 now,
440 stableId("policy", org.tenant, policy.policy.id, pattern),
441 org.tenant,
442 policy.policy.owner,
443 policy.policy.subject,
444 ],
445 );
446 }
447 }
448
449 for (const monolith of org.deleteMonoliths) {
450 await tx.unsafe("DELETE FROM tool WHERE tenant = $1 AND integration = $2", [
451 org.tenant,
452 monolith.slug,
453 ]);
454 await tx.unsafe("DELETE FROM connection WHERE tenant = $1 AND integration = $2", [
455 org.tenant,
456 monolith.slug,
457 ]);
458 await tx.unsafe(
459 "DELETE FROM integration WHERE tenant = $1 AND slug = $2 AND plugin_id = $3",
460 [org.tenant, monolith.slug, monolith.plugin_id],
461 );
462 await tx.unsafe(
463 `
464 DELETE FROM plugin_storage
465 WHERE tenant = $1
466 AND owner = 'org'
467 AND subject = ''
468 AND plugin_id = $2
469 AND collection = 'operation'
470 AND (
471 data::jsonb ->> 'integration' = $3
472 OR key LIKE $4
473 )
474 `,
475 [org.tenant, monolith.plugin_id, monolith.slug, operationKeyPrefix(monolith.slug)],
476 );
477 }
478
479 await tx.unsafe(
480 `
481 INSERT INTO ${LEDGER_TABLE} (tenant, time_completed)
482 VALUES ($1, $2)
483 ON CONFLICT (tenant) DO UPDATE SET time_completed = EXCLUDED.time_completed
484 `,
485 [org.tenant, Date.now()],
486 );
487 });
488};
489
490const writeDryRun = (planInput: MigrationInput): void => {
491 const plan = planMigration({
492 ...planInput,
493 collectPolicyErrors: true,
494 orphanPolicyMode: "retarget_all",
495 });
496 const neverWiden = verifyPolicyRewriteNeverWidens(plan, planInput);
497 const r2Copies = plan.orgs
498 .filter((org) => !org.completed && org.hardErrors.length === 0)
499 .flatMap((org) =>
500 org.blobCopies
501 .filter((copy) => copy.backend === "external")
502 .map((copy) => ({
503 tenantHash: org.tenantHash,
504 tenant: org.tenant,
505 sourceKey: copy.sourceObjectName,
506 destKey: copy.targetObjectName,
507 })),
508 );
509 mkdirSync(outputDir, { recursive: true });
510 for (const org of plan.orgs) {
511 writeFileSync(`${outputDir}/org-${org.tenantHash}.md`, renderOrgDiff(org));
512 }
513 writeFileSync(`${outputDir}/r2-copy-manifest.json`, `${JSON.stringify(r2Copies, null, 2)}\n`);
514 writeFileSync(
515 `${outputDir}/summary.md`,
516 `${renderSummary(plan)}\npolicy_never_widen=${neverWiden.ok}\npolicy_never_widen_checked=${neverWiden.checkedPolicies}\nr2_copy_manifest=${outputDir}/r2-copy-manifest.json\nr2_copy_count=${r2Copies.length}\nr2_copy_command=wrangler r2 object get <bucket>/<sourceKey> | wrangler r2 object put <bucket>/<destKey> --pipe\n`,
517 );
518 console.log(renderSummary(plan));
519 console.log(`policy_never_widen=${neverWiden.ok}`);
520 console.log(`policy_never_widen_checked=${neverWiden.checkedPolicies}`);
521 console.log(`r2_copy_manifest=${outputDir}/r2-copy-manifest.json`);
522 console.log(`r2_copy_count=${r2Copies.length}`);
523 console.log(
524 "r2_copy_command=wrangler r2 object get <bucket>/<sourceKey> | wrangler r2 object put <bucket>/<destKey> --pipe",
525 );
526 console.log(`diff_dir=${outputDir}`);
527 if (!neverWiden.ok) {
528 throw new Error(
529 `Policy rewrite failed coverage checks: widened=${neverWiden.widened.length}, narrowed=${neverWiden.narrowed.length}`,
530 );
531 }
532 if (
533 plan.summary.integrationsMissingSpecBlob > 0 ||
534 plan.summary.integrationsMissingDefsBlob > 0
535 ) {
536 throw new Error(
537 `Serving blob check failed: missing spec=${plan.summary.integrationsMissingSpecBlob}, missing defs=${plan.summary.integrationsMissingDefsBlob}`,
538 );
539 }
540 const zeroOperationIntegrations = plan.orgs
541 .filter((org) => !org.completed && org.hardErrors.length === 0)
542 .flatMap((org) => org.integrations)
543 .filter(
544 (integration) =>
545 integration.servingState.operationsToBuild === 0 &&
546 !integration.servingState.expectedZeroOperations,
547 );
548 if (zeroOperationIntegrations.length > 0) {
549 throw new Error(
550 `Serving operation check failed for ${zeroOperationIntegrations.length} planned integration(s)`,
551 );
552 }
553};
554
555const main = async (): Promise<void> => {
556 if (inputJson) {
557 const input = parseInputJson(inputJson);
558 writeDryRun(
559 filterInputByOrg(
560 {
561 ...input,
562 trafficLastTenant: trafficLastTenant ?? input.trafficLastTenant,
563 },
564 orgFilter,
565 ),
566 );
567 return;
568 }
569 if (!databaseUrl) {
570 throw new Error("Set DATABASE_URL or pass --input-json");
571 }
572 const usesLocalDatabase = databaseUrl.includes("127.0.0.1") || databaseUrl.includes("localhost");
573 const sql = postgres(databaseUrl, {
574 max: 1,
575 prepare: false,
576 ...(usesLocalDatabase ? {} : { ssl: "require" as const }),
577 }) as unknown as SqlClient;
578 try {
579 const input = filterInputByOrg(
580 await readDatabaseInput(sql, !apply, usesLocalDatabase ? "database" : "external"),
581 orgFilter,
582 );
583 if (!apply) {
584 writeDryRun(input);
585 return;
586 }
587 // Orphan block/require_approval policies (a guardrail targeting a service
588 // the org never derived, e.g. a gmail send-block on a Search-Console-only
589 // org) retarget to every derived service as dormant rows rather than
590 // blocking apply. Approved by Rhys 2026-07-08; matches the boot rail.
591 const plan = planMigration({ ...input, orphanPolicyMode: "retarget_all" });
592 const neverWiden = verifyPolicyRewriteNeverWidens(plan, input);
593 if (!neverWiden.ok) {
594 throw new Error(
595 `Policy rewrite failed coverage checks: widened=${neverWiden.widened.length}, narrowed=${neverWiden.narrowed.length}`,
596 );
597 }
598 if (
599 plan.summary.integrationsMissingSpecBlob > 0 ||
600 plan.summary.integrationsMissingDefsBlob > 0
601 ) {
602 throw new Error(
603 `Serving blob check failed: missing spec=${plan.summary.integrationsMissingSpecBlob}, missing defs=${plan.summary.integrationsMissingDefsBlob}`,
604 );
605 }
606 const zeroOperationIntegrations = plan.orgs
607 .filter((org) => !org.completed && org.hardErrors.length === 0)
608 .flatMap((org) => org.integrations)
609 .filter(
610 (integration) =>
611 integration.servingState.operationsToBuild === 0 &&
612 !integration.servingState.expectedZeroOperations,
613 );
614 if (zeroOperationIntegrations.length > 0) {
615 throw new Error(
616 `Serving operation check failed for ${zeroOperationIntegrations.length} planned integration(s)`,
617 );
618 }
619 const requiredR2Copies = plan.orgs
620 .filter((org) => !org.completed && org.hardErrors.length === 0)
621 .flatMap((org) => org.blobCopies.filter((copy) => copy.backend === "external"));
622 if (requiredR2Copies.length > 0 && !r2CopiesDone) {
623 throw new Error(
624 `External R2 blob copies are required before apply (${requiredR2Copies.length}); run dry-run, execute r2-copy-manifest.json, then pass --r2-copies-done`,
625 );
626 }
627 for (const org of plan.orgs) {
628 await applyOrg(sql, org);
629 console.log(`applied_org=${tenantHash(org.tenant)}`);
630 }
631 console.log(renderSummary(plan));
632 } finally {
633 await sql.end({ timeout: 0 });
634 }
635};
636
637if (import.meta.url === pathToFileURL(process.argv[1] ?? "").href) {
638 await main();
639}