1import {
2 ArchiveCatalogEntry,
3 type AssetResource,
4 ArchiveError,
5 ArchivePublication,
6 EnvironmentId,
7 ThreadId,
8 type OrchestrationV2ThreadDetailSnapshot,
9} from "@t3tools/contracts";
10// Streaming verification keeps archive media memory bounded.
11// @effect-diagnostics nodeBuiltinImport:off
12import * as NodeCrypto from "node:crypto";
13import * as NodePath from "@effect/platform-node/NodePath";
14import { isWindowsAbsolutePath } from "@t3tools/shared/path";
15import { hostPreviewMimeTypeFromExtension } from "@t3tools/shared/filePreview";
16import { parseAttachmentIdFromRelativePath } from "../attachmentStore.ts";
17import * as HostProcess from "@t3tools/shared/HostProcess";
18import * as Context from "effect/Context";
19import * as Effect from "effect/Effect";
20import * as FileSystem from "effect/FileSystem";
21import * as Layer from "effect/Layer";
22import * as Option from "effect/Option";
23import * as Path from "effect/Path";
24import * as Redacted from "effect/Redacted";
25import * as Schema from "effect/Schema";
26import * as Semaphore from "effect/Semaphore";
27import * as Stream from "effect/Stream";
28import * as SqlClient from "effect/sql/SqlClient";
29import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
30import * as SqlitePersistence from "../persistence/Sqlite.ts";
31import * as ArchiveObjects from "./ArchiveObjects.ts";
32import { Bundle, Row } from "./ArchiveFormat.ts";
33
34const decodeBundle = Schema.decodeEffect(Schema.fromJsonString(Bundle));
35const decodeRows = Schema.decodeEffect(Schema.fromJsonString(Schema.Array(Row)));
36const decodeEntry = Schema.decodeEffect(
37 Schema.fromJsonString(Schema.toCodecJson(ArchiveCatalogEntry)),
38);
39const encodeEntry = Schema.encodeEffect(
40 Schema.fromJsonString(Schema.toCodecJson(ArchiveCatalogEntry)),
41);
42const isArchiveError = Schema.is(ArchiveError);
43const storageError = (cause: unknown) =>
44 isArchiveError(cause)
45 ? cause
46 : new ArchiveError({ reason: "storage", cause: Redacted.make(cause) });
47
48export class ArchiveCatalog extends Context.Service<
49 ArchiveCatalog,
50 {
51 readonly publish: (
52 input: ArchivePublication,
53 ) => Effect.Effect<ArchiveCatalogEntry, ArchiveError>;
54 readonly list: Effect.Effect<
55 { readonly entries: ReadonlyArray<ArchiveCatalogEntry> },
56 ArchiveError
57 >;
58 readonly read: (
59 environmentId: EnvironmentId,
60 threadId: ThreadId,
61 ) => Effect.Effect<OrchestrationV2ThreadDetailSnapshot, ArchiveError>;
62 readonly asset: (
63 environmentId: EnvironmentId,
64 resource: AssetResource,
65 ) => Effect.Effect<
66 {
67 readonly threadId: ThreadId;
68 readonly manifestHash: string;
69 readonly hash: string;
70 readonly fileName: string;
71 readonly mimeType: string;
72 readonly download: boolean;
73 readonly renderIntent?: "mcp-app";
74 },
75 ArchiveError
76 >;
77 readonly openAsset: (
78 environmentId: EnvironmentId,
79 threadId: ThreadId,
80 manifestHash: string,
81 hash: string,
82 ) => Effect.Effect<string, ArchiveError>;
83 readonly withdraw: (
84 environmentId: EnvironmentId,
85 threadId: ThreadId,
86 manifestHash: string,
87 ) => Effect.Effect<void, ArchiveError>;
88 }
89>()("t3/archive/ArchiveCatalog") {}
90
91export const layer = Layer.effect(
92 ArchiveCatalog,
93 Effect.gen(function* () {
94 const objects = yield* ArchiveObjects.ArchiveObjects;
95 const fs = yield* FileSystem.FileSystem;
96 const path = yield* Path.Path;
97 const windowsPath = yield* Path.Path.pipe(Effect.provide(NodePath.layerWin32));
98 const lock = yield* Semaphore.make(1);
99 const directory = Effect.gen(function* () {
100 if (Option.isNone(objects.root)) return yield* new ArchiveError({ reason: "unavailable" });
101 // Object lookup verifies the configured archive storage before index access.
102 yield* objects.directory;
103 const directory = path.join(objects.root.value, "catalog");
104 yield* fs.makeDirectory(directory, { recursive: true, mode: 0o700 });
105 return directory;
106 });
107 const entryPath = (directory: string, environmentId: EnvironmentId, threadId: ThreadId) =>
108 path.join(
109 directory,
110 NodeCrypto.createHash("sha256")
111 .update(JSON.stringify([environmentId, threadId]))
112 .digest("hex") + ".json",
113 );
114 const verifiedFile = Effect.fn("ArchiveCatalog.verifiedFile")(function* (hash: string) {
115 const file = yield* objects.file(hash);
116 const digest = NodeCrypto.createHash("sha256");
117 yield* fs.stream(file).pipe(
118 Stream.runForEach((bytes) =>
119 Effect.sync(() => {
120 digest.update(bytes);
121 }),
122 ),
123 );
124 if (digest.digest("hex") !== hash) return yield* new ArchiveError({ reason: "integrity" });
125 return file;
126 });
127 const readManifest = (hash: string) =>
128 verifiedFile(hash).pipe(Effect.flatMap(fs.readFileString), Effect.flatMap(decodeBundle));
129 const load = Effect.fn("ArchiveCatalog.load")(function* (hash: string) {
130 const manifest = yield* readManifest(hash);
131 const hashes = new Set([
132 manifest.events,
133 ...manifest.segments.map((segment) => segment.hash),
134 ...manifest.context.map((segment) => segment.hash),
135 ...manifest.assets.map((asset) => asset.hash),
136 ]);
137 for (const hash of hashes) yield* verifiedFile(hash);
138 return manifest;
139 });
140 const get = Effect.fn("ArchiveCatalog.get")(function* (
141 environmentId: EnvironmentId,
142 threadId: ThreadId,
143 ) {
144 const file = entryPath(yield* directory, environmentId, threadId);
145 if (!(yield* fs.exists(file))) return yield* new ArchiveError({ reason: "unavailable" });
146 const entry = yield* fs.readFileString(file).pipe(Effect.flatMap(decodeEntry));
147 if (entry.environmentId !== environmentId || entry.thread.id !== threadId)
148 return yield* new ArchiveError({ reason: "integrity" });
149 return entry;
150 });
151 const materialize = (manifest: typeof Bundle.Type) =>
152 Effect.gen(function* () {
153 const segments = yield* Effect.forEach(
154 [
155 ...manifest.context,
156 ...manifest.segments.map((segment) => ({
157 ...segment,
158 table: `orchestration_v2_projection_${segment.table}`,
159 })),
160 ],
161 (segment) =>
162 Effect.gen(function* () {
163 const rows = yield* fs
164 .readFileString(yield* objects.file(segment.hash))
165 .pipe(Effect.flatMap(decodeRows));
166 return { table: segment.table, rows };
167 }),
168 );
169 const events = yield* fs
170 .readFileString(yield* objects.file(manifest.events))
171 .pipe(Effect.flatMap(decodeRows));
172 return yield* Effect.gen(function* () {
173 const sql = yield* SqlClient.SqlClient;
174 for (const segment of segments)
175 for (const row of segment.rows)
176 yield* sql`INSERT OR REPLACE INTO ${sql(segment.table)} ${sql.insert(row)}`;
177 for (const row of events)
178 yield* sql`INSERT OR REPLACE INTO orchestration_events ${sql.insert(row)}`;
179 return yield* (yield* ProjectionStore.ProjectionStoreV2).getThreadSnapshot(
180 manifest.threadId,
181 );
182 }).pipe(
183 Effect.provide(
184 Layer.fresh(
185 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
186 ),
187 ),
188 );
189 });
190 const list = Effect.gen(function* () {
191 if (Option.isNone(objects.root)) return { entries: [] };
192 const dir = yield* directory;
193 const names = (yield* fs.readDirectory(dir)).filter((name) =>
194 /^[a-f0-9]{64}\.json$/.test(name),
195 );
196 const entries = yield* Effect.forEach(
197 names,
198 (name) => fs.readFileString(path.join(dir, name)).pipe(Effect.flatMap(decodeEntry)),
199 { concurrency: 4 },
200 );
201 return {
202 entries: entries.sort(
203 (a, b) => b.thread.updatedAt.epochMilliseconds - a.thread.updatedAt.epochMilliseconds,
204 ),
205 };
206 }).pipe(Effect.mapError(storageError));
207 return ArchiveCatalog.of({
208 publish: (input) =>
209 lock
210 .withPermit(
211 Effect.gen(function* () {
212 yield* objects.verifyDurableStore;
213 const manifest = yield* load(input.manifestHash);
214 const snapshot = yield* materialize(manifest);
215 const thread = ProjectionStore.threadShellFromProjection(snapshot.projection);
216 if (
217 thread.id !== manifest.threadId ||
218 !thread.archivedAt ||
219 thread.deletedAt ||
220 thread.projectId !== input.project.id
221 )
222 return yield* new ArchiveError({ reason: "integrity" });
223 const entry = ArchiveCatalogEntry.make({
224 environmentId: input.environmentId,
225 manifestHash: input.manifestHash,
226 throughSequence: manifest.throughSequence,
227 thread,
228 project: input.project,
229 });
230 const dir = yield* directory;
231 const file = entryPath(dir, input.environmentId, thread.id);
232 if (yield* fs.exists(file)) {
233 const previous = yield* fs.readFileString(file).pipe(Effect.flatMap(decodeEntry));
234 if (
235 previous.throughSequence > entry.throughSequence ||
236 (previous.throughSequence === entry.throughSequence &&
237 previous.manifestHash !== entry.manifestHash)
238 )
239 return yield* new ArchiveError({ reason: "changed" });
240 }
241 yield* Effect.scoped(
242 Effect.gen(function* () {
243 const temporary = yield* fs.makeTempFileScoped({
244 directory: dir,
245 prefix: ".publish-",
246 });
247 yield* fs.writeFileString(temporary, yield* encodeEntry(entry), { mode: 0o600 });
248 yield* fs
249 .open(temporary, { flag: "r+" })
250 .pipe(Effect.flatMap((file) => file.sync));
251 yield* fs.rename(temporary, file);
252 if (!(yield* HostProcess.isWindows))
253 yield* fs.open(dir).pipe(Effect.flatMap((file) => file.sync));
254 }),
255 );
256 return entry;
257 }),
258 )
259 .pipe(Effect.mapError(storageError)),
260 list,
261 read: (environmentId, threadId) =>
262 Effect.gen(function* () {
263 const entry = yield* get(environmentId, threadId);
264 const manifest = yield* load(entry.manifestHash);
265 if (manifest.threadId !== threadId || manifest.throughSequence !== entry.throughSequence)
266 return yield* new ArchiveError({ reason: "integrity" });
267 return yield* materialize(manifest);
268 }).pipe(Effect.mapError(storageError)),
269 asset: (environmentId, resource) =>
270 Effect.gen(function* () {
271 if (
272 resource._tag !== "attachment" &&
273 resource._tag !== "media-file" &&
274 resource._tag !== "workspace-file"
275 )
276 return yield* new ArchiveError({ reason: "unavailable" });
277 const entries = (yield* list).entries.filter(
278 (entry) =>
279 entry.environmentId === environmentId &&
280 (resource._tag === "attachment" || entry.thread.id === resource.threadId),
281 );
282 for (const entry of entries) {
283 const manifest = yield* readManifest(entry.manifestHash);
284 const sourcePath = isWindowsAbsolutePath(entry.project.workspaceRoot)
285 ? windowsPath
286 : path;
287 const requested =
288 resource._tag === "attachment"
289 ? undefined
290 : sourcePath.resolve(
291 entry.thread.worktreePath ?? entry.project.workspaceRoot,
292 resource.path,
293 );
294 const found = manifest.assets.find((asset) =>
295 resource._tag === "attachment"
296 ? parseAttachmentIdFromRelativePath(sourcePath.basename(asset.path)) ===
297 resource.attachmentId
298 : sourcePath.normalize(asset.path) === requested,
299 );
300 if (!found) continue;
301 yield* verifiedFile(found.hash);
302 const fileName =
303 resource._tag === "attachment"
304 ? (resource.fileName ?? sourcePath.basename(found.path))
305 : sourcePath.basename(found.path);
306 const mimeType =
307 hostPreviewMimeTypeFromExtension(sourcePath.extname(found.path)) ??
308 hostPreviewMimeTypeFromExtension(sourcePath.extname(fileName)) ??
309 "application/octet-stream";
310 return {
311 threadId: entry.thread.id,
312 manifestHash: entry.manifestHash,
313 hash: found.hash,
314 fileName,
315 mimeType,
316 download:
317 resource._tag === "attachment" &&
318 resource.disposition !== "inline" &&
319 !/^(image|video)\//u.test(mimeType),
320 ...(resource._tag === "attachment" && resource.renderIntent === "mcp-app"
321 ? { renderIntent: "mcp-app" as const }
322 : {}),
323 };
324 }
325 return yield* new ArchiveError({ reason: "unavailable" });
326 }).pipe(Effect.mapError(storageError)),
327 openAsset: (environmentId, threadId, manifestHash, hash) =>
328 Effect.gen(function* () {
329 const entry = yield* get(environmentId, threadId);
330 if (entry.manifestHash !== manifestHash)
331 return yield* new ArchiveError({ reason: "changed" });
332 const manifest = yield* readManifest(manifestHash);
333 if (!manifest.assets.some((asset) => asset.hash === hash))
334 return yield* new ArchiveError({ reason: "unavailable" });
335 return yield* objects.file(hash);
336 }).pipe(Effect.mapError(storageError)),
337 withdraw: (environmentId, threadId, manifestHash) =>
338 Effect.scoped(
339 lock
340 .withPermit(
341 Effect.gen(function* () {
342 const dir = yield* directory;
343 const file = entryPath(dir, environmentId, threadId);
344 if (!(yield* fs.exists(file))) return;
345 const current = yield* fs.readFileString(file).pipe(Effect.flatMap(decodeEntry));
346 if (current.manifestHash !== manifestHash) return;
347 yield* fs.remove(entryPath(dir, environmentId, threadId), { force: true });
348 if (!(yield* HostProcess.isWindows))
349 yield* fs.open(dir).pipe(Effect.flatMap((file) => file.sync));
350 }),
351 )
352 .pipe(Effect.mapError(storageError)),
353 ),
354 });
355 }),
356);