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 TranscriptArchive from "../archive/TranscriptArchive.ts";
12import * as ProjectionStore from "./ProjectionStore.ts";
13
14export interface ProjectionVerificationV2 {
15 readonly valid: boolean;
16 readonly schemaVersion: number;
17 readonly expectedSequence: number;
18 readonly projectionSequence: number;
19 readonly unreadableThreadIds: ReadonlyArray<ThreadId>;
20 readonly missingThreadIds: ReadonlyArray<ThreadId>;
21 readonly unexpectedThreadIds: ReadonlyArray<ThreadId>;
22}
23
24export class ProjectionMaintenanceError extends Schema.TaggedError<ProjectionMaintenanceError>()(
25 "ProjectionMaintenanceError",
26 {
27 operation: Schema.String,
28 cause: Schema.optional(Schema.Defect()),
29 },
30) {}
31
32export interface ProjectionMaintenanceV2Shape {
33 readonly verify: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;
34 readonly rebuild: Effect.Effect<ProjectionVerificationV2, ProjectionMaintenanceError>;
35 readonly compactEventStore: Effect.Effect<
36 {
37 readonly deletedEventCount: number;
38 readonly deletedReceiptCount: number;
39 readonly reclaimableBytes: number;
40 },
41 ProjectionMaintenanceError
42 >;
43}
44
45export class ProjectionMaintenanceV2 extends Context.Service<
46 ProjectionMaintenanceV2,
47 ProjectionMaintenanceV2Shape
48>()("t3/orchestration-v2/ProjectionMaintenance/ProjectionMaintenanceV2") {}
49
50type ProjectionMetadataRow = {
51 readonly schema_version: number;
52 readonly last_sequence: number;
53};
54
55const encodeEntityKey = Schema.encodeSync(
56 Schema.fromJsonString(Schema.Tuple([Schema.String, Schema.String, Schema.NullOr(Schema.String)])),
57);
58
59export const layer: Layer.Layer<
60 ProjectionMaintenanceV2,
61 never,
62 EventStore.EventStoreV2 | ProjectionStore.ProjectionStoreV2 | SqlClient.SqlClient
63> = Layer.effect(
64 ProjectionMaintenanceV2,
65 Effect.gen(function* () {
66 const sql = yield* SqlClient.SqlClient;
67 const eventStore = yield* EventStore.EventStoreV2;
68 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
69
70 /**
71 * EventSink commits the event, its projection updates, and projection metadata in one SQL
72 * transaction. Startup verification therefore checks that transaction boundary and that every
73 * stored projection can be decoded. It intentionally does not replay domain events through a
74 * second projector: doing so creates another implementation of projection semantics that must
75 * evolve in lockstep with ProjectionStore.
76 */
77 const verify = Effect.gen(function* () {
78 const expectedThreadRows = yield* sql<{ readonly thread_id: string }>`
79 SELECT DISTINCT stream_id AS thread_id
80 FROM orchestration_events INDEXED BY orchestration_events_v2_created_threads_idx
81 WHERE application_event_version = 2
82 AND aggregate_kind = 'thread'
83 AND event_type = 'thread.created'
84 ORDER BY stream_id ASC
85 `;
86 const projectionRows = yield* sql<{ readonly thread_id: string }>`
87 SELECT thread_id
88 FROM orchestration_v2_projection_threads
89 ORDER BY thread_id ASC
90 `;
91 const actualIds = projectionRows.map((row) => ThreadId.make(row.thread_id));
92 const expectedIds = expectedThreadRows.map((row) => ThreadId.make(row.thread_id));
93 const actualSet = new Set(actualIds);
94 const expectedSet = new Set(expectedIds);
95 const missingThreadIds = expectedIds.filter((threadId) => !actualSet.has(threadId));
96 const unexpectedThreadIds = actualIds.filter((threadId) => !expectedSet.has(threadId));
97 const unreadableThreadIds = yield* projectionStore.getUnreadableThreadIds();
98 const metadata = yield* sql<ProjectionMetadataRow>`
99 SELECT schema_version, last_sequence
100 FROM orchestration_v2_projection_metadata
101 WHERE projection_name = 'thread-projections'
102 LIMIT 1
103 `;
104 const expectedSequence = yield* eventStore.latestSequence();
105 const schemaVersion = metadata[0]?.schema_version ?? 0;
106 const projectionSequence = metadata[0]?.last_sequence ?? 0;
107 return {
108 valid:
109 schemaVersion === ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION &&
110 projectionSequence === expectedSequence &&
111 missingThreadIds.length === 0 &&
112 unexpectedThreadIds.length === 0 &&
113 unreadableThreadIds.length === 0,
114 schemaVersion,
115 expectedSequence,
116 projectionSequence,
117 unreadableThreadIds,
118 missingThreadIds,
119 unexpectedThreadIds,
120 } satisfies ProjectionVerificationV2;
121 }).pipe(sql.withTransaction);
122
123 const rebuild = Effect.gen(function* () {
124 yield* sql.withTransaction(
125 Effect.gen(function* () {
126 const throughSequence = yield* eventStore.latestSequence();
127 yield* sql`DELETE FROM orchestration_v2_projection_context_transfers`;
128 yield* sql`DELETE FROM orchestration_v2_projection_context_handoffs`;
129 yield* sql`DELETE FROM orchestration_v2_projection_checkpoints`;
130 yield* sql`DELETE FROM orchestration_v2_projection_checkpoint_scopes`;
131 yield* sql`DELETE FROM orchestration_v2_projection_turn_items`;
132 yield* sql`DELETE FROM orchestration_v2_projection_plans`;
133 yield* sql`DELETE FROM orchestration_v2_projection_messages`;
134 yield* sql`DELETE FROM orchestration_v2_projection_runtime_requests`;
135 yield* sql`DELETE FROM orchestration_v2_projection_provider_turns`;
136 yield* sql`DELETE FROM orchestration_v2_projection_provider_threads`;
137 yield* sql`DELETE FROM orchestration_v2_projection_provider_session_bindings`;
138 yield* sql`DELETE FROM orchestration_v2_projection_provider_sessions`;
139 yield* sql`DELETE FROM orchestration_v2_projection_subagents`;
140 yield* sql`DELETE FROM orchestration_v2_projection_nodes`;
141 yield* sql`DELETE FROM orchestration_v2_projection_run_attempts`;
142 yield* sql`DELETE FROM orchestration_v2_projection_runs`;
143 yield* sql`DELETE FROM orchestration_v2_projection_threads`;
144 yield* sql`DELETE FROM orchestration_v2_turn_item_positions`;
145
146 const pageSize = 500;
147 let lastSequence = 0;
148 while (true) {
149 const page = yield* eventStore
150 .read({ afterSequence: lastSequence, throughSequence, limit: pageSize })
151 .pipe(Stream.runCollect);
152 for (const stored of page) {
153 yield* projectionStore.apply(stored.event);
154 if (stored.event.type === "turn-item.updated") {
155 yield* sql`
156 INSERT INTO orchestration_v2_turn_item_positions (
157 thread_id,
158 turn_item_id,
159 ordinal
160 )
161 VALUES (
162 ${stored.event.threadId},
163 ${stored.event.payload.id},
164 ${stored.event.payload.ordinal}
165 )
166 ON CONFLICT(thread_id, turn_item_id) DO UPDATE SET
167 ordinal = excluded.ordinal
168 `;
169 }
170 lastSequence = stored.sequence;
171 }
172 if (page.length < pageSize) break;
173 yield* Effect.yieldNow;
174 }
175 yield* TranscriptArchive.rebuildMetadata.pipe(
176 Effect.provideService(SqlClient.SqlClient, sql),
177 );
178 yield* sql`INSERT INTO orchestration_v2_turn_item_positions(thread_id, turn_item_id, ordinal)
179 SELECT item.thread_id, item.turn_item_id, item.ordinal FROM orchestration_v2_projection_turn_items item
180 WHERE EXISTS (SELECT 1 FROM transcript_archives a WHERE a.thread_id = item.thread_id)
181 ON CONFLICT(thread_id, turn_item_id) DO UPDATE SET ordinal = excluded.ordinal`;
182 const now = DateTime.formatIso(yield* DateTime.now);
183 yield* sql`
184 INSERT INTO orchestration_v2_projection_metadata (
185 projection_name,
186 schema_version,
187 last_sequence,
188 updated_at
189 )
190 VALUES (
191 'thread-projections',
192 ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION},
193 ${throughSequence},
194 ${now}
195 )
196 ON CONFLICT(projection_name) DO UPDATE SET
197 schema_version = excluded.schema_version,
198 last_sequence = excluded.last_sequence,
199 updated_at = excluded.updated_at
200 `;
201 }),
202 );
203 return yield* verify;
204 });
205
206 const mapError =
207 (operation: string) =>
208 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
209 effect.pipe(
210 Effect.mapError((cause) => new ProjectionMaintenanceError({ operation, cause })),
211 );
212
213 // Thread-state events whose payload is the complete thread: only the
214 // newest per thread can influence a replay or an afterSequence catch-up.
215 // thread.created stays out — verify derives the expected thread set from
216 // it, and it anchors replay ordering.
217 const SUPERSEDABLE_THREAD_EVENT_TYPES = [
218 "thread.archived",
219 "thread.unarchived",
220 "thread.deleted",
221 "thread.settled",
222 "thread.unsettled",
223 "thread.snoozed",
224 "thread.unsnoozed",
225 "thread.pinned",
226 "thread.auto-settle-set",
227 "thread.unpinned",
228 "thread.pin-reordered",
229 "thread.active-reordered",
230 "thread.metadata-updated",
231 "thread.pull-request-synced",
232 "thread.runtime-mode-updated",
233 "thread.interaction-mode-updated",
234 "thread.model-selection-updated",
235 "thread.provider-switched",
236 "thread.visited",
237 "thread.marked-unread",
238 ];
239
240 const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES);
241 // Keep subagent.updated and turn-item.updated rows: slimmed payloads reference their prompts.
242 const COMPACTION_PAGE_SIZE = 500;
243
244 /**
245 * Scan newest first to retain the newest state for each entity.
246 * Page every event, including non-candidates: filtering before LIMIT could
247 * still scan the entire history when superseded events are sparse.
248 * turn-item.updated stays intact because replay assigns positions on first write.
249 */
250 const compactEventStore = Effect.gen(function* () {
251 const bounds = yield* sql<{
252 readonly event_sequence: number;
253 readonly receipt_row_id: number;
254 }>`
255 SELECT
256 COALESCE((SELECT MAX(sequence) FROM orchestration_events), 0) AS event_sequence,
257 COALESCE((SELECT MAX(rowid) FROM orchestration_command_receipts), 0) AS receipt_row_id
258 `;
259 let throughSequence = bounds[0]?.event_sequence ?? 0;
260 let throughReceiptRowId = bounds[0]?.receipt_row_id ?? 0;
261 const retainedThreadIds = new Set<string>();
262 const retainedEntityKeys = new Set<string>();
263 let deletedEventCount = 0;
264 let deletedReceiptCount = 0;
265
266 while (throughSequence > 0) {
267 const rows = yield* sql<{
268 readonly sequence: number;
269 readonly application_event_version: number;
270 readonly aggregate_kind: string;
271 readonly stream_id: string;
272 readonly event_type: string;
273 readonly entity_id: string | null;
274 readonly imported_legacy_thread: number;
275 }>`
276 SELECT
277 event.sequence,
278 event.application_event_version,
279 event.aggregate_kind,
280 event.stream_id,
281 event.event_type,
282 CASE
283 WHEN event.application_event_version = 2
284 AND event.event_type IN ('message.updated', 'node.updated')
285 THEN json_extract(event.payload_json, '$.id')
286 ELSE NULL
287 END AS entity_id,
288 CASE
289 WHEN event.application_event_version = 1 AND event.aggregate_kind = 'thread'
290 THEN EXISTS (
291 SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import
292 WHERE legacy_import.thread_id = event.stream_id
293 AND legacy_import.transcript_imported_at IS NOT NULL
294 )
295 ELSE 0
296 END AS imported_legacy_thread
297 FROM orchestration_events AS event
298 WHERE event.sequence <= ${throughSequence}
299 ORDER BY event.sequence DESC
300 LIMIT ${COMPACTION_PAGE_SIZE}
301 `;
302 const obsolete: number[] = [];
303 for (const row of rows) {
304 if (row.imported_legacy_thread === 1) {
305 obsolete.push(row.sequence);
306 } else if (row.application_event_version === 2) {
307 if (
308 row.aggregate_kind === "thread" &&
309 supersedableThreadEventTypes.has(row.event_type)
310 ) {
311 if (retainedThreadIds.has(row.stream_id)) obsolete.push(row.sequence);
312 else retainedThreadIds.add(row.stream_id);
313 } else if (
314 row.entity_id !== null &&
315 (row.event_type === "message.updated" || row.event_type === "node.updated")
316 ) {
317 const key = encodeEntityKey([row.event_type, row.stream_id, row.entity_id]);
318 if (retainedEntityKeys.has(key)) obsolete.push(row.sequence);
319 else retainedEntityKeys.add(key);
320 }
321 }
322 }
323 // Both discovery and deletion use the synchronous connection. Yield even
324 // when this page has nothing to delete so startup and requests can progress.
325 yield* Effect.yieldNow;
326 if (obsolete.length > 0) {
327 yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`;
328 deletedEventCount += obsolete.length;
329 yield* Effect.yieldNow;
330 }
331 throughSequence = (rows.at(-1)?.sequence ?? 1) - 1;
332 if (rows.length < COMPACTION_PAGE_SIZE) break;
333 }
334
335 // Legacy receipts are removable only after their thread's v1 import finishes.
336 // Page by rowid before checking eligibility, as with event discovery above.
337 while (throughReceiptRowId > 0) {
338 const rows = yield* sql<{
339 readonly row_id: number;
340 readonly command_id: string;
341 readonly imported_legacy_thread: number;
342 }>`
343 SELECT
344 receipt.rowid AS row_id,
345 receipt.command_id,
346 CASE
347 WHEN receipt.command_type = 'legacy' AND receipt.aggregate_kind = 'thread'
348 THEN EXISTS (
349 SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import
350 WHERE legacy_import.thread_id = receipt.aggregate_id
351 AND legacy_import.transcript_imported_at IS NOT NULL
352 )
353 ELSE 0
354 END AS imported_legacy_thread
355 FROM orchestration_command_receipts AS receipt
356 WHERE receipt.rowid <= ${throughReceiptRowId}
357 ORDER BY receipt.rowid DESC
358 LIMIT ${COMPACTION_PAGE_SIZE}
359 `;
360 const obsolete = rows
361 .filter((row) => row.imported_legacy_thread === 1)
362 .map((row) => row.command_id);
363 yield* Effect.yieldNow;
364 if (obsolete.length > 0) {
365 yield* sql`DELETE FROM orchestration_command_receipts WHERE command_id IN ${sql.in(obsolete)}`;
366 deletedReceiptCount += obsolete.length;
367 yield* Effect.yieldNow;
368 }
369 throughReceiptRowId = (rows.at(-1)?.row_id ?? 1) - 1;
370 if (rows.length < COMPACTION_PAGE_SIZE) break;
371 }
372
373 const freelistRows = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`;
374 const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`;
375 const reclaimableBytes =
376 (freelistRows[0]?.freelist_count ?? 0) * (pageSizeRows[0]?.page_size ?? 0);
377
378 return { deletedEventCount, deletedReceiptCount, reclaimableBytes };
379 });
380
381 return ProjectionMaintenanceV2.of({
382 verify: mapError("verify")(verify),
383 rebuild: mapError("rebuild")(rebuild),
384 compactEventStore: mapError("compact event store")(compactEventStore),
385 });
386 }),
387);