Add migration cutover reconciliation journal

This commit is contained in:
ZacharyZhang-NY committed 2026-05-23 21:30:44 -04:00
1 parent a21d35a165
commit 62ac397459
12 files changed
+319 -6

No files matched your search

@@ -0,0 +1,62 @@
export const migrationCutoverJournalSchema = `
create table if not exists migration_cutover_events (
id text primary key,
org_id text not null references organizations(id) on delete cascade,
migration_report_id text not null references migration_reports(id) on delete cascade,
rollback_snapshot_id text references rollback_snapshots(id) on delete set null,
event_type text not null,
actor_type text not null,
actor_id text,
payload_json text not null,
created_at text not null
);
create table if not exists migration_reconciliation_items (
id text primary key,
org_id text not null references organizations(id) on delete cascade,
migration_report_id text not null references migration_reports(id) on delete cascade,
rollback_snapshot_id text references rollback_snapshots(id) on delete set null,
source_event_id text not null,
target_type text not null,
target_id text not null,
reason text not null,
state text not null,
payload_json text not null,
created_at text not null,
resolved_at text,
resolved_by_employee_id text references employees(id) on delete set null,
unique(rollback_snapshot_id, source_event_id)
);
insert into migration_cutover_events(
id, org_id, migration_report_id, rollback_snapshot_id, event_type,
actor_type, actor_id, payload_json, created_at
)
select
'mce_' || lower(hex(randomblob(16))),
org_id,
migration_report_id,
id,
case active when 1 then 'rollback_snapshot_activated' else 'rollback_snapshot_created' end,
'system',
null,
json_object('sourceFingerprint', source_fingerprint, 'sourceDatabaseSha256', source_database_sha256),
created_at
from rollback_snapshots
where not exists (
select 1 from migration_cutover_events
where migration_cutover_events.rollback_snapshot_id = rollback_snapshots.id
);
create index if not exists idx_migration_cutover_events_org_created
on migration_cutover_events(org_id, created_at);
create index if not exists idx_migration_cutover_events_report
on migration_cutover_events(migration_report_id, created_at);
create index if not exists idx_migration_reconciliation_items_org_state
on migration_reconciliation_items(org_id, state, created_at);
create index if not exists idx_migration_reconciliation_items_report
on migration_reconciliation_items(migration_report_id, created_at);
insert into schema_migrations(version, name, applied_at)
values (28, 'migration_cutover_journal', datetime('now'));
`;
@@ -26,6 +26,7 @@ import { taskRunProviderRequestSchema } from "./migrations/024-task-run-provider
import { chatMessageDeliverySchema } from "./migrations/025-chat-message-delivery";
import { browserPlatformGovernanceSchema } from "./migrations/026-browser-platform-governance";
import { socialAuditRecordsSchema } from "./migrations/027-social-audit-records";
import { migrationCutoverJournalSchema } from "./migrations/028-migration-cutover-journal";
interface Migration {
version: number;
@@ -61,6 +62,7 @@ const MIGRATIONS: Migration[] = [
{ version: 25, name: "chat_message_delivery", sql: chatMessageDeliverySchema },
{ version: 26, name: "browser_platform_governance", sql: browserPlatformGovernanceSchema },
{ version: 27, name: "social_audit_records", sql: socialAuditRecordsSchema },
{ version: 28, name: "migration_cutover_journal", sql: migrationCutoverJournalSchema },
];
const SCHEMA_VERSION = MIGRATIONS[MIGRATIONS.length - 1].version;
@@ -0,0 +1,103 @@
import type { DatabaseSync } from "node:sqlite";
import { createId } from "../../shared/ids";
const REPLAYABLE_ENTITY_TYPES = new Set([
"agent",
"approval",
"browser_action",
"chat_message",
"chat_space",
"file",
"scheduled_post",
"social_account",
"task",
"workspace",
]);
export interface CutoverDeltaEvent {
id: string;
entityType: string;
entityId: string;
actorType: string;
actorId: string | null;
action: string;
payload: Record<string, unknown>;
createdAt: string;
}
export interface RecordCutoverEventInput {
orgId: string;
migrationReportId: string;
rollbackSnapshotId: string | null;
eventType: string;
actorType: string;
actorId: string | null;
payload: Record<string, unknown>;
createdAt?: string;
}
export class MigrationCutoverJournalService {
constructor(private readonly db: DatabaseSync) {}
recordEvent(input: RecordCutoverEventInput): string {
const id = createId("mce");
this.db.prepare(`
insert into migration_cutover_events(
id, org_id, migration_report_id, rollback_snapshot_id, event_type,
actor_type, actor_id, payload_json, created_at
) values (?, ?, ?, ?, ?, ?, ?, ?, ?)
`).run(
id,
input.orgId,
input.migrationReportId,
input.rollbackSnapshotId,
input.eventType,
input.actorType,
input.actorId,
JSON.stringify(input.payload),
input.createdAt ?? new Date().toISOString(),
);
return id;
}
queueReconciliation(input: {
orgId: string;
migrationReportId: string;
rollbackSnapshotId: string;
events: CutoverDeltaEvent[];
createdAt?: string;
}): number {
let queued = 0;
const createdAt = input.createdAt ?? new Date().toISOString();
const statement = this.db.prepare(`
insert or ignore into migration_reconciliation_items(
id, org_id, migration_report_id, rollback_snapshot_id, source_event_id,
target_type, target_id, reason, state, payload_json, created_at, resolved_at, resolved_by_employee_id
) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, null, null)
`);
for (const event of input.events) {
if (!REPLAYABLE_ENTITY_TYPES.has(event.entityType)) continue;
const result = statement.run(
createId("mri"),
input.orgId,
input.migrationReportId,
input.rollbackSnapshotId,
event.id,
event.entityType,
event.entityId,
"manual_delta_replay",
"open",
JSON.stringify({
action: event.action,
actorType: event.actorType,
actorId: event.actorId,
payload: event.payload,
createdAt: event.createdAt,
}),
createdAt,
);
queued += Number(result.changes);
}
return queued;
}
}
@@ -77,6 +77,8 @@ const LEGACY_TABLES = [
"search_documents",
"audit_replay_runs",
"rollback_snapshots",
"migration_cutover_events",
"migration_reconciliation_items",
"audit_events",
];
@@ -248,6 +250,13 @@ const RELATION_CHECKS: RelationCheck[] = [
{ table: "rollback_snapshots", column: "org_id", referencedTable: "organizations", referencedColumn: "id", reason: "dangling_org_id" },
{ table: "rollback_snapshots", column: "migration_report_id", referencedTable: "migration_reports", referencedColumn: "id", reason: "dangling_migration_report_id" },
{ table: "rollback_snapshots", column: "created_by_employee_id", referencedTable: "employees", referencedColumn: "id", reason: "dangling_creator_employee_id" },
{ table: "migration_cutover_events", column: "org_id", referencedTable: "organizations", referencedColumn: "id", reason: "dangling_org_id" },
{ table: "migration_cutover_events", column: "migration_report_id", referencedTable: "migration_reports", referencedColumn: "id", reason: "dangling_migration_report_id" },
{ table: "migration_cutover_events", column: "rollback_snapshot_id", referencedTable: "rollback_snapshots", referencedColumn: "id", reason: "dangling_rollback_snapshot_id" },
{ table: "migration_reconciliation_items", column: "org_id", referencedTable: "organizations", referencedColumn: "id", reason: "dangling_org_id" },
{ table: "migration_reconciliation_items", column: "migration_report_id", referencedTable: "migration_reports", referencedColumn: "id", reason: "dangling_migration_report_id" },
{ table: "migration_reconciliation_items", column: "rollback_snapshot_id", referencedTable: "rollback_snapshots", referencedColumn: "id", reason: "dangling_rollback_snapshot_id" },
{ table: "migration_reconciliation_items", column: "resolved_by_employee_id", referencedTable: "employees", referencedColumn: "id", reason: "dangling_resolver_employee_id" },
{ table: "audit_events", column: "org_id", referencedTable: "organizations", referencedColumn: "id", reason: "dangling_org_id" },
];
@@ -4,6 +4,7 @@ import type { DatabaseSync } from "node:sqlite";
import type { CreateRollbackSnapshotInput, SetRollbackSnapshotActiveInput } from "../../shared/commands";
import { createId } from "../../shared/ids";
import { AuditService } from "./audit-service";
import { MigrationCutoverJournalService, type CutoverDeltaEvent } from "./migration-cutover-journal-service";
interface MigrationReportRow {
id: string;
@@ -31,6 +32,13 @@ interface WorkspaceRow {
updated_at: string;
}
interface RollbackSnapshotRow {
id: string;
migration_report_id: string;
source_fingerprint: string;
delta_journal_json: string;
}
interface FileRow {
id: string;
name: string;
@@ -54,9 +62,11 @@ interface AuditRow {
export class RollbackService {
private readonly audit: AuditService;
private readonly cutoverJournal: MigrationCutoverJournalService;
constructor(private readonly db: DatabaseSync) {
this.audit = new AuditService(db);
this.cutoverJournal = new MigrationCutoverJournalService(db);
}
createSnapshot(input: CreateRollbackSnapshotInput): string {
@@ -109,13 +119,26 @@ export class RollbackService {
action: "created",
payload: { migrationReportId: report.id, sourceFingerprint: body.sourceFingerprint },
});
this.cutoverJournal.recordEvent({
orgId: input.orgId,
migrationReportId: report.id,
rollbackSnapshotId: id,
eventType: "rollback_snapshot_created",
actorType: "employee",
actorId: input.createdByEmployeeId,
payload: { sourceFingerprint: body.sourceFingerprint, sourceDatabaseSha256, workspaceManifestSha256 },
createdAt: now,
});
return id;
}
setActive(input: SetRollbackSnapshotActiveInput): void {
this.requireEmployee(input.orgId, input.employeeId);
const snapshot = this.db.prepare("select id from rollback_snapshots where id = ? and org_id = ?")
.get(input.snapshotId, input.orgId) as { id: string } | undefined;
const snapshot = this.db.prepare(`
select id, migration_report_id, source_fingerprint, delta_journal_json
from rollback_snapshots
where id = ? and org_id = ?
`).get(input.snapshotId, input.orgId) as RollbackSnapshotRow | undefined;
if (!snapshot) throw new Error("Rollback snapshot was not found.");
const now = new Date().toISOString();
if (input.active) this.db.prepare("update rollback_snapshots set active = 0, activated_at = null where org_id = ?").run(input.orgId);
@@ -131,6 +154,25 @@ export class RollbackService {
action: input.active ? "activated" : "deactivated",
payload: { active: input.active },
});
const queuedReconciliationItems = input.active
? this.cutoverJournal.queueReconciliation({
orgId: input.orgId,
migrationReportId: snapshot.migration_report_id,
rollbackSnapshotId: snapshot.id,
events: parseDeltaJournal(snapshot.delta_journal_json),
createdAt: now,
})
: 0;
this.cutoverJournal.recordEvent({
orgId: input.orgId,
migrationReportId: snapshot.migration_report_id,
rollbackSnapshotId: snapshot.id,
eventType: input.active ? "rollback_snapshot_activated" : "rollback_snapshot_deactivated",
actorType: "employee",
actorId: input.employeeId,
payload: { active: input.active, sourceFingerprint: snapshot.source_fingerprint, queuedReconciliationItems },
createdAt: now,
});
}
private requireReport(orgId: string, reportId: string): MigrationReportRow {
@@ -248,3 +290,34 @@ function parseRecord(json: string): Record<string, unknown> {
if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) throw new Error("JSON record is invalid.");
return parsed as Record<string, unknown>;
}
function parseDeltaJournal(json: string): CutoverDeltaEvent[] {
const parsed = JSON.parse(json) as unknown;
if (!Array.isArray(parsed)) throw new Error("Rollback delta journal is invalid.");
return parsed.map((item) => {
if (typeof item !== "object" || item === null || Array.isArray(item)) throw new Error("Rollback delta journal entry is invalid.");
const record = item as Record<string, unknown>;
if (
typeof record.id !== "string"
|| typeof record.entityType !== "string"
|| typeof record.entityId !== "string"
|| typeof record.actorType !== "string"
|| typeof record.action !== "string"
|| typeof record.createdAt !== "string"
) {
throw new Error("Rollback delta journal entry is invalid.");
}
const payload = record.payload;
if (typeof payload !== "object" || payload === null || Array.isArray(payload)) throw new Error("Rollback delta journal entry is invalid.");
return {
id: record.id,
entityType: record.entityType,
entityId: record.entityId,
actorType: record.actorType,
actorId: typeof record.actorId === "string" ? record.actorId : null,
action: record.action,
payload: payload as Record<string, unknown>,
createdAt: record.createdAt,
};
});
}
@@ -262,7 +262,7 @@ export class SdlcService {
{ key: "provider_adapter_coverage", passed: missing.length === 0, value: `${adapterKeys.length}/${catalogKeys.length}` },
{ key: "provider_rate_limit_registry", passed: rateLimitRegistry === 1, value: rateLimitRegistry },
{ key: "provider_runtime_contract", passed: runtimeContractReady, value: runtimeContractReady },
{ key: "schema_version", passed: schemaVersion >= 27, value: schemaVersion },
{ key: "schema_version", passed: schemaVersion >= 28, value: schemaVersion },
];
}
@@ -126,6 +126,8 @@ export class SnapshotService {
policyRules: list(this.db, "policy_rules", "created_at asc"),
migrationReports: list(this.db, "migration_reports", "created_at desc"),
rollbackSnapshots,
migrationCutoverEvents: list(this.db, "migration_cutover_events", "created_at desc"),
migrationReconciliationItems: list(this.db, "migration_reconciliation_items", "created_at desc"),
complianceExports: list(this.db, "compliance_exports", "created_at desc"),
legalHolds,
retentionFreezes,