root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(core): converge multi-project SQLite cutover

Browse Source 
Migrate central SQLite state once per cluster, isolate project metadata, and preserve file-local revision identities while verifying accumulated shared tables.
This commit is contained in:
gsxdsm
2026-07-14 10:50:59 -07:00
parent 1b9dce22c0
commit 7c8a84fb2f
11 changed files with 288 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/quiet-migrations-converge.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Make SQLite cutover converge when multiple registered projects share embedded PostgreSQL.
category: fix
dev: Central data migrates once; project metadata and local revision identities are isolated during retries.

19
packages/core/src/__tests__/postgres/schema-applier.test.ts
View File

@@ -35,6 +35,7 @@ import {
import {
LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION,
MONITOR_APPROVAL_ISOLATION_SCHEMA_VERSION,
MULTI_PROJECT_CUTOVER_SCHEMA_VERSION,
} from "../../postgres/schema-applier.js";
const PG_ADMIN_URL =
@@ -55,7 +56,13 @@ describe("schema-applier: immutable migration identities", () => {
it("keeps legacy cutover preservation assigned to version 0004", () => {
expect(LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION).toBe("0004");
expect(SCHEMA_BASELINE_VERSION).toBe(LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION);
expect(Number(SCHEMA_BASELINE_VERSION))
.toBeGreaterThanOrEqual(Number(LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION));
});
it("keeps multi-project cutover assigned to version 0005", () => {
expect(MULTI_PROJECT_CUTOVER_SCHEMA_VERSION).toBe("0005");
expect(SCHEMA_BASELINE_VERSION).toBe(MULTI_PROJECT_CUTOVER_SCHEMA_VERSION);
});
});
@@ -548,7 +555,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
const versions = (await ctx.db.execute(sql`
SELECT version FROM public.fusion_schema_migrations ORDER BY version
`)) as unknown as Array<{ version: string }>;
expect(versions.map(({ version }) => version)).toEqual(["0000", "0001", "0002", "0003", SCHEMA_BASELINE_VERSION]);
expect(versions.map(({ version }) => version)).toEqual(["0000", "0001", "0002", "0003", "0004", SCHEMA_BASELINE_VERSION]);
expect((await applySchemaBaseline(ctx.db, { pluginHooks: [] })).applied).toBe(false);
});
@@ -572,7 +579,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
applySchemaBaseline(ctx.db, { pluginHooks: [] }),
]);
expect(results.filter(({ applied }) => applied)).toHaveLength(1);
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", SCHEMA_BASELINE_VERSION]);
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", SCHEMA_BASELINE_VERSION]);
});
it("upgrades a 0001 database by backfilling analytics ownership", async () => {
@@ -602,7 +609,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
))) as unknown as Array<{ project_id: string }>;
expect(rows).toEqual([{ project_id: "project-a" }]);
}
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004"]);
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005"]);
});
/**
@@ -634,7 +641,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
))) as unknown as Array<{ project_id: string }>;
expect(rows).toEqual([{ project_id: "project-a" }]);
}
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004"]);
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005"]);
});
/*
@@ -672,7 +679,7 @@ pgDescribe("schema-applier: automation project-isolation upgrade", () => {
"project_auth_users",
"task_reviewer_runs",
]);
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004"]);
expect(await getAppliedMigrations(ctx.db)).toEqual(["0000", "0001", "0002", "0003", "0004", "0005"]);
});
});

62
packages/core/src/__tests__/postgres/sqlite-migrator.test.ts
View File

@@ -664,6 +664,68 @@ pgDescribe("SQLite-to-PostgreSQL migrator", () => {
]);
});
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
Sequential project cutovers share one PostgreSQL schema. Verification must accept accumulated shared-table rows, keep __meta isolated by project, and generate a new task-revision identity when both SQLite files start their local sequence at 1.
*/
it("converges sequential project migrations in one shared PostgreSQL database", async () => {
const makeProjectDb = (name: string, projectId: string, taskId: string): string => {
const sqlitePath = join(ctx!.fusionDir, name);
const legacy = new DatabaseSync(sqlitePath);
try {
legacy.exec(`
CREATE TABLE agents (
id TEXT PRIMARY KEY, name TEXT NOT NULL, role TEXT NOT NULL,
state TEXT NOT NULL, createdAt TEXT NOT NULL, updatedAt TEXT NOT NULL
);
CREATE TABLE task_document_revisions (
id INTEGER PRIMARY KEY AUTOINCREMENT, taskId TEXT NOT NULL, key TEXT NOT NULL,
content TEXT NOT NULL, revision INTEGER NOT NULL, author TEXT NOT NULL,
metadata TEXT, createdAt TEXT NOT NULL
);
CREATE TABLE __meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
`);
legacy.prepare("INSERT INTO agents VALUES (?, ?, ?, ?, ?, ?)").run(
`agent-${projectId}`, `Agent ${projectId}`, "worker", "idle", "2026-07-14", "2026-07-14",
);
legacy.prepare("INSERT INTO task_document_revisions VALUES (1, ?, 'docs', ?, 1, 'agent', '{}', '2026-07-14')").run(
taskId, `content-${projectId}`,
);
legacy.prepare("INSERT INTO __meta VALUES ('projectId', ?)").run(projectId);
} finally {
legacy.close();
}
return sqlitePath;
};
const firstPath = makeProjectDb("project-a.db", "project-a", "A-1");
const secondPath = makeProjectDb("project-b.db", "project-b", "B-1");
const first = await migrateTest(
ctx!.db, [{ sqlitePath: firstPath, pgSchema: "project" as const }], { projectId: "project-a" },
);
const second = await migrateTest(
ctx!.db, [{ sqlitePath: secondPath, pgSchema: "project" as const }], { projectId: "project-b" },
);
expect(first.tables.every((table) => table.verified)).toBe(true);
expect(second.tables.every((table) => table.verified)).toBe(true);
expect(second.tables.find((table) => table.table === "agents")).toEqual(
expect.objectContaining({ sourceRows: 1, targetRows: 2, verified: true }),
);
const revisions = await ctx!.db.execute(sql`
SELECT id, task_id FROM project.task_document_revisions ORDER BY task_id
`) as unknown as Array<{ id: number; task_id: string }>;
expect(revisions.map(({ task_id }) => task_id)).toEqual(["A-1", "B-1"]);
expect(new Set(revisions.map(({ id }) => id)).size).toBe(2);
const metadata = await ctx!.db.execute(sql`
SELECT project_id, key, value FROM project.__meta ORDER BY project_id
`) as unknown as Array<{ project_id: string; key: string; value: string }>;
expect(metadata).toEqual([
{ project_id: "project-a", key: "projectId", value: "project-a" },
{ project_id: "project-b", key: "projectId", value: "project-b" },
]);
});
/*
FNXC:PostgresMigrationRetry 2026-07-14-09:06:
Retrying after the former non-transactional migrator must repair a globally keyed row copied under a NULL project partition. Reconciliation may replace only an exact migrated-content match, so unrelated PostgreSQL state remains untouched.

23
packages/core/src/__tests__/postgres/startup-factory-integration.test.ts
View File

@@ -516,12 +516,35 @@ pgDescribe("startup-factory: external PostgreSQL boot (integration)", () => {
expect(first).not.toBeNull();
await first!.shutdown();
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
Central SQLite is a one-time cluster source. A legitimate PostgreSQL-side update after project A's cutover must survive project B startup; re-verifying central content for B caused the reported plugin_installs checksum failure.
*/
const betweenProjects = postgres(testUrl, { max: 1 });
try {
await betweenProjects`UPDATE central.projects SET name = 'Updated in PostgreSQL' WHERE id = 'project-a'`;
} finally {
await betweenProjects.end();
}
const second = await createTaskStoreForBackend({ rootDir: projectB, globalSettingsDir: globalDir, env: { DATABASE_URL: testUrl } });
expect(second).not.toBeNull();
try {
expect(second!.taskStore.getAsyncLayer()!.projectId).toBe("project-b");
expect((await second!.taskStore.getTask("B-1")).title).toBe("Project B task");
await expect(second!.taskStore.getTask("A-1")).rejects.toThrow();
const client = postgres(testUrl, { max: 1 });
try {
const projects = await client<{ name: string }[]>`SELECT name FROM central.projects WHERE id = 'project-a'`;
expect(projects).toEqual([{ name: "Updated in PostgreSQL" }]);
const centralMarkers = await client<{ status: string }[]>`
SELECT status FROM public.fusion_sqlite_migrations
WHERE migration_key = 'central:legacy-sqlite'
`;
expect(centralMarkers).toEqual([{ status: "complete" }]);
} finally {
await client.end();
}
} finally {
await second!.shutdown();
}

