Archived
chore: merge origin/main
This commit is contained in:
Federico Jaramillo Martinez
@@ -0,0 +1,555 @@
|
||||
import { constants } from "node:fs";
|
||||
import {
|
||||
access,
|
||||
appendFile,
|
||||
copyFile,
|
||||
link as createHardLink,
|
||||
lstat,
|
||||
mkdtemp,
|
||||
mkdir,
|
||||
readFile,
|
||||
readdir,
|
||||
realpath,
|
||||
rm,
|
||||
unlink,
|
||||
writeFile,
|
||||
} from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { dirname, join, resolve, 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("leaves an interrupted pre-commit file untouched instead of adopting a non-empty destination archive", async () => {
|
||||
const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true });
|
||||
await writeFile(fixture.destinationFilePath, "partial destination copy\n", "utf8");
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "skipped",
|
||||
reason: "destination-archive-not-empty-or-invalid",
|
||||
});
|
||||
await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("partial destination copy\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("uses Windows case-insensitive path semantics without weakening Linux containment", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const casedArchivePath = resolve(fixture.legacyFilePath.toUpperCase());
|
||||
const firstRecord = requiredRecord(fixture.document.sessions, 0);
|
||||
await writeArchiveIndex(fixture.legacyIndexPath, {
|
||||
...fixture.document,
|
||||
sessions: [
|
||||
{ ...firstRecord, archivePath: casedArchivePath },
|
||||
...fixture.document.sessions.slice(1),
|
||||
],
|
||||
});
|
||||
const mapCaseVariant = (path: string): string => path.toLowerCase() === casedArchivePath.toLowerCase()
|
||||
? fixture.legacyFilePath
|
||||
: path;
|
||||
const fileSystem = {
|
||||
lstat: (path: string) => lstat(mapCaseVariant(path)),
|
||||
realpath: (path: string) => realpath(mapCaseVariant(path)),
|
||||
};
|
||||
|
||||
await expect(inspectLegacySessionArchiveMigration({
|
||||
...fixture.options,
|
||||
platform: "linux",
|
||||
fileSystem,
|
||||
})).resolves.toEqual({ status: "skipped", reason: "legacy-archive-layout-invalid" });
|
||||
await expect(inspectLegacySessionArchiveMigration({
|
||||
...fixture.options,
|
||||
platform: "win32",
|
||||
fileSystem,
|
||||
})).resolves.toEqual({
|
||||
status: "eligible",
|
||||
legacyIndexPath: fixture.legacyIndexPath,
|
||||
destinationIndexPath: fixture.destinationIndexPath,
|
||||
archiveFileCount: 1,
|
||||
});
|
||||
});
|
||||
|
||||
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("copies across the source boundary, atomically publishes the rewritten index, then removes legacy state", async () => {
|
||||
const fixture = await createLegacyArchiveFixture({ createDestinationArchive: true });
|
||||
const copies: { source: string; destination: string; mode: number }[] = [];
|
||||
const links: { source: string; destination: string }[] = [];
|
||||
|
||||
await expect(migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
copyFile: async (source, destination, mode) => {
|
||||
copies.push({ source, destination, mode });
|
||||
await copyFile(source, destination, mode);
|
||||
},
|
||||
link: async (source, destination) => {
|
||||
links.push({ source, destination });
|
||||
await createHardLink(source, destination);
|
||||
},
|
||||
unlink: async (path) => {
|
||||
if (path === fixture.legacyFilePath || path === fixture.legacyIndexPath) {
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(true);
|
||||
}
|
||||
await unlink(path);
|
||||
},
|
||||
},
|
||||
})).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 });
|
||||
expect(links).toEqual([{
|
||||
source: join(fixture.destinationRoot, ".archived-sessions-migration-test-attempt", "archived-sessions.json"),
|
||||
destination: fixture.destinationIndexPath,
|
||||
}]);
|
||||
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("retries safely when an interrupted staging-only attempt left an unowned sibling tree", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const abandonedFile = join(
|
||||
fixture.destinationRoot,
|
||||
".archived-sessions-migration-interrupted-attempt",
|
||||
"files",
|
||||
"abandoned.jsonl",
|
||||
);
|
||||
await mkdir(dirname(abandonedFile), { recursive: true });
|
||||
await writeFile(abandonedFile, "unowned staging data\n", "utf8");
|
||||
|
||||
await expect(migrateLegacySessionArchive(fixture.options)).resolves.toEqual({
|
||||
status: "migrated",
|
||||
archiveFileCount: 1,
|
||||
cleanup: "complete",
|
||||
});
|
||||
|
||||
await expect(readFile(abandonedFile, "utf8")).resolves.toBe("unowned staging data\n");
|
||||
await expect(readFile(fixture.destinationFilePath, "utf8")).resolves.toBe("legacy session\n");
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(true);
|
||||
expect(await exists(fixture.legacyIndexPath)).toBe(false);
|
||||
});
|
||||
|
||||
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("revalidates source files before commit and rolls back if one changes during migration", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
copyFile: async (source, destination, mode) => {
|
||||
await copyFile(source, destination, mode);
|
||||
if (destination === fixture.destinationFilePath) {
|
||||
await appendFile(fixture.legacyFilePath, "changed during migration\n", "utf8");
|
||||
}
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({ status: "failed", phase: "commit-index", rollbackErrors: [] });
|
||||
await expect(readFile(fixture.legacyIndexPath, "utf8")).resolves.toBe(fixture.sourceIndexContents);
|
||||
await expect(readFile(fixture.legacyFilePath, "utf8")).resolves.toBe(
|
||||
"legacy session\nchanged during migration\n",
|
||||
);
|
||||
await expect(readFile(fixture.activeFilePath, "utf8")).resolves.toBe("active session\n");
|
||||
expect(await exists(fixture.destinationIndexPath)).toBe(false);
|
||||
expect(await exists(fixture.destinationArchiveDir)).toBe(false);
|
||||
});
|
||||
|
||||
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("does not overwrite a destination index that appears at the atomic commit boundary", async () => {
|
||||
const fixture = await createLegacyArchiveFixture();
|
||||
const publicationError = Object.assign(new Error("destination index won the race"), { code: "EEXIST" });
|
||||
|
||||
const result = await migrateLegacySessionArchive({
|
||||
...fixture.options,
|
||||
fileSystem: {
|
||||
link: async (_source, destination) => {
|
||||
await writeFile(destination, "destination owner\n", { encoding: "utf8", flag: "wx" });
|
||||
throw publicationError;
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
expect(result).toMatchObject({
|
||||
status: "failed",
|
||||
phase: "commit-index",
|
||||
error: publicationError,
|
||||
rollbackErrors: [],
|
||||
});
|
||||
await expect(readFile(fixture.destinationIndexPath, "utf8")).resolves.toBe("destination owner\n");
|
||||
expect(await exists(fixture.destinationArchiveDir)).toBe(false);
|
||||
await expectLegacyStateUntouched(fixture);
|
||||
});
|
||||
|
||||
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<string, unknown>[] };
|
||||
options: LegacySessionArchiveMigrationOptions;
|
||||
}
|
||||
|
||||
async function createLegacyArchiveFixture(
|
||||
setup: { createDestinationArchive?: boolean } = {},
|
||||
): Promise<LegacyArchiveFixture> {
|
||||
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<string> {
|
||||
const contents = `${JSON.stringify(document, null, 2)}\n`;
|
||||
await writeFile(path, contents, "utf8");
|
||||
return contents;
|
||||
}
|
||||
|
||||
async function expectLegacyStateUntouched(fixture: LegacyArchiveFixture): Promise<void> {
|
||||
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<string, unknown>[], index: number): Record<string, unknown> {
|
||||
const record = records[index];
|
||||
if (record === undefined) throw new Error(`Missing fixture record ${index.toString()}`);
|
||||
return record;
|
||||
}
|
||||
|
||||
async function exists(path: string): Promise<boolean> {
|
||||
try {
|
||||
await access(path, constants.F_OK);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -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<void>;
|
||||
}
|
||||
|
||||
export interface SessionArchiveMigrationFileSystem {
|
||||
lstat(path: string): Promise<Stats>;
|
||||
realpath(path: string): Promise<string>;
|
||||
readFile(path: string): Promise<string>;
|
||||
readdir(path: string): Promise<Dirent[]>;
|
||||
mkdir(path: string, options?: { recursive?: boolean }): Promise<void>;
|
||||
copyFile(source: string, destination: string, mode: number): Promise<void>;
|
||||
writeFile(path: string, contents: string): Promise<void>;
|
||||
link(source: string, destination: string): Promise<void>;
|
||||
unlink(path: string): Promise<void>;
|
||||
rmdir(path: string): Promise<void>;
|
||||
rmOwnedTree(path: string): Promise<void>;
|
||||
open(path: string): Promise<SessionArchiveMigrationReadHandle>;
|
||||
}
|
||||
|
||||
export interface LegacySessionArchiveMigrationOptions {
|
||||
env?: NodeJS.ProcessEnv;
|
||||
cwd?: string;
|
||||
homeDir?: string;
|
||||
platform?: NodeJS.Platform;
|
||||
createAttemptId?: () => string;
|
||||
fileSystem?: Partial<SessionArchiveMigrationFileSystem>;
|
||||
}
|
||||
|
||||
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<LegacySessionArchiveMigrationPreflight> {
|
||||
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<LegacySessionArchiveMigrationResult> {
|
||||
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<string, string>();
|
||||
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<InternalPreflight> {
|
||||
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<string, unknown>;
|
||||
rawSessions: Record<string, unknown>[];
|
||||
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<LegacySessionArchiveMigrationSkipped | { status: "planned"; files: PlannedArchiveFile[]; destinationPathsByRecord: (string | undefined)[] }> {
|
||||
const sessionIds = new Set<string>();
|
||||
const sourcePaths = new Set<string>();
|
||||
const destinationPaths = new Set<string>();
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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<unknown[]> {
|
||||
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<unknown[]> {
|
||||
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<boolean> {
|
||||
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<number> {
|
||||
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<string> {
|
||||
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<Stats | undefined> {
|
||||
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<void> {
|
||||
try {
|
||||
await fileSystem.unlink(path);
|
||||
} catch (error: unknown) {
|
||||
if (!isNodeErrorWithCode(error, "ENOENT")) throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function removeDirectoryIfPresent(path: string, fileSystem: SessionArchiveMigrationFileSystem): Promise<void> {
|
||||
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<string, unknown>[] | undefined {
|
||||
if (!Array.isArray(value)) return undefined;
|
||||
const records: Record<string, unknown>[] = [];
|
||||
for (const item of value) {
|
||||
if (!isRecord(item)) return undefined;
|
||||
records.push(item);
|
||||
}
|
||||
return records;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
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;
|
||||
}
|
||||
@@ -1,17 +1,55 @@
|
||||
import { constants } from "node:fs";
|
||||
import { access, mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { join, resolve } from "node:path";
|
||||
import { tmpdir } from "node:os";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import { SessionArchiveStore } from "./sessionArchiveStore.js";
|
||||
import { homedir, tmpdir } from "node:os";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { defaultSessionArchiveFilePath, SessionArchiveStore } from "./sessionArchiveStore.js";
|
||||
|
||||
const tempRoots: string[] = [];
|
||||
|
||||
describe("defaultSessionArchiveFilePath", () => {
|
||||
it("uses PI_WEB_DATA_DIR when configured", () => {
|
||||
expect(defaultSessionArchiveFilePath({ PI_WEB_DATA_DIR: "managed-state" }, "/tmp/pi-web")).toBe(resolve("/tmp/pi-web", "managed-state", "archived-sessions.json"));
|
||||
});
|
||||
|
||||
it("preserves the ~/.pi-web default when PI_WEB_DATA_DIR is unset", () => {
|
||||
expect(defaultSessionArchiveFilePath({}, "/tmp/pi-web")).toBe(join(homedir(), ".pi-web", "archived-sessions.json"));
|
||||
});
|
||||
});
|
||||
|
||||
describe("SessionArchiveStore", () => {
|
||||
afterEach(async () => {
|
||||
vi.unstubAllEnvs();
|
||||
await Promise.all(tempRoots.splice(0).map((path) => rm(path, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
it("stores its default index and archived files under PI_WEB_DATA_DIR", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "pi-web-archive-data-dir-"));
|
||||
tempRoots.push(root);
|
||||
const dataDir = join(root, "managed-state");
|
||||
const activeDir = join(root, "active");
|
||||
await mkdir(activeDir, { recursive: true });
|
||||
const sourcePath = join(activeDir, "2026-01-01_managed.jsonl");
|
||||
await writeFile(sourcePath, "session contents\n", "utf8");
|
||||
vi.stubEnv("PI_WEB_DATA_DIR", dataDir);
|
||||
|
||||
const store = new SessionArchiveStore();
|
||||
const record = await store.archive({
|
||||
sessionId: "managed",
|
||||
cwd: "/workspace",
|
||||
path: sourcePath,
|
||||
created: "2026-01-01T00:00:00.000Z",
|
||||
modified: "2026-01-01T00:01:00.000Z",
|
||||
messageCount: 1,
|
||||
firstMessage: "hello",
|
||||
});
|
||||
|
||||
const archivePath = join(dataDir, "archived-sessions", "2026-01-01_managed.jsonl");
|
||||
expect(record.archivePath).toBe(archivePath);
|
||||
await expect(readFile(archivePath, "utf8")).resolves.toBe("session contents\n");
|
||||
await expect(readFile(join(dataDir, "archived-sessions.json"), "utf8")).resolves.toContain('"sessionId": "managed"');
|
||||
});
|
||||
|
||||
it("moves archived session files out of the active session directory and restores them", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "pi-web-archive-"));
|
||||
tempRoots.push(root);
|
||||
|
||||
@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto";
|
||||
import { constants } from "node:fs";
|
||||
import { access, copyFile, mkdir, readFile, rename, unlink, writeFile } from "node:fs/promises";
|
||||
import { basename, dirname, join } from "node:path";
|
||||
import { homedir } from "node:os";
|
||||
import { piWebDataDir } from "../../config.js";
|
||||
import { canonicalizeStoredCwd } from "../workingDirectory.js";
|
||||
|
||||
export interface ArchiveSessionInput {
|
||||
@@ -31,15 +31,19 @@ export interface ArchivedSessionRecord {
|
||||
parentSessionPath?: string;
|
||||
}
|
||||
|
||||
interface ArchiveFile {
|
||||
export interface SessionArchiveFile {
|
||||
sessions: ArchivedSessionRecord[];
|
||||
}
|
||||
|
||||
export function defaultSessionArchiveFilePath(env: NodeJS.ProcessEnv = process.env, cwd = process.cwd()): string {
|
||||
return join(piWebDataDir(env, cwd), "archived-sessions.json");
|
||||
}
|
||||
|
||||
export class SessionArchiveStore {
|
||||
private operationQueue: Promise<void> = Promise.resolve();
|
||||
|
||||
constructor(
|
||||
private readonly filePath = join(homedir(), ".pi-web", "archived-sessions.json"),
|
||||
private readonly filePath = defaultSessionArchiveFilePath(),
|
||||
private readonly archiveDir = join(dirname(filePath), "archived-sessions"),
|
||||
) {}
|
||||
|
||||
@@ -149,17 +153,17 @@ export class SessionArchiveStore {
|
||||
}
|
||||
}
|
||||
|
||||
private async read(): Promise<ArchiveFile> {
|
||||
private async read(): Promise<SessionArchiveFile> {
|
||||
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<void> {
|
||||
private async write(data: SessionArchiveFile): Promise<void> {
|
||||
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 {
|
||||
@@ -227,7 +231,7 @@ async function pathExists(path: string): Promise<boolean> {
|
||||
}
|
||||
}
|
||||
|
||||
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) };
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user