1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
5import * as TranscriptArchive from "./TranscriptArchive.ts";
6import type { ThreadId } from "@t3tools/contracts";
7
8export const layer = Layer.effect(
9 ProjectionStore.ProjectionStoreV2,
10 Effect.gen(function* () {
11 const base = yield* ProjectionStore.ProjectionStoreV2;
12 const archive = yield* TranscriptArchive.TranscriptArchive;
13 const read = <A>(
14 id: ThreadId,
15 effect: Effect.Effect<A, ProjectionStore.ProjectionStoreV2Error>,
16 includeHistory = false,
17 ) =>
18 archive
19 .read(id, effect, includeHistory)
20 .pipe(
21 Effect.mapError(
22 (cause) => new ProjectionStore.ProjectionStoreReadError({ threadId: id, cause }),
23 ),
24 );
25 return ProjectionStore.ProjectionStoreV2.of({
26 ...base,
27 apply: (event) => {
28 const restore =
29 event.type === "thread.unarchived" ||
30 ["message.updated", "turn-item.updated", "subagent.updated"].includes(event.type);
31 return restore
32 ? archive.restore(event.threadId).pipe(
33 Effect.mapError(
34 (cause) =>
35 new ProjectionStore.ProjectionStoreApplyEventError({
36 eventType: event.type,
37 cause,
38 }),
39 ),
40 Effect.andThen(base.apply(event)),
41 )
42 : base.apply(event);
43 },
44 getThreadProjection: (id) => read(id, base.getThreadProjection(id), true),
45 getThreadRecords: (id, fields, filter) =>
46 fields.some((field) => ["messages", "turnItems", "subagents"].includes(field))
47 ? read(id, base.getThreadRecords(id, fields, filter))
48 : base.getThreadRecords(id, fields, filter),
49 getThreadSnapshot: (id) => read(id, base.getThreadSnapshot(id), true),
50 getThreadSnapshotWindow: (id, options) =>
51 read(id, base.getThreadSnapshotWindow(id, options), true),
52 getThreadHistoryPage: (id, cursor, through, conversation) =>
53 read(id, base.getThreadHistoryPage(id, cursor, through, conversation), true),
54 getTimelinePage: (id, options) => read(id, base.getTimelinePage(id, options), true),
55 getTurnItem: (input) => read(input.threadId, base.getTurnItem(input)),
56 getThreadAttachmentIds: (id) => read(id, base.getThreadAttachmentIds(id)),
57 getTurnStartContext: (id, run) => read(id, base.getTurnStartContext(id, run)),
58 getTurnStartHistory: (id, runs) => read(id, base.getTurnStartHistory(id, runs)),
59 getRuntimeRecoveryProjection: (id) => read(id, base.getRuntimeRecoveryProjection(id)),
60 getPendingNativeUserInputs: (id, turn) => read(id, base.getPendingNativeUserInputs(id, turn)),
61
62 getProviderControlContext: (id, target) =>
63 read(id, base.getProviderControlContext(id, target)),
64
65 getRuntimeResponseContext: (id, request) =>
66 read(id, base.getRuntimeResponseContext(id, request)),
67 getRunMessage: (id, run) => read(id, base.getRunMessage(id, run)),
68 searchThread: (input) => read(input.threadId, base.searchThread(input), true),
69 searchThreadStream: (input) =>
70 archive
71 .readStream(input.threadId, base.searchThreadStream(input))
72 .pipe(
73 Stream.mapError(
74 (cause) =>
75 new ProjectionStore.ProjectionStoreReadError({ threadId: input.threadId, cause }),
76 ),
77 ),
78 });
79 }),
80).pipe(Layer.provide(ProjectionStore.layer));