15
packages/core/src/async-agent-store.ts
View File

@@ -896,11 +896,15 @@ export async function getAllBlockedStates(
export async function getMetaValue(
handle: QueryHandle,
key: string,
projectId = "",
): Promise<string | undefined> {
const rows = await handle
.select({ value: schema.project.projectMeta.value })
.from(schema.project.projectMeta)
.where(eq(schema.project.projectMeta.key, key));
.where(and(
eq(schema.project.projectMeta.projectId, projectId),
eq(schema.project.projectMeta.key, key),
));
return rows[0]?.value ?? undefined;
}
@@ -911,12 +915,17 @@ export async function upsertMetaValue(
handle: QueryHandle,
key: string,
value: string,
projectId = "",
): Promise<void> {
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
Agent-store migration markers share the project schema but not project ownership. Include the bound project in their composite key; the empty binding remains the explicit project-agnostic compatibility partition.
*/
await handle
.insert(schema.project.projectMeta)
.values({ key, value })
.values({ projectId, key, value })
.onConflictDoUpdate({
target: schema.project.projectMeta.key,
target: [schema.project.projectMeta.projectId, schema.project.projectMeta.key],
set: { value },
});
}

44
packages/core/src/postgres/migrations/0005_multi_project_cutover.sql Normal file
View File

@@ -0,0 +1,44 @@
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
Project metadata keys and SQLite revision identities are file-local. Upgrade shared PostgreSQL targets so each registered project retains its own __meta rows and task-document revisions preserve their project plus original-SQLite identity instead of colliding on an integer copied from another file.
*/
DO $$
DECLARE
completed_project_id text := '';
BEGIN
IF to_regclass('project.__meta') IS NOT NULL THEN
ALTER TABLE project.__meta
ADD COLUMN IF NOT EXISTS project_id text NOT NULL DEFAULT '';
IF to_regclass('public.fusion_sqlite_migrations') IS NOT NULL THEN
SELECT project_id INTO completed_project_id
FROM public.fusion_sqlite_migrations
WHERE status = 'complete' AND project_id IS NOT NULL
ORDER BY updated_at ASC
LIMIT 1;
END IF;
UPDATE project.__meta
SET project_id = COALESCE(completed_project_id, '')
WHERE project_id = '';
ALTER TABLE project.__meta DROP CONSTRAINT IF EXISTS __meta_pkey;
ALTER TABLE project.__meta
ADD CONSTRAINT __meta_pkey PRIMARY KEY (project_id, key);
END IF;
IF to_regclass('project.task_document_revisions') IS NOT NULL THEN
DROP INDEX IF EXISTS project.task_document_revisions_natural_key_unique;
ALTER TABLE project.task_document_revisions
ADD COLUMN IF NOT EXISTS project_id text,
ADD COLUMN IF NOT EXISTS legacy_sqlite_id integer;
UPDATE project.task_document_revisions
SET project_id = COALESCE(completed_project_id, ''),
legacy_sqlite_id = id
WHERE legacy_sqlite_id IS NULL;
CREATE UNIQUE INDEX IF NOT EXISTS task_document_revisions_legacy_identity_unique
ON project.task_document_revisions(project_id, legacy_sqlite_id);
END IF;
END $$;

22
packages/core/src/postgres/schema-applier.ts
View File

@@ -27,7 +27,7 @@ import { sql } from "drizzle-orm";
import { runPluginSchemaInitHooks, DEFAULT_PLUGIN_SCHEMA_INIT_HOOKS, type PluginSchemaInitHook } from "./plugin-schema-hook.js";
/** The latest PostgreSQL schema version known to this applier. */
export const SCHEMA_BASELINE_VERSION = "0004";
export const SCHEMA_BASELINE_VERSION = "0005";
const INITIAL_SCHEMA_VERSION = "0000";
const AUTOMATION_ISOLATION_SCHEMA_VERSION = "0001";
const ANALYTICS_ISOLATION_SCHEMA_VERSION = "0002";
@@ -37,6 +37,7 @@ const ANALYTICS_ISOLATION_SCHEMA_VERSION = "0002";
*/
export const MONITOR_APPROVAL_ISOLATION_SCHEMA_VERSION = "0003";
export const LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION = "0004";
export const MULTI_PROJECT_CUTOVER_SCHEMA_VERSION = "0005";
/** Bookkeeping table for the fresh Drizzle migration history. */
export const MIGRATION_BOOKKEEPING_TABLE = "fusion_schema_migrations";
@@ -63,6 +64,11 @@ const LEGACY_CUTOVER_PRESERVATION_MIGRATION_PATH = join(
"migrations",
"0004_legacy_cutover_preservation.sql",
);
const MULTI_PROJECT_CUTOVER_MIGRATION_PATH = join(
__dirname,
"migrations",
"0005_multi_project_cutover.sql",
);
/**
* Ensure the migration bookkeeping table exists. Lives in the public schema so
@@ -124,6 +130,7 @@ export async function applySchemaBaseline(
const analyticsIsolationAlreadyApplied = applied.includes(ANALYTICS_ISOLATION_SCHEMA_VERSION);
const monitorApprovalIsolationAlreadyApplied = applied.includes(MONITOR_APPROVAL_ISOLATION_SCHEMA_VERSION);
const legacyCutoverPreservationAlreadyApplied = applied.includes(LEGACY_CUTOVER_PRESERVATION_SCHEMA_VERSION);
const multiProjectCutoverAlreadyApplied = applied.includes(MULTI_PROJECT_CUTOVER_SCHEMA_VERSION);
let schemaChanged = false;
if (!baselineAlreadyApplied) {
@@ -190,6 +197,19 @@ export async function applySchemaBaseline(
schemaChanged = true;
}
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
Existing targets may already contain one completed project plus a partially copied second project. Apply metadata partitioning and collision-safe revision identity before any retry builds its migration plan.
*/
if (!multiProjectCutoverAlreadyApplied) {
const migrationSql = await readFile(MULTI_PROJECT_CUTOVER_MIGRATION_PATH, "utf8");
await tx.execute(sql.raw(migrationSql));
await tx.execute(
sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${MULTI_PROJECT_CUTOVER_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`,
);
schemaChanged = true;
}
// Run plugin schema-init hooks regardless of whether the baseline was just
// applied or already present — plugin tables must exist on every connection
// the applier touches. The hooks are themselves idempotent (CREATE TABLE IF

16
packages/core/src/postgres/schema/project.ts
View File

@@ -886,6 +886,8 @@ export const artifacts = projectSchema.table("artifacts", {
export const taskDocumentRevisions = projectSchema.table("task_document_revisions", {
id: integer("id").generatedAlwaysAsIdentity().primaryKey(),
projectId: text("project_id"),
legacySqliteId: integer("legacy_sqlite_id"),
taskId: text("task_id").notNull(),
key: text("key").notNull(),
content: text("content").notNull(),
@@ -893,7 +895,10 @@ export const taskDocumentRevisions = projectSchema.table("task_document_revision
author: text("author").notNull(),
metadata: jsonb("metadata"),
createdAt: text("created_at").notNull(),
}, (t) => [index("idxTaskDocumentRevisionsTaskKey").on(t.taskId, t.key)]);
}, (t) => [
index("idxTaskDocumentRevisionsTaskKey").on(t.taskId, t.key),
unique("task_document_revisions_legacy_identity_unique").on(t.projectId, t.legacySqliteId),
]);
// ── Research runs ────────────────────────────────────────────────────
export const researchRuns = projectSchema.table("research_runs", {
@@ -1081,9 +1086,14 @@ export const secrets = projectSchema.table("secrets", {
// ── Schema version meta ──────────────────────────────────────────────
export const projectMeta = projectSchema.table("__meta", {
key: text("key").primaryKey(),
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
Embedded PostgreSQL is shared by every registered project, so SQLite __meta keys must be partitioned by the authoritative registry project ID. A global key primary key makes the second project inherit or overwrite the first project's identity and migration markers.
*/
projectId: text("project_id").notNull(),
key: text("key").notNull(),
value: text("value"),
});
}, (t) => [primaryKey({ columns: [t.projectId, t.key] })]);
// ── Missions hierarchy ───────────────────────────────────────────────
export const missions = projectSchema.table("missions", {

95
packages/core/src/postgres/sqlite-migrator.ts
View File

@@ -182,6 +182,7 @@ export interface MigrationOptions {
}
const SQLITE_MIGRATION_STATE_TABLE = "fusion_sqlite_migrations";
export const CENTRAL_SQLITE_MIGRATION_KEY = "central:legacy-sqlite";
async function ensureMigrationStateTable(db: PostgresJsDatabase<Record<string, never>>): Promise<void> {
await db.execute(sql.raw(`CREATE TABLE IF NOT EXISTS public.${SQLITE_MIGRATION_STATE_TABLE} (
@@ -218,6 +219,22 @@ export async function completeSqliteMigration(
`);
}
/** Record a verified source independently of a project cutover marker. */
export async function recordSqliteMigrationComplete(
db: PostgresJsDatabase<Record<string, never>>,
migrationKey: string,
projectId?: string,
): Promise<void> {
await ensureMigrationStateTable(db);
await db.execute(sql`
INSERT INTO public.${sql.identifier(SQLITE_MIGRATION_STATE_TABLE)}
(migration_key, project_id, status, last_error, updated_at)
VALUES (${migrationKey}, ${projectId ?? null}, 'complete', NULL, now())
ON CONFLICT (migration_key) DO UPDATE
SET project_id = EXCLUDED.project_id, status = 'complete', last_error = NULL, updated_at = now()
`);
}
/**
* FNXC:PostgresMigration 2026-06-24-08:15:
* Migrate one or more SQLite databases into PostgreSQL schemas.
@@ -450,13 +467,22 @@ async function buildMigrationPlan(
// Legacy SQLite table names are camelCase; PostgreSQL tables are
// snake_case. toSnakeCase is the identity for already-snake names.
const pgTable = toSnakeCase(table);
const { columns: cols, targetColumnNames } = resolveColumnMapping(
const { columns: resolvedColumns, targetColumnNames } = resolveColumnMapping(
pgTable,
table,
sqlite,
targetColumnsByTable,
);
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:46:
Preserve task-document revision identities as provenance, not as the shared table primary key. Two SQLite files can both use id=1, while one file can also contain distinct rows with the same task/key/revision; map the local id into legacy_sqlite_id and let PostgreSQL generate its runtime id.
*/
const cols = source.pgSchema === PROJECT_SCHEMA && pgTable === "task_document_revisions"
? resolvedColumns.map((column) => column.sqliteName === "id"
? { ...column, pgName: "legacy_sqlite_id", type: "plain" as const }
: column)
: resolvedColumns;
/*
FNXC:PostgresMigration 2026-07-14-08:52:
A legacy per-project SQLite table can already contain a nullable projectId column while its existing rows still hold NULL. The registry identity resolved for this one-project-file cutover is authoritative whenever the PostgreSQL target is partitioned, so insertion and verification must override absent, NULL, or stale source project IDs instead of copying them into the required project_id partition.
*/
@@ -909,27 +935,30 @@ async function migrateTable(
// (md5(string_agg(...)) on PostgreSQL, and a Node-side md5 over the SQLite
// converted stream) so the comparison is a single short string per side.
const targetRows = await countTargetRows(db, plan.pgSchema, plan.pgTable, plan.partitionProjectId);
const rowCountOk = targetRows === sourceRows;
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
A bound project imports into a shared PostgreSQL schema. Tables with a project_id column remain exact per-partition checks; intentionally cluster-shared project tables must instead prove the converted source multiset is contained in the accumulated target. A same-key/different-content conflict still fails because the exact source row is absent.
*/
const verifiesSharedProjectTable =
plan.pgSchema === PROJECT_SCHEMA && plan.partitionProjectId === undefined;
const rowCountOk = verifiesSharedProjectTable
? targetRows >= sourceRows
: targetRows === sourceRows;
let contentOk = true;
if (rowCountOk && sourceRows > 0) {
const sourceChecksum = computeSourceContentChecksum(
sqlite,
plan.table,
insertableCols,
plan.partitionProjectId,
const sourceCanonicalRows = computeSourceCanonicalRows(
sqlite, plan.table, insertableCols, plan.partitionProjectId,
);
const targetChecksum = await computeTargetContentChecksum(
db,
plan.pgSchema,
plan.pgTable,
insertableCols,
plan.partitionProjectId,
const targetCanonicalRows = await computeTargetCanonicalRows(
db, plan.pgSchema, plan.pgTable, insertableCols, plan.partitionProjectId,
);
contentOk = sourceChecksum === targetChecksum;
contentOk = verifiesSharedProjectTable
? isCanonicalMultisetSubset(sourceCanonicalRows, targetCanonicalRows)
: checksumCanonicalRows(sourceCanonicalRows) === checksumCanonicalRows(targetCanonicalRows);
if (!contentOk) {
log.warn(
`Content checksum mismatch for ${plan.pgSchema}.${plan.pgTable}: ` +
`source=${sourceChecksum}, target=${targetChecksum}`,
`source=${checksumCanonicalRows(sourceCanonicalRows)}, target=${checksumCanonicalRows(targetCanonicalRows)}`,
);
}
} else if (!rowCountOk) {
@@ -1260,19 +1289,19 @@ function stableJsonStringify(value: unknown): string {
* is the correct semantic: it verifies the copy faithfully reproduced what the
* conversion produced.
*/
function computeSourceContentChecksum(
function computeSourceCanonicalRows(
sqlite: DatabaseSync,
table: string,
cols: readonly ColumnMapping[],
partitionProjectId?: string,
): string {
if (cols.length === 0) return "";
): string[] {
if (cols.length === 0) return [];
const selectCols = cols.map((c) => quoteIdent(c.sqliteName)).join(", ");
const rows = sqlite
.prepare(`SELECT ${selectCols} FROM ${quoteIdent(table)}`)
.all() as Array<Record<string, unknown>>;
const canonicalRows = rows.map((row) => {
return rows.map((row) => {
let canonical = "";
for (const col of cols) {
const converted = col.pgName === "project_id" && partitionProjectId
@@ -1282,6 +1311,9 @@ function computeSourceContentChecksum(
}
return canonical;
}).sort();
}
function checksumCanonicalRows(canonicalRows: readonly string[]): string {
const hash = createHash("md5");
for (const row of canonicalRows) {
hash.update(row);
@@ -1290,6 +1322,17 @@ function computeSourceContentChecksum(
return hash.digest("hex");
}
function isCanonicalMultisetSubset(sourceRows: readonly string[], targetRows: readonly string[]): boolean {
const targetCounts = new Map<string, number>();
for (const row of targetRows) targetCounts.set(row, (targetCounts.get(row) ?? 0) + 1);
for (const row of sourceRows) {
const available = targetCounts.get(row) ?? 0;
if (available === 0) return false;
targetCounts.set(row, available - 1);
}
return true;
}
/**
* Compute a content checksum over the PostgreSQL target rows for a table.
* Selects the SAME insertable columns the copy used and MD5s the canonical
@@ -1302,14 +1345,14 @@ function computeSourceContentChecksum(
* and doing both sides in Node with the same canonicalizeCell function
* guarantees they agree.
*/
async function computeTargetContentChecksum(
async function computeTargetCanonicalRows(
db: PostgresJsDatabase<Record<string, never>>,
pgSchema: string,
table: string,
cols: readonly ColumnMapping[],
projectId?: string,
): Promise<string> {
if (cols.length === 0) return "";
): Promise<string[]> {
if (cols.length === 0) return [];
const selectCols = cols.map((c) => quoteIdent(c.pgName)).join(", ");
const rows = (await db.execute(
sql`SELECT ${sql.raw(selectCols)} FROM ${sql.raw(quoteIdent(pgSchema))}.${sql.raw(
@@ -1321,17 +1364,11 @@ async function computeTargetContentChecksum(
FNXC:PostgresMigrationCompleteness 2026-07-14-09:27:
Content verification must not depend on database collation. SQLite BINARY and PostgreSQL locale collation can order identical mixed-case paths differently, so both sides canonicalize complete rows and sort those strings in Node before hashing.
*/
const canonicalRows = rows.map((row) => {
return rows.map((row) => {
let canonical = "";
for (const col of cols) {
canonical += `${canonicalizeCell(row[col.pgName])}\u0001`;
}
return canonical;
}).sort();
const hash = createHash("md5");
for (const row of canonicalRows) {
hash.update(row);
hash.update("\u0002");
}
return hash.digest("hex");
}

15
packages/core/src/postgres/startup-factory.ts
View File

@@ -440,13 +440,21 @@ export async function createTaskStoreForBackend(
considered.
*/
const migrationKey = `project:${migrationProjectId ?? rootDir}`;
const { migrateSqliteToPostgres, defaultMigrationSources, isSqliteMigrationComplete, completeSqliteMigration } = await import("./sqlite-migrator.js");
const { migrateSqliteToPostgres, defaultMigrationSources, isSqliteMigrationComplete, completeSqliteMigration, recordSqliteMigrationComplete, CENTRAL_SQLITE_MIGRATION_KEY } = await import("./sqlite-migrator.js");
const migrationComplete = await isSqliteMigrationComplete(connections.migration, migrationKey);
if (!migrationComplete && isValidSqliteDatabaseFile(legacySqlitePath)) {
// The central (global-dir) source is optional: when no global dir is
// resolvable (e.g. tests without an explicit dir), migrate only the
// project-local sources rather than failing the boot.
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
The central SQLite database is cluster-global, not a per-project source. Migrate and verify it once, then exclude it from later registered-project cutovers so mutable global rows are not compared with each project's accumulated PostgreSQL state.
*/
const centralMigrationComplete = await isSqliteMigrationComplete(
connections.migration, CENTRAL_SQLITE_MIGRATION_KEY,
);
const sources = defaultMigrationSources(fusionDir, globalDir ?? join(fusionDir, "__no-global-dir__"))
.filter((source) => !centralMigrationComplete || source.pgSchema !== "central")
.filter((source) => existsSync(source.sqlitePath) && isValidSqliteDatabaseFile(source.sqlitePath));
if (sources.length > 0) {
log.log(`startup-factory: empty PostgreSQL database with legacy SQLite data present — auto-migrating ${sources.length} source(s) (SQLite files are kept as backups)`);
@@ -510,6 +518,11 @@ export async function createTaskStoreForBackend(
});
}
await completeSqliteMigration(connections.migration, migrationKey);
if (sources.some((source) => source.pgSchema === "central")) {
await recordSqliteMigrationComplete(
connections.migration, CENTRAL_SQLITE_MIGRATION_KEY, stampProjectId,
);
}
/*
FNXC:PostgresMigrationBanner 2026-07-12:
Remember the successful auto-migration so the dashboard can show a

18
packages/core/src/project-identity.ts
View File

@@ -1,6 +1,6 @@
import { existsSync, mkdirSync } from "node:fs";
import { basename, join } from "node:path";
import { eq } from "drizzle-orm";
import { and, eq } from "drizzle-orm";
import { DatabaseSync } from "./sqlite-adapter.js";
import { createLogger } from "./logger.js";
import * as schema from "./postgres/schema/index.js";
@@ -111,20 +111,28 @@ export function writeProjectIdentity(fusionDir: string, identity: ProjectIdentit
// ─────────────────────────────────────────────────────────────────────
async function readMetaAsync(layer: AsyncDataLayer, key: string): Promise<string | null> {
const projectId = layer.projectId ?? "";
const rows = await layer.db
.select({ value: schema.project.projectMeta.value })
.from(schema.project.projectMeta)
.where(eq(schema.project.projectMeta.key, key));
.where(and(
eq(schema.project.projectMeta.projectId, projectId),
eq(schema.project.projectMeta.key, key),
));
return rows[0]?.value ?? null;
}
async function upsertMetaAsync(layer: AsyncDataLayer, key: string, value: string): Promise<void> {
// The __meta table has a primary key on `key`; upsert via ON CONFLICT.
const projectId = layer.projectId ?? "";
/*
FNXC:PostgresMultiProjectCutover 2026-07-14-11:18:
Backend identity reads and writes must use the data layer's project binding so one registered project cannot inherit or overwrite another project's PostgreSQL __meta stamp.
*/
await layer.db
.insert(schema.project.projectMeta)
.values({ key, value })
.values({ projectId, key, value })
.onConflictDoUpdate({
target: schema.project.projectMeta.key,
target: [schema.project.projectMeta.projectId, schema.project.projectMeta.key],
set: { value },
});
}