281 lines
17 KiB
TypeScript
281 lines
17 KiB
TypeScript
import { constants } from "node:fs";
|
|
import { createHash, randomUUID } from "node:crypto";
|
|
import { copyFile, chmod, link, lstat, mkdir, readFile, readdir, rm, writeFile } from "node:fs/promises";
|
|
import { join, relative, resolve, sep } from "node:path";
|
|
import {
|
|
ensureSessionWorkspaceDirectory, resolveConversationWorkspace, sessionWorkspaceDirectory,
|
|
type SessionWorkspaceContext, type SessionWorkspaceScope,
|
|
} from "../runtime/conversationWorkspace.js";
|
|
import { atomicWriteJson, listJsonFiles, readJsonFile } from "../utils/fileStore.js";
|
|
import { SessionContextMismatchError, type SessionRecord } from "./metadataStore.js";
|
|
|
|
type ChatRecordName = "messages.json" | "transcript.json" | "learning.json" | "answer-view-state.json" | "interaction-requests.json";
|
|
type LegacySources = { metadataDir: string; uiStateDir: string; transcriptDir: string; learningStateDir?: string };
|
|
const exists = async (path: string) => {
|
|
try { await lstat(path); return true; }
|
|
catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return false; throw error; }
|
|
};
|
|
|
|
const createFileExclusively = async (path: string, write: (temporary: string) => Promise<void>) => {
|
|
const temporary = `${path}.${randomUUID()}.tmp`;
|
|
try {
|
|
await write(temporary);
|
|
await link(temporary, path);
|
|
return true;
|
|
} catch (error) {
|
|
if ((error as NodeJS.ErrnoException).code === "EEXIST") return false;
|
|
throw error;
|
|
} finally { await rm(temporary, { force: true }); }
|
|
};
|
|
|
|
export class SessionWorkspaceStore {
|
|
private readonly writeQueues = new Map<string, Promise<void>>();
|
|
private legacySources?: LegacySources;
|
|
private deletionCheck?: (context: SessionWorkspaceContext) => Promise<boolean>;
|
|
constructor(readonly root: string, readonly backupRoot = resolve(root, "..", "backup", "legacy-chat")) {}
|
|
|
|
setDeletionCheck(check: (context: SessionWorkspaceContext) => Promise<boolean>) {
|
|
this.deletionCheck = check;
|
|
}
|
|
|
|
path(context: SessionWorkspaceContext, name: ChatRecordName) {
|
|
return join(sessionWorkspaceDirectory(this.root, context), "chat", name);
|
|
}
|
|
|
|
async ensure(context: SessionWorkspaceContext, allowDeleted = false) {
|
|
const directory = await ensureSessionWorkspaceDirectory(this.root, context);
|
|
const chat = join(directory, "chat");
|
|
await mkdir(chat, { recursive: true, mode: 0o700 });
|
|
if ((await lstat(chat)).isSymbolicLink()) throw new Error("chat directory must not be a symbolic link");
|
|
const ownerPath = join(chat, "owner.json");
|
|
if (!await exists(ownerPath)) {
|
|
const owner = { sessionId: context.sessionId, actorKey: context.actorKey, projectKey: context.projectKey };
|
|
await createFileExclusively(ownerPath, (temporary) => writeFile(temporary, JSON.stringify(owner, null, 2), { flag: "wx", mode: 0o600 }));
|
|
}
|
|
if (!await this.validate(context, undefined, allowDeleted)) throw new SessionContextMismatchError();
|
|
return directory;
|
|
}
|
|
|
|
async validate(context: SessionWorkspaceContext, name?: ChatRecordName, allowDeleted = false) {
|
|
const directory = sessionWorkspaceDirectory(this.root, context);
|
|
if (!await exists(directory)) return false;
|
|
// Check each component before following it, including the persisted record.
|
|
let current = resolve(this.root);
|
|
for (const part of ["", ...relative(current, join(directory, "chat")).split(sep)]) {
|
|
if (part) current = join(current, part);
|
|
if (!await exists(current)) return false;
|
|
if ((await lstat(current)).isSymbolicLink()) throw new Error("chat workspace path must not be a symbolic link");
|
|
}
|
|
const ownerPath = join(directory, "chat", "owner.json");
|
|
if (!await exists(ownerPath)) throw new Error("chat workspace owner is missing");
|
|
if ((await lstat(ownerPath)).isSymbolicLink()) throw new Error("chat workspace owner must not be a symbolic link");
|
|
const owner = await readJsonFile<SessionWorkspaceContext>(ownerPath);
|
|
if (!owner || owner.actorKey !== context.actorKey || owner.projectKey !== context.projectKey || owner.sessionId !== context.sessionId) {
|
|
throw new Error("chat workspace ownership mismatch");
|
|
}
|
|
if (name && await exists(this.path(context, name)) && (await lstat(this.path(context, name))).isSymbolicLink()) {
|
|
throw new Error("chat record must not be a symbolic link");
|
|
}
|
|
const deletedPath = join(directory, "chat", "deleted.json");
|
|
if (await exists(deletedPath)) {
|
|
if ((await lstat(deletedPath)).isSymbolicLink()) throw new Error("chat deletion marker must not be a symbolic link");
|
|
const deleted = await readJsonFile<SessionWorkspaceContext>(deletedPath);
|
|
if (!deleted || deleted.actorKey !== context.actorKey || deleted.projectKey !== context.projectKey || deleted.sessionId !== context.sessionId) throw new Error("chat deletion ownership mismatch");
|
|
if (!allowDeleted) return false;
|
|
}
|
|
return allowDeleted || !await this.deletionCheck?.(context);
|
|
}
|
|
|
|
private async serialize<T>(context: SessionWorkspaceContext, operation: () => Promise<T>) {
|
|
const key = sessionWorkspaceDirectory(this.root, context);
|
|
const previous = this.writeQueues.get(key) ?? Promise.resolve();
|
|
const result = previous.catch(() => undefined).then(operation);
|
|
const pending = result.then(() => undefined, () => undefined);
|
|
this.writeQueues.set(key, pending);
|
|
try { return await result; }
|
|
finally { if (this.writeQueues.get(key) === pending) this.writeQueues.delete(key); }
|
|
}
|
|
|
|
async writeRecord(context: SessionWorkspaceContext, name: ChatRecordName, value: object) {
|
|
return this.serialize(context, async () => {
|
|
await this.ensure(context);
|
|
if (!await this.validate(context, name)) throw new SessionContextMismatchError();
|
|
await atomicWriteJson(this.path(context, name), value as Record<string, unknown>, 0o600);
|
|
});
|
|
}
|
|
|
|
/** Read, compare and replace under the same queue as message writes and deletion. */
|
|
async updateRecord<T>(
|
|
context: SessionWorkspaceContext,
|
|
name: ChatRecordName,
|
|
operation: (records: { messages: unknown; current: unknown }) => Promise<{ result: T; value?: object }>,
|
|
): Promise<T> {
|
|
return this.serialize(context, async () => {
|
|
if (!await this.validate(context, "messages.json") || !await this.validate(context, name)) throw new SessionContextMismatchError();
|
|
const messages = await readJsonFile<unknown>(this.path(context, "messages.json"));
|
|
const current = await readJsonFile<unknown>(this.path(context, name));
|
|
const update = await operation({ messages, current });
|
|
if (!await this.validate(context, "messages.json") || !await this.validate(context, name)) throw new SessionContextMismatchError();
|
|
if (update.value !== undefined) await atomicWriteJson(this.path(context, name), update.value as Record<string, unknown>, 0o600);
|
|
return update.result;
|
|
});
|
|
}
|
|
|
|
async markDeleted(context: SessionWorkspaceContext) {
|
|
return this.serialize(context, async () => {
|
|
const directory = await this.ensure(context, true);
|
|
const path = join(directory, "chat", "deleted.json");
|
|
if (!await exists(path)) await atomicWriteJson(path, {
|
|
sessionId: context.sessionId, actorKey: context.actorKey, projectKey: context.projectKey,
|
|
deletedAt: new Date().toISOString(),
|
|
}, 0o600);
|
|
});
|
|
}
|
|
|
|
async purge(context: SessionWorkspaceContext) {
|
|
return this.serialize(context, async () => {
|
|
if (!await this.validate(context, undefined, true)) return;
|
|
const directory = sessionWorkspaceDirectory(this.root, context);
|
|
if (!await exists(join(directory, "chat", "deleted.json"))) throw new Error("chat must be marked deleted before cleanup");
|
|
const chat = join(directory, "chat");
|
|
for (const name of await readdir(chat)) {
|
|
if (name === "owner.json" || name === "deleted.json") continue;
|
|
const file = join(chat, name);
|
|
if ((await lstat(file)).isSymbolicLink()) throw new Error("chat record must not be a symbolic link");
|
|
await rm(file, { recursive: true, force: true });
|
|
}
|
|
const workspace = join(directory, "workspace");
|
|
if (await exists(workspace)) await rm(await resolveConversationWorkspace(workspace, this.root), { recursive: true });
|
|
if (this.legacySources) await this.removeLegacyRecords(context, this.legacySources);
|
|
const archives = new SessionWorkspaceStore(this.backupRoot);
|
|
if (await archives.validate(context, undefined, true)) {
|
|
await rm(sessionWorkspaceDirectory(archives.root, context), { recursive: true });
|
|
}
|
|
});
|
|
}
|
|
|
|
private async removeLegacyRecords(context: SessionWorkspaceContext, sources: LegacySources) {
|
|
for (const [directory, scoped] of [[sources.uiStateDir, false], [sources.transcriptDir, true], [sources.learningStateDir, false]] as const) {
|
|
if (!directory) continue;
|
|
if (await exists(directory) && (await lstat(directory)).isSymbolicLink()) throw new Error("legacy storage must not be a symbolic link");
|
|
for (const file of await listJsonFiles(directory)) {
|
|
if ((await lstat(file)).isSymbolicLink()) throw new Error("legacy chat record must not be a symbolic link");
|
|
const record = await readJsonFile<Partial<SessionWorkspaceContext> & { clientSessionId?: string }>(file);
|
|
if ((record?.clientSessionId || record?.sessionId) !== context.sessionId) continue;
|
|
if (scoped && (record?.actorKey !== context.actorKey || record.projectKey !== context.projectKey)) continue;
|
|
await rm(file);
|
|
}
|
|
}
|
|
}
|
|
|
|
async list(scope: SessionWorkspaceScope, name: ChatRecordName) {
|
|
const probe = sessionWorkspaceDirectory(this.root, { ...scope, sessionId: "list" });
|
|
const project = resolve(probe, "..");
|
|
if (!await exists(project)) return [];
|
|
for (const path of [this.root, resolve(project, ".."), project]) {
|
|
if ((await lstat(path)).isSymbolicLink()) throw new Error("chat workspace path must not be a symbolic link");
|
|
}
|
|
const paths: string[] = [];
|
|
for (const entry of await readdir(project, { withFileTypes: true })) {
|
|
if (!entry.isDirectory() || !/^session-[a-f0-9]{64}$/.test(entry.name)) continue;
|
|
const chat = join(project, entry.name, "chat");
|
|
if (await exists(chat) && (await lstat(chat)).isSymbolicLink()) throw new Error("chat directory must not be a symbolic link");
|
|
const ownerPath = join(project, entry.name, "chat", "owner.json");
|
|
if (!await exists(ownerPath)) continue;
|
|
if ((await lstat(ownerPath)).isSymbolicLink()) throw new Error("chat workspace owner must not be a symbolic link");
|
|
const owner = await readJsonFile<SessionWorkspaceContext>(ownerPath);
|
|
if (!owner || owner.actorKey !== scope.actorKey || owner.projectKey !== scope.projectKey) continue;
|
|
if (resolve(sessionWorkspaceDirectory(this.root, owner)) !== resolve(project, entry.name)) throw new Error("chat workspace ownership mismatch");
|
|
if (await this.validate(owner, name) && await exists(this.path(owner, name))) paths.push(this.path(owner, name));
|
|
}
|
|
return paths;
|
|
}
|
|
|
|
async migrateLegacyRecords(options: LegacySources) {
|
|
this.legacySources = options;
|
|
const sourceStamp = async () => Promise.all([options.metadataDir, options.uiStateDir, options.transcriptDir, options.learningStateDir].filter((path): path is string => Boolean(path)).map(async (path) => {
|
|
try { const value = await lstat(path); if (value.isSymbolicLink()) throw new Error("legacy storage must not be a symbolic link"); return [resolve(path), value.mtimeMs]; }
|
|
catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return [resolve(path), null]; throw error; }
|
|
}));
|
|
await mkdir(this.backupRoot, { recursive: true, mode: 0o700 });
|
|
if ((await lstat(this.backupRoot)).isSymbolicLink()) throw new Error("legacy backup root must not be a symbolic link");
|
|
const manifestPath = join(this.backupRoot, "migration.json");
|
|
if (await exists(manifestPath) && (await lstat(manifestPath)).isSymbolicLink()) throw new Error("migration manifest must not be a symbolic link");
|
|
const stamp = await sourceStamp();
|
|
const manifest = await readJsonFile<{ version: number; sources: unknown }>(manifestPath);
|
|
const result = { uiStates: 0, transcripts: 0, learningStates: 0, retainedUnownedUiStates: 0 };
|
|
if (manifest?.version === 1 && JSON.stringify(manifest.sources) === JSON.stringify(stamp)) return result;
|
|
const owners = new Map<string, SessionWorkspaceContext>();
|
|
const deleted = new Map<string, SessionWorkspaceContext>();
|
|
for (const file of await listJsonFiles(options.metadataDir)) {
|
|
const record = await readJsonFile<SessionRecord>(file);
|
|
if (!record?.sessionId || !record.actorKey || !record.projectKey) continue;
|
|
const context = { sessionId: record.sessionId, actorKey: record.actorKey, projectKey: record.projectKey };
|
|
if (record.deletedAt) { await this.markDeleted(context); deleted.set(record.sessionId, context); continue; }
|
|
await this.ensure(context);
|
|
owners.set(record.sessionId, context);
|
|
}
|
|
for (const file of await listJsonFiles(options.uiStateDir)) {
|
|
const record = await readJsonFile<{ sessionId: string }>(file);
|
|
if (record && deleted.has(record.sessionId)) { await rm(file); continue; }
|
|
const owner = record && owners.get(record.sessionId);
|
|
if (!owner) { result.retainedUnownedUiStates++; continue; }
|
|
if (await this.copyLegacyRecord(owner, "messages.json", file)) result.uiStates++;
|
|
await this.archiveLegacyRecord(owner, "messages.json", file);
|
|
}
|
|
for (const file of await listJsonFiles(options.transcriptDir)) {
|
|
const record = await readJsonFile<{ actorKey: string; projectKey: string; sessionId: string; clientSessionId?: string }>(file);
|
|
if (!record?.actorKey || !record.projectKey || !record.sessionId) continue;
|
|
const context = { actorKey: record.actorKey, projectKey: record.projectKey, sessionId: record.clientSessionId || record.sessionId };
|
|
const removed = deleted.get(context.sessionId);
|
|
if (removed) {
|
|
if (removed.actorKey !== context.actorKey || removed.projectKey !== context.projectKey) throw new Error("legacy transcript ownership mismatch");
|
|
await rm(file); continue;
|
|
}
|
|
const owner = owners.get(context.sessionId);
|
|
if (owner && (owner.actorKey !== context.actorKey || owner.projectKey !== context.projectKey)) throw new Error("legacy transcript ownership mismatch");
|
|
if (await this.copyLegacyRecord(context, "transcript.json", file)) result.transcripts++;
|
|
await this.archiveLegacyRecord(context, "transcript.json", file);
|
|
}
|
|
if (options.learningStateDir) for (const file of await listJsonFiles(options.learningStateDir)) {
|
|
const record = await readJsonFile<{ sessionId: string }>(file);
|
|
if (record && deleted.has(record.sessionId)) { await rm(file); continue; }
|
|
const owner = record && owners.get(record.sessionId);
|
|
if (!owner) continue;
|
|
if (await this.copyLegacyRecord(owner, "learning.json", file)) result.learningStates++;
|
|
await this.archiveLegacyRecord(owner, "learning.json", file);
|
|
}
|
|
await atomicWriteJson(manifestPath, { version: 1, completedAt: new Date().toISOString(), sources: await sourceStamp(), result }, 0o600);
|
|
return result;
|
|
}
|
|
|
|
async archiveLegacyRecord(context: SessionWorkspaceContext, name: ChatRecordName, source: string) {
|
|
return this.serialize(context, async () => {
|
|
if (!await this.validate(context, name)) throw new SessionContextMismatchError();
|
|
if ((await lstat(source)).isSymbolicLink()) throw new Error("legacy chat record must not be a symbolic link");
|
|
const content = await readFile(source);
|
|
const digest = createHash("sha256").update(content).digest("hex");
|
|
const archives = new SessionWorkspaceStore(this.backupRoot);
|
|
const directory = await archives.ensure(context);
|
|
const path = join(directory, "chat", `${name.replace(".json", "")}-${digest}.json`);
|
|
if (await exists(path) && (await lstat(path)).isSymbolicLink()) throw new Error("legacy backup must not be a symbolic link");
|
|
await createFileExclusively(path, (temporary) => writeFile(temporary, content, { flag: "wx", mode: 0o600 }));
|
|
if (!(await readFile(path)).equals(content) || !(await readFile(source)).equals(content)) throw new Error("legacy backup verification failed");
|
|
await rm(source);
|
|
});
|
|
}
|
|
|
|
private async copyLegacyRecord(context: SessionWorkspaceContext, name: ChatRecordName, source: string) {
|
|
return this.serialize(context, async () => {
|
|
await this.ensure(context);
|
|
await this.validate(context, name);
|
|
if (await exists(this.path(context, name))) return false;
|
|
if ((await lstat(source)).isSymbolicLink()) throw new Error("legacy chat record must not be a symbolic link");
|
|
return createFileExclusively(this.path(context, name), async (temporary) => {
|
|
await copyFile(source, temporary, constants.COPYFILE_EXCL);
|
|
await chmod(temporary, 0o600);
|
|
});
|
|
});
|
|
}
|
|
}
|