diff --git a/src/server/sessions/sessionArchiveMigration.test.ts b/src/server/sessions/sessionArchiveMigration.test.ts new file mode 100644 index 0000000..6f1b6cf --- /dev/null +++ b/src/server/sessions/sessionArchiveMigration.test.ts @@ -0,0 +1,430 @@ +import { constants } from "node:fs"; +import { + access, + appendFile, + copyFile, + lstat, + mkdtemp, + mkdir, + readFile, + readdir, + rm, + unlink, + writeFile, +} from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, sep } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { + inspectLegacySessionArchiveMigration, + migrateLegacySessionArchive, + type LegacySessionArchiveMigrationOptions, +} from "./sessionArchiveMigration.js"; + +const tempRoots: string[] = []; + +describe("legacy session archive migration preflight", () => { + afterEach(async () => { + await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); + }); + + it("proves the ordinary legacy layout eligible without creating destination state", async () => { + const fixture = await createLegacyArchiveFixture(); + + await expect(inspectLegacySessionArchiveMigration(fixture.options)).resolves.toEqual({ + status: "eligible", + legacyIndexPath: fixture.legacyIndexPath, + destinationIndexPath: fixture.destinationIndexPath, + archiveFileCount: 1, + }); + expect(await exists(fixture.destinationRoot)).toBe(false); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + }); + + it("skips when PI_WEB_DATA_DIR is not explicitly configured", async () => { + const fixture = await createLegacyArchiveFixture(); + + await expect(migrateLegacySessionArchive({ ...fixture.options, env: {} })).resolves.toEqual({ + status: "skipped", + reason: "data-dir-not-configured", + }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips when the configured data root resolves to the legacy root", async () => { + const fixture = await createLegacyArchiveFixture(); + + await expect(migrateLegacySessionArchive({ + ...fixture.options, + env: { PI_WEB_DATA_DIR: fixture.legacyRoot }, + })).resolves.toEqual({ status: "skipped", reason: "data-dir-not-distinct" }); + await expectLegacyStateUntouched(fixture); + }); + + it("skips when no legacy index exists without adopting loose legacy files", async () => { + const fixture = await createLegacyArchiveFixture(); + await unlink(fixture.legacyIndexPath); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-index-missing", + }); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips malformed legacy indexes without creating destination state", async () => { + const fixture = await createLegacyArchiveFixture(); + await writeFile(fixture.legacyIndexPath, "not json\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-index-invalid", + }); + await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe("not json\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("treats inconclusive filesystem inspection as a mutation-free skip", async () => { + const fixture = await createLegacyArchiveFixture(); + const inspectionError = Object.assign(new Error("inspection denied"), { code: "EACCES" }); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + lstat: (path) => path === fixture.destinationIndexPath + ? Promise.reject(inspectionError) + : lstat(path), + }, + }); + + expect(result).toMatchObject({ + status: "skipped", + reason: "inspection-failed", + error: inspectionError, + }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("does not merge with or overwrite an existing destination index", async () => { + const fixture = await createLegacyArchiveFixture(); + await mkdir(fixture.destinationRoot, { recursive: true }); + await writeFile(fixture.destinationIndexPath, "destination owner\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "destination-index-exists", + }); + await expect(readFile(fixture.destinationIndexPath, "utf8")).resolves.toBe("destination owner\n"); + await expectLegacyStateUntouched(fixture); + }); + + it("skips a non-empty destination archive directory without changing either side", async () => { + const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true }); + const destinationEntry = join(fixture.destinationArchiveDir, "already-here.jsonl"); + await writeFile(destinationEntry, "destination owner\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "destination-archive-not-empty-or-invalid", + }); + await expect(readFile(destinationEntry, "utf8")).resolves.toBe("destination owner\n"); + await expectLegacyStateUntouched(fixture); + }); + + it("skips archive paths that are not direct regular files in the legacy archive directory", async () => { + const fixture = await createLegacyArchiveFixture(); + const outsidePath = join(fixture.root, "outside.jsonl"); + await writeFile(outsidePath, "outside\n", "utf8"); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + const changedDocument = { + ...fixture.document, + sessions: [{ ...firstRecord, archivePath: outsidePath }, ...fixture.document.sessions.slice(1)], + }; + await writeArchiveIndex(fixture.legacyIndexPath, changedDocument); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-archive-layout-invalid", + }); + await expect(readFile(outsidePath, "utf8")).resolves.toBe("outside\n"); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips legacy directories containing unindexed entries", async () => { + const fixture = await createLegacyArchiveFixture(); + const unexpectedPath = join(fixture.legacyArchiveDir, "unexpected.tmp"); + await writeFile(unexpectedPath, "unexpected\n", "utf8"); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "legacy-archive-layout-invalid", + }); + await expect(readFile(unexpectedPath, "utf8")).resolves.toBe("unexpected\n"); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips duplicate session IDs", async () => { + const fixture = await createLegacyArchiveFixture(); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + const secondRecord = requiredRecord(fixture.document.sessions, 1); + await writeArchiveIndex(fixture.legacyIndexPath, { + ...fixture.document, + sessions: [firstRecord, { ...secondRecord, sessionId: firstRecord["sessionId"] }], + }); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "archive-record-conflict", + }); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); + + it("skips duplicate source and destination file mappings", async () => { + const fixture = await createLegacyArchiveFixture(); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + await writeArchiveIndex(fixture.legacyIndexPath, { + ...fixture.document, + sessions: [firstRecord, { ...firstRecord, sessionId: "second-file-record" }], + }); + + await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({ + status: "skipped", + reason: "archive-record-conflict", + }); + await expectLegacyStateUntouched({ + ...fixture, + sourceIndexContents: await readFile(fixture.legacyIndexPath, "utf8"), + }); + expect(await exists(fixture.destinationRoot)).toBe(false); + }); +}); + +describe("legacy session archive migration execution", () => { + afterEach(async () => { + await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true }))); + }); + + it("stages and verifies copies, atomically publishes the rewritten index, then removes legacy state", async () => { + const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true }); + const copies: { source: string; destination: string; mode: number }[] = []; + + await expect(migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + copyFile: async (source, destination, mode) => { + copies.push({ source, destination, mode }); + await copyFile(source, destination, mode); + }, + }, + })).resolves.toEqual({ status: "migrated", archiveFileCount: 1, cleanup: "complete" }); + + expect(copies).toHaveLength(2); + expect(copies[0]).toMatchObject({ source: fixture.legacyFilePath, mode: constants.COPYFILE_EXCL }); + expect(copies[0]?.destination).toContain(".archived-sessions-migration-test-attempt"); + expect(copies[1]).toMatchObject({ destination: fixture.destinationFilePath, mode: constants.COPYFILE_EXCL }); + await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n"); + + const migratedDocument: unknown = JSON.parse(await readFile(fixture.destinationIndexPath, "utf8")); + const firstRecord = requiredRecord(fixture.document.sessions, 0); + expect(migratedDocument).toEqual({ + ...fixture.document, + sessions: [ + { ...firstRecord, archivePath: fixture.destinationFilePath }, + requiredRecord(fixture.document.sessions, 1), + ], + }); + expect(await exists(fixture.legacyIndexPath)).toBe(false); + expect(await exists(fixture.legacyFilePath)).toBe(false); + expect(await exists(fixture.legacyArchiveDir)).toBe(false); + await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n"); + expect(new Set(await readdir(fixture.destinationRoot))).toEqual(new Set([ + "archived-sessions", + "archived-sessions.json", + ])); + }); + + it("rolls back destination artifacts and preserves all legacy state when staged-copy verification fails", async () => { + const fixture = await createLegacyArchiveFixture(); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + copyFile: async (source, destination, mode) => { + await copyFile(source, destination, mode); + if (source === fixture.legacyFilePath) await appendFile(destination, "corrupt", "utf8"); + }, + }, + }); + + expect(result).toMatchObject({ status: "failed", phase: "stage", rollbackErrors: [] }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationIndexPath)).toBe(false); + await expect(readdir(fixture.destinationRoot)).resolves.toEqual([]); + }); + + it("rolls back published files but never source state when atomic index publication fails", async () => { + const fixture = await createLegacyArchiveFixture(); + const publicationError = Object.assign(new Error("link failed"), { code: "EIO" }); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + link: () => Promise.reject(publicationError), + }, + }); + + expect(result).toMatchObject({ + status: "failed", + phase: "commit-index", + error: publicationError, + rollbackErrors: [], + }); + await expectLegacyStateUntouched(fixture); + expect(await exists(fixture.destinationIndexPath)).toBe(false); + expect(await exists(fixture.destinationArchiveDir)).toBe(false); + await expect(readdir(fixture.destinationRoot)).resolves.toEqual([]); + }); + + it("keeps the committed destination authoritative when legacy cleanup fails", async () => { + const fixture = await createLegacyArchiveFixture(); + const cleanupError = Object.assign(new Error("source cleanup failed"), { code: "EACCES" }); + + const result = await migrateLegacySessionArchive({ + ...fixture.options, + fileSystem: { + unlink: async (path) => { + if (path === fixture.legacyFilePath) throw cleanupError; + await unlink(path); + }, + }, + }); + + expect(result).toMatchObject({ + status: "migrated", + cleanup: "incomplete", + cleanupErrors: [cleanupError], + }); + await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n"); + expect(await exists(fixture.destinationIndexPath)).toBe(true); + await expectLegacyStateUntouched(fixture); + }); +}); + +interface LegacyArchiveFixture { + root: string; + legacyRoot: string; + legacyIndexPath: string; + legacyArchiveDir: string; + legacyFilePath: string; + activeFilePath: string; + destinationRoot: string; + destinationIndexPath: string; + destinationArchiveDir: string; + destinationFilePath: string; + sourceIndexContents: string; + document: { marker: string; sessions: Record[] }; + options: LegacySessionArchiveMigrationOptions; +} + +async function createLegacyArchiveFixture( + setup: { createDestinationArchive?: boolean } = {}, +): Promise { + const root = await mkdtemp(join(tmpdir(), "pi-web-archive-migration-")); + tempRoots.push(root); + const homeDir = join(root, "home"); + const legacyRoot = join(homeDir, ".pi-web"); + const legacyIndexPath = join(legacyRoot, "archived-sessions.json"); + const legacyArchiveDir = join(legacyRoot, "archived-sessions"); + const legacyFilePath = join(legacyArchiveDir, "2026-01-01_file-session.jsonl"); + const activeFilePath = join(root, "active", "2026-01-01_file-session.jsonl"); + const destinationRoot = join(root, "managed-state"); + const destinationIndexPath = join(destinationRoot, "archived-sessions.json"); + const destinationArchiveDir = join(destinationRoot, "archived-sessions"); + const destinationFilePath = join(destinationArchiveDir, "2026-01-01_file-session.jsonl"); + const document = { + marker: "preserve root metadata", + sessions: [ + { + sessionId: "file-session", + cwd: `${join(root, "workspace")}${sep}..${sep}workspace`, + archivedAt: "2026-01-01T00:02:00.000Z", + originalPath: activeFilePath, + archivePath: legacyFilePath, + created: "2026-01-01T00:00:00.000Z", + modified: "2026-01-01T00:01:00.000Z", + messageCount: 2, + firstMessage: "hello", + name: "Legacy file session", + customMetadata: { preserved: true }, + }, + { + sessionId: "metadata-only", + cwd: "/workspace", + archivedAt: "2026-01-02T00:00:00.000Z", + name: "Metadata only", + }, + ], + }; + + await mkdir(legacyArchiveDir, { recursive: true }); + await mkdir(join(root, "active"), { recursive: true }); + await writeFile(legacyFilePath, "legacy session\n", "utf8"); + await writeFile(activeFilePath, "active session\n", "utf8"); + const sourceIndexContents = await writeArchiveIndex(legacyIndexPath, document); + if (setup.createDestinationArchive === true) await mkdir(destinationArchiveDir, { recursive: true }); + + return { + root, + legacyRoot, + legacyIndexPath, + legacyArchiveDir, + legacyFilePath, + activeFilePath, + destinationRoot, + destinationIndexPath, + destinationArchiveDir, + destinationFilePath, + sourceIndexContents, + document, + options: { + env: { PI_WEB_DATA_DIR: destinationRoot }, + cwd: root, + homeDir, + createAttemptId: () => "test-attempt", + }, + }; +} + +async function writeArchiveIndex(path: string, document: unknown): Promise { + const contents = `${JSON.stringify(document, null, 2)}\n`; + await writeFile(path, contents, "utf8"); + return contents; +} + +async function expectLegacyStateUntouched(fixture: LegacyArchiveFixture): Promise { + await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe(fixture.sourceIndexContents); + await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe("legacy session\n"); + await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n"); +} + +function requiredRecord(records: Record[], index: number): Record { + const record = records[index]; + if (record === undefined) throw new Error(`Missing fixture record ${index.toString()}`); + return record; +} + +async function exists(path: string): Promise { + try { + await access(path, constants.F_OK); + return true; + } catch { + return false; + } +} diff --git a/src/server/sessions/sessionArchiveMigration.ts b/src/server/sessions/sessionArchiveMigration.ts new file mode 100644 index 0000000..4d43c5c --- /dev/null +++ b/src/server/sessions/sessionArchiveMigration.ts @@ -0,0 +1,781 @@ +import { randomUUID } from "node:crypto"; +import { constants, type Dirent, type Stats } from "node:fs"; +import { + copyFile, + link, + lstat, + mkdir, + open, + readFile, + readdir, + realpath, + rm, + rmdir, + unlink, + writeFile, +} from "node:fs/promises"; +import { homedir } from "node:os"; +import { basename, dirname, isAbsolute, join, relative, resolve } from "node:path"; +import { piWebDataDir } from "../../config.js"; +import { parseSessionArchiveFile, type ArchivedSessionRecord } from "./sessionArchiveStore.js"; + +export type LegacySessionArchiveMigrationSkipReason = + | "data-dir-not-configured" + | "data-dir-not-distinct" + | "legacy-index-missing" + | "legacy-index-invalid" + | "destination-index-exists" + | "destination-archive-not-empty-or-invalid" + | "legacy-archive-layout-invalid" + | "archive-record-conflict" + | "inspection-failed"; + +export interface LegacySessionArchiveMigrationSkipped { + status: "skipped"; + reason: LegacySessionArchiveMigrationSkipReason; + error?: unknown; +} + +export type LegacySessionArchiveMigrationPreflight = + | LegacySessionArchiveMigrationSkipped + | { + status: "eligible"; + legacyIndexPath: string; + destinationIndexPath: string; + archiveFileCount: number; + }; + +export type LegacySessionArchiveMigrationPhase = "stage" | "publish-files" | "commit-index"; + +export type LegacySessionArchiveMigrationResult = + | LegacySessionArchiveMigrationSkipped + | { + status: "failed"; + phase: LegacySessionArchiveMigrationPhase; + error: unknown; + rollbackErrors: unknown[]; + } + | { + status: "migrated"; + archiveFileCount: number; + cleanup: "complete"; + } + | { + status: "migrated"; + archiveFileCount: number; + cleanup: "incomplete"; + cleanupErrors: unknown[]; + }; + +export interface SessionArchiveMigrationReadHandle { + read(buffer: Buffer, offset: number, length: number, position: null): Promise<{ bytesRead: number }>; + close(): Promise; +} + +export interface SessionArchiveMigrationFileSystem { + lstat(path: string): Promise; + realpath(path: string): Promise; + readFile(path: string): Promise; + readdir(path: string): Promise; + mkdir(path: string, options?: { recursive?: boolean }): Promise; + copyFile(source: string, destination: string, mode: number): Promise; + writeFile(path: string, contents: string): Promise; + link(source: string, destination: string): Promise; + unlink(path: string): Promise; + rmdir(path: string): Promise; + rmOwnedTree(path: string): Promise; + open(path: string): Promise; +} + +export interface LegacySessionArchiveMigrationOptions { + env?: NodeJS.ProcessEnv; + cwd?: string; + homeDir?: string; + platform?: NodeJS.Platform; + createAttemptId?: () => string; + fileSystem?: Partial; +} + +interface PlannedArchiveFile { + sourcePath: string; + destinationPath: string; + fileName: string; +} + +interface MigrationPlan { + legacyRoot: string; + canonicalLegacyRoot: string; + legacyIndexPath: string; + legacyArchiveDir: string; + canonicalLegacyArchiveDir: string; + legacyArchiveDirExists: boolean; + destinationRoot: string; + canonicalDestinationRoot: string; + destinationIndexPath: string; + destinationArchiveDir: string; + canonicalDestinationArchiveDir: string; + destinationArchiveDirExists: boolean; + sourceIndexContents: string; + destinationIndexContents: string; + files: PlannedArchiveFile[]; + platform: NodeJS.Platform; +} + +interface InternalEligiblePreflight { + status: "eligible"; + plan: MigrationPlan; +} + +type InternalPreflight = LegacySessionArchiveMigrationSkipped | InternalEligiblePreflight; + +const defaultFileSystem: SessionArchiveMigrationFileSystem = { + lstat: async (path) => lstat(path), + realpath: async (path) => realpath(path), + readFile: async (path) => readFile(path, "utf8"), + readdir: async (path) => readdir(path, { withFileTypes: true }), + mkdir: async (path, options) => { + await mkdir(path, options); + }, + copyFile: async (source, destination, mode) => copyFile(source, destination, mode), + writeFile: async (path, contents) => { + await writeFile(path, contents, { encoding: "utf8", flag: "wx" }); + }, + link: async (source, destination) => link(source, destination), + unlink: async (path) => unlink(path), + rmdir: async (path) => rmdir(path), + rmOwnedTree: async (path) => { + await rm(path, { recursive: true, force: true }); + }, + open: async (path) => open(path, "r"), +}; + +export async function inspectLegacySessionArchiveMigration( + options: LegacySessionArchiveMigrationOptions = {}, +): Promise { + const preflight = await buildMigrationPreflight(options, migrationFileSystem(options)); + if (preflight.status === "skipped") return preflight; + return { + status: "eligible", + legacyIndexPath: preflight.plan.legacyIndexPath, + destinationIndexPath: preflight.plan.destinationIndexPath, + archiveFileCount: preflight.plan.files.length, + }; +} + +export async function migrateLegacySessionArchive( + options: LegacySessionArchiveMigrationOptions = {}, +): Promise { + const fileSystem = migrationFileSystem(options); + const preflight = await buildMigrationPreflight(options, fileSystem); + if (preflight.status === "skipped") return preflight; + + const plan = preflight.plan; + let phase: LegacySessionArchiveMigrationPhase = "stage"; + let stagingRoot: string | undefined; + let stagingCreated = false; + let destinationArchiveCreated = false; + const ownedDestinationFiles: string[] = []; + + try { + await fileSystem.mkdir(plan.destinationRoot, { recursive: true }); + const currentDestinationRoot = await canonicalizeAllowMissing(plan.destinationRoot, fileSystem); + if (!pathsEqual(currentDestinationRoot, plan.canonicalDestinationRoot, plan.platform)) { + throw new Error("Destination data root changed after migration preflight"); + } + await assertDestinationStateUnchanged(plan, fileSystem); + + const attemptId = safeAttemptId((options.createAttemptId ?? randomUUID)()); + // Keep staging beside, not inside, the authoritative archive directory so an + // interrupted staging copy cannot be mistaken for destination archive data. + stagingRoot = join(plan.destinationRoot, `.archived-sessions-migration-${attemptId}`); + await fileSystem.mkdir(stagingRoot); + stagingCreated = true; + const stagingFilesDir = join(stagingRoot, "files"); + await fileSystem.mkdir(stagingFilesDir); + + const stagedFiles = new Map(); + for (const file of plan.files) { + const stagedPath = join(stagingFilesDir, file.fileName); + await fileSystem.copyFile(file.sourcePath, stagedPath, constants.COPYFILE_EXCL); + if (!await filesEqual(file.sourcePath, stagedPath, fileSystem)) { + throw new Error(`Staged archive file verification failed: ${file.fileName}`); + } + stagedFiles.set(file.destinationPath, stagedPath); + } + + const stagedIndexPath = join(stagingRoot, "archived-sessions.json"); + await fileSystem.writeFile(stagedIndexPath, plan.destinationIndexContents); + if (await fileSystem.readFile(stagedIndexPath) !== plan.destinationIndexContents) { + throw new Error("Staged archive index verification failed"); + } + + phase = "publish-files"; + await assertDestinationStateUnchanged(plan, fileSystem); + if (plan.files.length > 0 && !plan.destinationArchiveDirExists) { + await fileSystem.mkdir(plan.destinationArchiveDir); + destinationArchiveCreated = true; + } + + for (const file of plan.files) { + const stagedPath = stagedFiles.get(file.destinationPath); + if (stagedPath === undefined) throw new Error(`Missing staged archive file: ${file.fileName}`); + try { + await fileSystem.copyFile(stagedPath, file.destinationPath, constants.COPYFILE_EXCL); + ownedDestinationFiles.push(file.destinationPath); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "EEXIST")) ownedDestinationFiles.push(file.destinationPath); + throw error; + } + if (!await filesEqual(stagedPath, file.destinationPath, fileSystem)) { + throw new Error(`Published archive file verification failed: ${file.fileName}`); + } + } + + phase = "commit-index"; + await assertPlanReadyToCommit(plan, stagedIndexPath, fileSystem); + // A same-directory hard link atomically publishes the already-verified index + // and, unlike rename on POSIX, fails rather than replacing an existing index. + await fileSystem.link(stagedIndexPath, plan.destinationIndexPath); + } catch (error: unknown) { + const rollbackErrors = await rollbackUncommittedDestination({ + stagingRoot, + stagingCreated, + destinationArchiveDir: plan.destinationArchiveDir, + destinationArchiveCreated, + ownedDestinationFiles, + }, fileSystem); + return { status: "failed", phase, error, rollbackErrors }; + } + + const cleanupErrors: unknown[] = []; + try { + await fileSystem.rmOwnedTree(stagingRoot); + } catch (error: unknown) { + cleanupErrors.push(error); + } + cleanupErrors.push(...await removeCommittedLegacyState(plan, fileSystem)); + + return cleanupErrors.length === 0 + ? { status: "migrated", archiveFileCount: plan.files.length, cleanup: "complete" } + : { status: "migrated", archiveFileCount: plan.files.length, cleanup: "incomplete", cleanupErrors }; +} + +async function buildMigrationPreflight( + options: LegacySessionArchiveMigrationOptions, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const env = options.env ?? process.env; + const configuredDataDir = env["PI_WEB_DATA_DIR"]; + if (configuredDataDir === undefined || configuredDataDir.trim() === "") { + return migrationSkipped("data-dir-not-configured"); + } + + const cwd = options.cwd ?? process.cwd(); + const platform = options.platform ?? process.platform; + const legacyRoot = resolve(options.homeDir ?? homedir(), ".pi-web"); + const destinationRoot = piWebDataDir(env, cwd); + const legacyIndexPath = join(legacyRoot, "archived-sessions.json"); + const legacyArchiveDir = join(legacyRoot, "archived-sessions"); + const destinationIndexPath = join(destinationRoot, "archived-sessions.json"); + const destinationArchiveDir = join(destinationRoot, "archived-sessions"); + + try { + const canonicalLegacyRoot = await canonicalizeAllowMissing(legacyRoot, fileSystem); + const canonicalDestinationRoot = await canonicalizeAllowMissing(destinationRoot, fileSystem); + if (pathsEqual(canonicalLegacyRoot, canonicalDestinationRoot, platform)) { + return migrationSkipped("data-dir-not-distinct"); + } + + const legacyIndexStats = await lstatIfExists(legacyIndexPath, fileSystem); + if (legacyIndexStats === undefined) return migrationSkipped("legacy-index-missing"); + if (!legacyIndexStats.isFile() || legacyIndexStats.isSymbolicLink()) { + return migrationSkipped("legacy-index-invalid"); + } + + const sourceIndexContents = await fileSystem.readFile(legacyIndexPath); + const parsedDocument = parseArchiveDocument(sourceIndexContents); + if (parsedDocument === undefined) return migrationSkipped("legacy-index-invalid"); + + if (await lstatIfExists(destinationIndexPath, fileSystem) !== undefined) { + return migrationSkipped("destination-index-exists"); + } + + const destinationArchiveStats = await lstatIfExists(destinationArchiveDir, fileSystem); + let destinationArchiveDirExists = false; + if (destinationArchiveStats !== undefined) { + if (!destinationArchiveStats.isDirectory() || destinationArchiveStats.isSymbolicLink()) { + return migrationSkipped("destination-archive-not-empty-or-invalid"); + } + if ((await fileSystem.readdir(destinationArchiveDir)).length !== 0) { + return migrationSkipped("destination-archive-not-empty-or-invalid"); + } + destinationArchiveDirExists = true; + } + + const legacyArchiveStats = await lstatIfExists(legacyArchiveDir, fileSystem); + if (legacyArchiveStats !== undefined && (!legacyArchiveStats.isDirectory() || legacyArchiveStats.isSymbolicLink())) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + const legacyArchiveDirExists = legacyArchiveStats !== undefined; + const canonicalLegacyArchiveDir = legacyArchiveDirExists + ? await fileSystem.realpath(legacyArchiveDir) + : await canonicalizeAllowMissing(legacyArchiveDir, fileSystem); + const canonicalDestinationArchiveDir = destinationArchiveDirExists + ? await fileSystem.realpath(destinationArchiveDir) + : join(canonicalDestinationRoot, "archived-sessions"); + if (pathsOverlap(canonicalLegacyArchiveDir, canonicalDestinationArchiveDir, platform)) { + return migrationSkipped("data-dir-not-distinct"); + } + + const plannedFiles = await planArchiveFiles({ + sessions: parsedDocument.sessions, + legacyArchiveDir, + canonicalLegacyArchiveDir, + legacyArchiveDirExists, + destinationArchiveDir, + canonicalDestinationArchiveDir, + platform, + }, fileSystem); + if (plannedFiles.status === "skipped") return plannedFiles; + + if (!await legacyDirectoryMatchesPlan(legacyArchiveDir, legacyArchiveDirExists, plannedFiles.files, platform, fileSystem)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + + const rewrittenSessions = parsedDocument.rawSessions.map((record, index) => { + const destinationPath = plannedFiles.destinationPathsByRecord[index]; + return destinationPath === undefined ? record : { ...record, archivePath: destinationPath }; + }); + const destinationIndexContents = `${JSON.stringify({ ...parsedDocument.rawDocument, sessions: rewrittenSessions }, null, 2)}\n`; + + return { + status: "eligible", + plan: { + legacyRoot, + canonicalLegacyRoot, + legacyIndexPath, + legacyArchiveDir, + canonicalLegacyArchiveDir, + legacyArchiveDirExists, + destinationRoot, + canonicalDestinationRoot, + destinationIndexPath, + destinationArchiveDir, + canonicalDestinationArchiveDir, + destinationArchiveDirExists, + sourceIndexContents, + destinationIndexContents, + files: plannedFiles.files, + platform, + }, + }; + } catch (error: unknown) { + return migrationSkipped("inspection-failed", error); + } +} + +interface ParsedArchiveDocument { + rawDocument: Record; + rawSessions: Record[]; + sessions: ArchivedSessionRecord[]; +} + +function parseArchiveDocument(contents: string): ParsedArchiveDocument | undefined { + try { + const value: unknown = JSON.parse(contents); + if (!isRecord(value)) return undefined; + const rawSessions = recordArray(value["sessions"]); + if (rawSessions === undefined) return undefined; + const archive = parseSessionArchiveFile(value); + return { rawDocument: value, rawSessions, sessions: archive.sessions }; + } catch { + return undefined; + } +} + +async function planArchiveFiles( + input: { + sessions: ArchivedSessionRecord[]; + legacyArchiveDir: string; + canonicalLegacyArchiveDir: string; + legacyArchiveDirExists: boolean; + destinationArchiveDir: string; + canonicalDestinationArchiveDir: string; + platform: NodeJS.Platform; + }, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const sessionIds = new Set(); + const sourcePaths = new Set(); + const destinationPaths = new Set(); + const destinationPathsByRecord: (string | undefined)[] = Array.from({ length: input.sessions.length }); + const files: PlannedArchiveFile[] = []; + + for (const [index, session] of input.sessions.entries()) { + if (session.sessionId.trim() === "" || sessionIds.has(session.sessionId)) { + return migrationSkipped("archive-record-conflict"); + } + sessionIds.add(session.sessionId); + + if (session.archivePath === undefined) continue; + if (!input.legacyArchiveDirExists || !isAbsolute(session.archivePath)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + + const sourcePath = resolve(session.archivePath); + if (!pathsEqual(dirname(sourcePath), input.legacyArchiveDir, input.platform)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + const sourceStats = await lstatIfExists(sourcePath, fileSystem); + if (sourceStats === undefined || !sourceStats.isFile() || sourceStats.isSymbolicLink()) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + const canonicalSourcePath = await fileSystem.realpath(sourcePath); + if (!pathsEqual(dirname(canonicalSourcePath), input.canonicalLegacyArchiveDir, input.platform)) { + return migrationSkipped("legacy-archive-layout-invalid"); + } + + const fileName = basename(sourcePath); + const destinationPath = join(input.destinationArchiveDir, fileName); + const canonicalDestinationPath = join(input.canonicalDestinationArchiveDir, fileName); + const sourceKey = pathKey(canonicalSourcePath, input.platform); + const destinationKey = collisionKey(canonicalDestinationPath); + if (sourcePaths.has(sourceKey) || destinationPaths.has(destinationKey)) { + return migrationSkipped("archive-record-conflict"); + } + sourcePaths.add(sourceKey); + destinationPaths.add(destinationKey); + destinationPathsByRecord[index] = destinationPath; + files.push({ sourcePath, destinationPath, fileName }); + } + + return { status: "planned", files, destinationPathsByRecord }; +} + +async function legacyDirectoryMatchesPlan( + legacyArchiveDir: string, + legacyArchiveDirExists: boolean, + files: PlannedArchiveFile[], + platform: NodeJS.Platform, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const archiveStats = await lstatIfExists(legacyArchiveDir, fileSystem); + if (!legacyArchiveDirExists) return archiveStats === undefined && files.length === 0; + if (archiveStats === undefined || !archiveStats.isDirectory() || archiveStats.isSymbolicLink()) return false; + const entries = await fileSystem.readdir(legacyArchiveDir); + if (entries.some((entry) => !entry.isFile() || entry.isSymbolicLink())) return false; + const actualNames = new Set(entries.map((entry) => pathNameKey(entry.name, platform))); + const expectedNames = new Set(files.map((file) => pathNameKey(file.fileName, platform))); + if (actualNames.size !== entries.length || expectedNames.size !== files.length || actualNames.size !== expectedNames.size) return false; + return [...expectedNames].every((name) => actualNames.has(name)); +} + +async function assertDestinationStateUnchanged( + plan: MigrationPlan, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + if (await lstatIfExists(plan.destinationIndexPath, fileSystem) !== undefined) { + throw new Error("Destination archive index appeared after migration preflight"); + } + const archiveStats = await lstatIfExists(plan.destinationArchiveDir, fileSystem); + if (!plan.destinationArchiveDirExists) { + if (archiveStats !== undefined) throw new Error("Destination archive directory appeared after migration preflight"); + return; + } + if (archiveStats === undefined || !archiveStats.isDirectory() || archiveStats.isSymbolicLink()) { + throw new Error("Destination archive directory changed after migration preflight"); + } + if ((await fileSystem.readdir(plan.destinationArchiveDir)).length !== 0) { + throw new Error("Destination archive directory is no longer empty"); + } +} + +async function assertPlanReadyToCommit( + plan: MigrationPlan, + stagedIndexPath: string, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const currentLegacyRoot = await canonicalizeAllowMissing(plan.legacyRoot, fileSystem); + const currentDestinationRoot = await canonicalizeAllowMissing(plan.destinationRoot, fileSystem); + if (!pathsEqual(currentLegacyRoot, plan.canonicalLegacyRoot, plan.platform) + || !pathsEqual(currentDestinationRoot, plan.canonicalDestinationRoot, plan.platform)) { + throw new Error("Archive data root changed during migration"); + } + const legacyIndexStats = await lstatIfExists(plan.legacyIndexPath, fileSystem); + if (legacyIndexStats === undefined || !legacyIndexStats.isFile() || legacyIndexStats.isSymbolicLink()) { + throw new Error("Legacy archive index changed during migration"); + } + if (await fileSystem.readFile(plan.legacyIndexPath) !== plan.sourceIndexContents) { + throw new Error("Legacy archive index changed during migration"); + } + if (!await legacyDirectoryMatchesPlan(plan.legacyArchiveDir, plan.legacyArchiveDirExists, plan.files, plan.platform, fileSystem)) { + throw new Error("Legacy archive directory changed during migration"); + } + if (plan.legacyArchiveDirExists + && !pathsEqual(await fileSystem.realpath(plan.legacyArchiveDir), plan.canonicalLegacyArchiveDir, plan.platform)) { + throw new Error("Legacy archive directory changed during migration"); + } + if (await lstatIfExists(plan.destinationIndexPath, fileSystem) !== undefined) { + throw new Error("Destination archive index appeared during migration"); + } + if (await fileSystem.readFile(stagedIndexPath) !== plan.destinationIndexContents) { + throw new Error("Staged archive index changed during migration"); + } + + if (plan.files.length === 0) { + const destinationStats = await lstatIfExists(plan.destinationArchiveDir, fileSystem); + if (plan.destinationArchiveDirExists) { + if (destinationStats === undefined || !destinationStats.isDirectory() || destinationStats.isSymbolicLink() + || !pathsEqual(await fileSystem.realpath(plan.destinationArchiveDir), plan.canonicalDestinationArchiveDir, plan.platform) + || (await fileSystem.readdir(plan.destinationArchiveDir)).length !== 0) { + throw new Error("Destination archive directory changed during migration"); + } + } else if (destinationStats !== undefined) { + throw new Error("Destination archive directory appeared during migration"); + } + return; + } + + const destinationStats = await lstatIfExists(plan.destinationArchiveDir, fileSystem); + if (destinationStats === undefined || !destinationStats.isDirectory() || destinationStats.isSymbolicLink()) { + throw new Error("Destination archive directory is not a real directory"); + } + if (!pathsEqual(await fileSystem.realpath(plan.destinationArchiveDir), plan.canonicalDestinationArchiveDir, plan.platform)) { + throw new Error("Destination archive directory changed during migration"); + } + const destinationEntries = await fileSystem.readdir(plan.destinationArchiveDir); + if (destinationEntries.some((entry) => !entry.isFile() || entry.isSymbolicLink())) { + throw new Error("Destination archive directory contains an unexpected entry"); + } + const actualNames = new Set(destinationEntries.map((entry) => pathNameKey(entry.name, plan.platform))); + const expectedNames = new Set(plan.files.map((file) => pathNameKey(file.fileName, plan.platform))); + if (actualNames.size !== destinationEntries.length || actualNames.size !== expectedNames.size + || ![...expectedNames].every((name) => actualNames.has(name))) { + throw new Error("Destination archive directory does not match the migration plan"); + } + + for (const file of plan.files) { + const sourceStats = await lstatIfExists(file.sourcePath, fileSystem); + const destinationFileStats = await lstatIfExists(file.destinationPath, fileSystem); + if (sourceStats === undefined || !sourceStats.isFile() || sourceStats.isSymbolicLink() + || destinationFileStats === undefined || !destinationFileStats.isFile() || destinationFileStats.isSymbolicLink()) { + throw new Error(`Archive file changed during migration: ${file.fileName}`); + } + if (!await filesEqual(file.sourcePath, file.destinationPath, fileSystem)) { + throw new Error(`Archive file verification failed before commit: ${file.fileName}`); + } + } +} + +async function rollbackUncommittedDestination( + state: { + stagingRoot: string | undefined; + stagingCreated: boolean; + destinationArchiveDir: string; + destinationArchiveCreated: boolean; + ownedDestinationFiles: string[]; + }, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const errors: unknown[] = []; + for (const path of [...state.ownedDestinationFiles].reverse()) { + try { + await removeFileIfPresent(path, fileSystem); + } catch (error: unknown) { + errors.push(error); + } + } + if (state.destinationArchiveCreated) { + try { + await removeDirectoryIfPresent(state.destinationArchiveDir, fileSystem); + } catch (error: unknown) { + errors.push(error); + } + } + if (state.stagingCreated && state.stagingRoot !== undefined) { + try { + await fileSystem.rmOwnedTree(state.stagingRoot); + } catch (error: unknown) { + errors.push(error); + } + } + return errors; +} + +async function removeCommittedLegacyState( + plan: MigrationPlan, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const errors: unknown[] = []; + // Remove the index last. If any file/directory cleanup fails, the complete + // legacy index remains as a recovery marker while the destination is valid. + for (const file of plan.files) { + try { + await removeFileIfPresent(file.sourcePath, fileSystem); + } catch (error: unknown) { + errors.push(error); + return errors; + } + } + if (plan.legacyArchiveDirExists) { + try { + await removeDirectoryIfPresent(plan.legacyArchiveDir, fileSystem); + } catch (error: unknown) { + errors.push(error); + return errors; + } + } + try { + await removeFileIfPresent(plan.legacyIndexPath, fileSystem); + } catch (error: unknown) { + errors.push(error); + } + return errors; +} + +async function filesEqual( + firstPath: string, + secondPath: string, + fileSystem: SessionArchiveMigrationFileSystem, +): Promise { + const first = await fileSystem.open(firstPath); + try { + const second = await fileSystem.open(secondPath); + try { + const firstBuffer = Buffer.allocUnsafe(64 * 1024); + const secondBuffer = Buffer.allocUnsafe(64 * 1024); + for (;;) { + const [firstBytes, secondBytes] = await Promise.all([ + readChunk(first, firstBuffer), + readChunk(second, secondBuffer), + ]); + if (firstBytes !== secondBytes) return false; + if (firstBytes === 0) return true; + if (!firstBuffer.subarray(0, firstBytes).equals(secondBuffer.subarray(0, secondBytes))) return false; + } + } finally { + await second.close(); + } + } finally { + await first.close(); + } +} + +async function readChunk(handle: SessionArchiveMigrationReadHandle, buffer: Buffer): Promise { + let total = 0; + while (total < buffer.length) { + const { bytesRead } = await handle.read(buffer, total, buffer.length - total, null); + if (bytesRead === 0) break; + total += bytesRead; + } + return total; +} + +async function canonicalizeAllowMissing(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + let cursor = resolve(path); + const missingParts: string[] = []; + for (;;) { + try { + const canonical = await fileSystem.realpath(cursor); + return resolve(canonical, ...missingParts); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "ENOENT")) throw error; + if (await lstatIfExists(cursor, fileSystem) !== undefined) { + throw new Error(`Cannot resolve existing path: ${cursor}`, { cause: error }); + } + const parent = dirname(cursor); + if (parent === cursor) throw error; + missingParts.unshift(basename(cursor)); + cursor = parent; + } + } +} + +async function lstatIfExists(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + try { + return await fileSystem.lstat(path); + } catch (error: unknown) { + if (isNodeErrorWithCode(error, "ENOENT")) return undefined; + throw error; + } +} + +async function removeFileIfPresent(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + try { + await fileSystem.unlink(path); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "ENOENT")) throw error; + } +} + +async function removeDirectoryIfPresent(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise { + try { + await fileSystem.rmdir(path); + } catch (error: unknown) { + if (!isNodeErrorWithCode(error, "ENOENT")) throw error; + } +} + +function migrationFileSystem(options: LegacySessionArchiveMigrationOptions): SessionArchiveMigrationFileSystem { + return { ...defaultFileSystem, ...options.fileSystem }; +} + +function migrationSkipped( + reason: LegacySessionArchiveMigrationSkipReason, + error?: unknown, +): LegacySessionArchiveMigrationSkipped { + return error === undefined ? { status: "skipped", reason } : { status: "skipped", reason, error }; +} + +function pathsOverlap(first: string, second: string, platform: NodeJS.Platform): boolean { + return pathContains(first, second, platform) || pathContains(second, first, platform); +} + +function pathContains(parent: string, candidate: string, platform: NodeJS.Platform): boolean { + const parentPath = pathKey(parent, platform); + const candidatePath = pathKey(candidate, platform); + const pathFromParent = relative(parentPath, candidatePath); + const firstSegment = pathFromParent.split(/[\\/]/, 1)[0]; + return pathFromParent === "" || (firstSegment !== ".." && !isAbsolute(pathFromParent)); +} + +function pathsEqual(first: string, second: string, platform: NodeJS.Platform): boolean { + return pathKey(first, platform) === pathKey(second, platform); +} + +function pathKey(path: string, platform: NodeJS.Platform): string { + const normalized = resolve(path); + return platform === "win32" ? normalized.toLowerCase() : normalized; +} + +function pathNameKey(name: string, platform: NodeJS.Platform): string { + return platform === "win32" ? name.toLowerCase() : name; +} + +function collisionKey(path: string): string { + // Conservatively reject case/normalization variants even when the source + // platform happens to allow them; the destination filesystem may not. + return resolve(path).normalize("NFC").toLowerCase(); +} + +function safeAttemptId(value: string): string { + return value.replace(/[^a-zA-Z0-9._-]/g, "_") || "attempt"; +} + +function recordArray(value: unknown): Record[] | undefined { + if (!Array.isArray(value)) return undefined; + const records: Record[] = []; + for (const item of value) { + if (!isRecord(item)) return undefined; + records.push(item); + } + return records; +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isNodeErrorWithCode(error: unknown, code: string): error is NodeJS.ErrnoException { + return error instanceof Error && "code" in error && error.code === code; +} diff --git a/src/server/sessions/sessionArchiveStore.ts b/src/server/sessions/sessionArchiveStore.ts index 53085cd..dd80a4e 100644 --- a/src/server/sessions/sessionArchiveStore.ts +++ b/src/server/sessions/sessionArchiveStore.ts @@ -31,7 +31,7 @@ export interface ArchivedSessionRecord { parentSessionPath?: string; } -interface ArchiveFile { +export interface SessionArchiveFile { sessions: ArchivedSessionRecord[]; } @@ -153,17 +153,17 @@ export class SessionArchiveStore { } } - private async read(): Promise { + private async read(): Promise { try { const value: unknown = JSON.parse(await readFile(this.filePath, "utf8")); - return parseArchiveFile(value); + return parseSessionArchiveFile(value); } catch (error: unknown) { if (isNodeErrorWithCode(error, "ENOENT")) return { sessions: [] }; throw error; } } - private async write(data: ArchiveFile): Promise { + private async write(data: SessionArchiveFile): Promise { await mkdir(dirname(this.filePath), { recursive: true }); const tempPath = join(dirname(this.filePath), `.${basename(this.filePath)}.${String(process.pid)}.${Date.now().toString()}.${randomUUID()}.tmp`); try { @@ -231,7 +231,7 @@ async function pathExists(path: string): Promise { } } -function parseArchiveFile(value: unknown): ArchiveFile { +export function parseSessionArchiveFile(value: unknown): SessionArchiveFile { if (!isRecord(value) || !Array.isArray(value["sessions"])) throw new Error("Invalid archive file"); return { sessions: value["sessions"].map(parseArchivedSessionRecord) }; }