1import { ThreadId } from "@t3tools/contracts";
2import * as Context from "effect/Context";
3import * as DateTime from "effect/DateTime";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schema from "effect/Schema";
7import * as Stream from "effect/Stream";
8import * as SqlClient from "effect/sql/SqlClient";
9
10import * as EventStore from "../EventStore.ts";
11import * as ProjectionStore from "../ProjectionStore.ts";
12
13export interface ProjectionReplayVerification {
14 readonly valid: boolean;
15 readonly schemaVersion: number;
16 readonly expectedSequence: number;
17 readonly projectionSequence: number;
18 readonly unreadableThreadIds: ReadonlyArray<ThreadId>;
19 readonly missingThreadIds: ReadonlyArray<ThreadId>;
20 readonly unexpectedThreadIds: ReadonlyArray<ThreadId>;
21}
22
23export class ProjectionReplayError extends Schema.TaggedError<ProjectionReplayError>()(
24 "ProjectionReplayError",
25 { cause: Schema.Defect() },
26) {}
27
28/**
29 * Test oracle: projections must equal a replay of an unpruned event log. Production prunes
30 * events the projections cover, so it never replays (see ProjectionMaintenance).
31 */
32export class ProjectionReplay extends Context.Service<
33 ProjectionReplay,
34 {
35 readonly verify: Effect.Effect<ProjectionReplayVerification, ProjectionReplayError>;
36 readonly rebuild: Effect.Effect<ProjectionReplayVerification, ProjectionReplayError>;
37 }
38>()("t3/orchestration-v2/testkit/ProjectionReplay") {}
39
40type ProjectionMetadataRow = {
41 readonly schema_version: number;
42 readonly last_sequence: number;
43};
44
45export const layer: Layer.Layer<
46 ProjectionReplay,
47 never,
48 EventStore.EventStoreV2 | ProjectionStore.ProjectionStoreV2 | SqlClient.SqlClient
49> = Layer.effect(
50 ProjectionReplay,
51 Effect.gen(function* () {
52 const sql = yield* SqlClient.SqlClient;
53 const eventStore = yield* EventStore.EventStoreV2;
54 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
55
56 const verify = Effect.gen(function* () {
57 const expectedThreadRows = yield* sql<{ readonly thread_id: string }>`
58 SELECT DISTINCT stream_id AS thread_id
59 FROM orchestration_events INDEXED BY orchestration_events_v2_created_threads_idx
60 WHERE application_event_version = 2
61 AND aggregate_kind = 'thread'
62 AND event_type = 'thread.created'
63 ORDER BY stream_id ASC
64 `;
65 const projectionRows = yield* sql<{ readonly thread_id: string }>`
66 SELECT thread_id
67 FROM orchestration_v2_projection_threads
68 ORDER BY thread_id ASC
69 `;
70 const actualIds = projectionRows.map((row) => ThreadId.make(row.thread_id));
71 const expectedIds = expectedThreadRows.map((row) => ThreadId.make(row.thread_id));
72 const actualSet = new Set(actualIds);
73 const expectedSet = new Set(expectedIds);
74 const missingThreadIds = expectedIds.filter((threadId) => !actualSet.has(threadId));
75 const unexpectedThreadIds = actualIds.filter((threadId) => !expectedSet.has(threadId));
76 const unreadableThreadIds = yield* projectionStore.getUnreadableThreadIds();
77 const metadata = yield* sql<ProjectionMetadataRow>`
78 SELECT schema_version, last_sequence
79 FROM orchestration_v2_projection_metadata
80 WHERE projection_name = 'thread-projections'
81 LIMIT 1
82 `;
83 const expectedSequence = yield* eventStore.latestSequence();
84 const schemaVersion = metadata[0]?.schema_version ?? 0;
85 const projectionSequence = metadata[0]?.last_sequence ?? 0;
86 return {
87 valid:
88 schemaVersion === ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION &&
89 projectionSequence === expectedSequence &&
90 missingThreadIds.length === 0 &&
91 unexpectedThreadIds.length === 0 &&
92 unreadableThreadIds.length === 0,
93 schemaVersion,
94 expectedSequence,
95 projectionSequence,
96 unreadableThreadIds,
97 missingThreadIds,
98 unexpectedThreadIds,
99 } satisfies ProjectionReplayVerification;
100 }).pipe(sql.withTransaction);
101
102 const rebuild = Effect.gen(function* () {
103 yield* sql.withTransaction(
104 Effect.gen(function* () {
105 const throughSequence = yield* eventStore.latestSequence();
106 yield* sql`DELETE FROM orchestration_v2_projection_context_transfers`;
107 yield* sql`DELETE FROM orchestration_v2_projection_context_handoffs`;
108 yield* sql`DELETE FROM orchestration_v2_projection_checkpoints`;
109 yield* sql`DELETE FROM orchestration_v2_projection_checkpoint_scopes`;
110 yield* sql`DELETE FROM orchestration_v2_projection_turn_items`;
111 yield* sql`DELETE FROM orchestration_v2_projection_plans`;
112 yield* sql`DELETE FROM orchestration_v2_projection_messages`;
113 yield* sql`DELETE FROM orchestration_v2_projection_runtime_requests`;
114 yield* sql`DELETE FROM orchestration_v2_projection_provider_turns`;
115 yield* sql`DELETE FROM orchestration_v2_projection_provider_threads`;
116 yield* sql`DELETE FROM orchestration_v2_projection_provider_session_bindings`;
117 yield* sql`DELETE FROM orchestration_v2_projection_provider_sessions`;
118 yield* sql`DELETE FROM orchestration_v2_projection_subagents`;
119 yield* sql`DELETE FROM orchestration_v2_projection_nodes`;
120 yield* sql`DELETE FROM orchestration_v2_projection_run_attempts`;
121 yield* sql`DELETE FROM orchestration_v2_projection_runs`;
122 yield* sql`DELETE FROM orchestration_v2_projection_threads`;
123 yield* sql`DELETE FROM orchestration_v2_turn_item_positions`;
124
125 const pageSize = 500;
126 let lastSequence = 0;
127 while (true) {
128 const page = yield* eventStore
129 .read({ afterSequence: lastSequence, throughSequence, limit: pageSize })
130 .pipe(Stream.runCollect);
131 for (const stored of page) {
132 yield* projectionStore.apply(stored.event);
133 if (stored.event.type === "turn-item.updated") {
134 yield* sql`
135 INSERT INTO orchestration_v2_turn_item_positions (
136 thread_id,
137 turn_item_id,
138 ordinal
139 )
140 VALUES (
141 ${stored.event.threadId},
142 ${stored.event.payload.id},
143 ${stored.event.payload.ordinal}
144 )
145 ON CONFLICT(thread_id, turn_item_id) DO UPDATE SET
146 ordinal = excluded.ordinal
147 `;
148 }
149 lastSequence = stored.sequence;
150 }
151 if (page.length < pageSize) break;
152 yield* Effect.yieldNow;
153 }
154 const now = DateTime.formatIso(yield* DateTime.now);
155 yield* sql`
156 INSERT INTO orchestration_v2_projection_metadata (
157 projection_name,
158 schema_version,
159 last_sequence,
160 updated_at
161 )
162 VALUES (
163 'thread-projections',
164 ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION},
165 ${lastSequence},
166 ${now}
167 )
168 ON CONFLICT(projection_name) DO UPDATE SET
169 schema_version = excluded.schema_version,
170 last_sequence = excluded.last_sequence,
171 updated_at = excluded.updated_at
172 `;
173 }),
174 );
175 return yield* verify;
176 });
177
178 const mapError = <A, E>(effect: Effect.Effect<A, E>) =>
179 effect.pipe(Effect.mapError((cause) => new ProjectionReplayError({ cause })));
180 return ProjectionReplay.of({ verify: mapError(verify), rebuild: mapError(rebuild) });
181 }),
182);