1import {
2 threadPullRequestKeysEqual,
3 threadPullRequestsOf,
4} from "@t3tools/shared/threadPullRequests";
5import {
6 ChatAttachment,
7 OrchestrationMessageContext,
8 DEFAULT_MODEL,
9 EventId,
10 MessageId,
11 ModelSelection,
12 type OrchestrationV2AppThread,
13 OrchestrationV2AppThreadJson,
14 type OrchestrationV2ConversationMessage,
15 type OrchestrationV2DomainEvent,
16 type OrchestrationV2TurnItem,
17 ProjectId,
18 ProviderInstanceId,
19 ThreadId,
20 ThreadLinkedPullRequest,
21 ThreadPullRequestLink,
22 TurnItemId,
23} from "@t3tools/contracts";
24import * as KeyedLock from "@t3tools/shared/KeyedLock";
25import * as Context from "effect/Context";
26import * as DateTime from "effect/DateTime";
27import * as Effect from "effect/Effect";
28import * as Layer from "effect/Layer";
29import * as Option from "effect/Option";
30import * as Schema from "effect/Schema";
31import * as SqlClient from "effect/sql/SqlClient";
32
33import * as EventSink from "../EventSink.ts";
34import { randomUuidV4 } from "@t3tools/provider-core/server/randomUuid";
35
36const IMPORT_EVENT_PREFIX = "migration:v1";
37const TRANSCRIPT_EVENT_BATCH_SIZE = 100;
38
39interface LegacyThreadRow {
40 readonly thread_id: string;
41 readonly project_id: string;
42 readonly title: string;
43 readonly model_selection_json: string | null;
44 readonly runtime_mode: string;
45 readonly interaction_mode: string;
46 readonly branch: string | null;
47 readonly worktree_path: string | null;
48 readonly created_at: string;
49 readonly updated_at: string;
50 readonly archived_at: string | null;
51 readonly settled_override: string | null;
52 readonly settled_at: string | null;
53 readonly unsettled_at: string | null;
54 readonly snoozed_until: string | null;
55 readonly snoozed_at: string | null;
56 readonly pinned_at: string | null;
57 readonly auto_settle_disabled_at: string | null;
58 readonly pin_order_key: string | null;
59 readonly pull_requests_json: string;
60 readonly linked_pull_request_json: string | null;
61 readonly branch_pull_request_json: string | null;
62 readonly active_order_key: string | null;
63 readonly deleted_at: string | null;
64}
65
66interface LegacyRepairRow extends LegacyThreadRow {
67 readonly payload_json: string;
68}
69
70interface LegacyMessageRow {
71 readonly message_id: string;
72 readonly thread_id: string;
73 readonly role: "user" | "assistant";
74 readonly text: string;
75 readonly attachments_json: string | null;
76 readonly context_json?: string | null;
77 readonly is_streaming: number;
78 readonly created_at: string;
79 readonly updated_at: string;
80 readonly ordinal: number;
81}
82
83interface LegacyImportRow {
84 readonly thread_id: string;
85 readonly transcript_imported_at: string | null;
86}
87
88export interface LegacyV1ImportSummary {
89 readonly importedThreadCount: number;
90 readonly importedMessageCount: number;
91}
92
93export class LegacyV1ThreadImportError extends Schema.TaggedError<LegacyV1ThreadImportError>()(
94 "LegacyV1ThreadImportError",
95 {
96 operation: Schema.String,
97 threadId: Schema.optional(ThreadId),
98 cause: Schema.optional(Schema.Defect()),
99 },
100) {
101 override get message(): string {
102 return this.threadId === undefined
103 ? `Failed to ${this.operation} legacy v1 threads.`
104 : `Failed to ${this.operation} legacy v1 thread ${this.threadId}.`;
105 }
106}
107
108export interface LegacyV1ThreadImporterShape {
109 readonly pendingThreadCount: Effect.Effect<number, LegacyV1ThreadImportError>;
110 readonly reconcileShells: Effect.Effect<LegacyV1ImportSummary, LegacyV1ThreadImportError>;
111 readonly ensureTranscript: (
112 threadId: ThreadId,
113 ) => Effect.Effect<LegacyV1ImportSummary, LegacyV1ThreadImportError>;
114 readonly importPendingTranscripts: Effect.Effect<LegacyV1ImportSummary, never>;
115}
116
117export class LegacyV1ThreadImporter extends Context.Service<
118 LegacyV1ThreadImporter,
119 LegacyV1ThreadImporterShape
120>()("t3/orchestration-v2/legacy/LegacyV1ThreadImporter") {}
121
122const decodeModelSelection = Schema.decodeUnknownOption(ModelSelection);
123const decodeAttachments = Schema.decodeUnknownOption(Schema.Array(ChatAttachment));
124const decodePullRequests = Schema.decodeUnknownOption(Schema.Array(ThreadPullRequestLink));
125const decodeLinkedPullRequest = Schema.decodeUnknownOption(ThreadLinkedPullRequest);
126const decodeStoredThread = Schema.decodeUnknownOption(
127 Schema.fromJsonString(OrchestrationV2AppThreadJson),
128);
129
130function parseJson(json: string): unknown {
131 try {
132 return JSON.parse(json);
133 } catch {
134 return undefined;
135 }
136}
137
138function modelSelectionFor(row: LegacyThreadRow) {
139 const decoded =
140 row.model_selection_json === null
141 ? Option.none()
142 : decodeModelSelection(parseJson(row.model_selection_json));
143 return Option.getOrElse(decoded, () => ({
144 instanceId: ProviderInstanceId.make("codex"),
145 model: DEFAULT_MODEL,
146 }));
147}
148
149function attachmentsFor(row: LegacyMessageRow) {
150 if (row.attachments_json === null) return [];
151 return Option.getOrElse(decodeAttachments(parseJson(row.attachments_json)), () => []);
152}
153
154function linkedPullRequestFor(row: LegacyThreadRow) {
155 if (row.linked_pull_request_json === null) return null;
156 return Option.getOrNull(decodeLinkedPullRequest(parseJson(row.linked_pull_request_json)));
157}
158
159function branchPullRequestFor(row: LegacyThreadRow) {
160 if (row.branch_pull_request_json === null) return null;
161 return Option.getOrNull(decodeLinkedPullRequest(parseJson(row.branch_pull_request_json)));
162}
163
164function runtimeModeFor(value: string): OrchestrationV2AppThread["runtimeMode"] {
165 return value === "approval-required" ||
166 value === "auto-accept-edits" ||
167 value === "auto" ||
168 value === "full-access"
169 ? value
170 : "full-access";
171}
172
173function interactionModeFor(value: string): OrchestrationV2AppThread["interactionMode"] {
174 return value === "plan" ? "plan" : "default";
175}
176
177function settledOverrideFor(value: string | null): OrchestrationV2AppThread["settledOverride"] {
178 return value === "settled" || value === "active" ? value : null;
179}
180
181function dateTime(value: string): DateTime.Utc {
182 return DateTime.makeUnsafe(value);
183}
184
185function nullableDateTime(value: string | null): DateTime.Utc | null {
186 return value === null ? null : dateTime(value);
187}
188
189function importedThread(row: LegacyThreadRow): OrchestrationV2AppThread {
190 const threadId = ThreadId.make(row.thread_id);
191 const modelSelection = modelSelectionFor(row);
192 const branch = row.branch?.trim() || null;
193 const worktreePath = row.worktree_path?.trim() || null;
194 const pullRequests = Option.getOrElse(
195 decodePullRequests(parseJson(row.pull_requests_json)),
196 () => [],
197 );
198 const linkedPullRequest = linkedPullRequestFor(row);
199 const legacyLink = threadPullRequestsOf({ linkedPullRequest })[0];
200 const importedPullRequests =
201 legacyLink !== undefined &&
202 !pullRequests.some((link) => threadPullRequestKeysEqual(link, legacyLink))
203 ? [...pullRequests, legacyLink]
204 : pullRequests;
205 return {
206 createdBy: "system",
207 creationSource: "server",
208 id: threadId,
209 projectId: ProjectId.make(row.project_id),
210 title: row.title.trim() === "" ? "Untitled thread" : row.title,
211 providerInstanceId: modelSelection.instanceId,
212 modelSelection,
213 runtimeMode: runtimeModeFor(row.runtime_mode),
214 interactionMode: interactionModeFor(row.interaction_mode),
215 branch,
216 worktreePath,
217 linkedPullRequest,
218 pullRequests: importedPullRequests,
219 branchPullRequest: branchPullRequestFor(row),
220 activeOrderKey: row.active_order_key?.trim() || null,
221 activeProviderThreadId: null,
222 historyOrigin: "v1_import",
223 lineage: {
224 parentThreadId: null,
225 relationshipToParent: null,
226 rootThreadId: threadId,
227 },
228 forkedFrom: null,
229 createdAt: dateTime(row.created_at),
230 updatedAt: dateTime(row.updated_at),
231 archivedAt: nullableDateTime(row.archived_at),
232 settledOverride: settledOverrideFor(row.settled_override),
233 settledAt: nullableDateTime(row.settled_at),
234 unsettledAt: nullableDateTime(row.unsettled_at),
235 snoozedUntil: nullableDateTime(row.snoozed_until),
236 snoozedAt: nullableDateTime(row.snoozed_at),
237 pinnedAt: nullableDateTime(row.pinned_at),
238 autoSettleDisabledAt: nullableDateTime(row.auto_settle_disabled_at),
239 pinOrderKey: row.pin_order_key?.trim() || null,
240 lastVisitedAt: null,
241 deletedAt: nullableDateTime(row.deleted_at),
242 };
243}
244
245function messageEvents(row: LegacyMessageRow): ReadonlyArray<OrchestrationV2DomainEvent> {
246 const threadId = ThreadId.make(row.thread_id);
247 const messageId = MessageId.make(row.message_id);
248 const createdAt = dateTime(row.created_at);
249 const updatedAt = dateTime(row.updated_at);
250 const attachments = attachmentsFor(row);
251 const message: OrchestrationV2ConversationMessage = {
252 createdBy: row.role === "user" ? "user" : "agent",
253 creationSource: "server",
254 id: messageId,
255 threadId,
256 runId: null,
257 nodeId: null,
258 role: row.role,
259 text: row.text,
260 ...(row.context_json
261 ? {
262 context: Schema.decodeUnknownSync(OrchestrationMessageContext)(
263 parseJson(row.context_json),
264 ),
265 }
266 : {}),
267 attachments,
268 streaming: false,
269 createdAt,
270 updatedAt,
271 };
272 const baseTurnItem = {
273 id: TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${row.message_id}`),
274 threadId,
275 runId: null,
276 nodeId: null,
277 providerThreadId: null,
278 providerTurnId: null,
279 nativeItemRef: null,
280 parentItemId: null,
281 ordinal: row.ordinal,
282 status: row.is_streaming === 1 ? ("interrupted" as const) : ("completed" as const),
283 title: null,
284 startedAt: createdAt,
285 completedAt: updatedAt,
286 updatedAt,
287 };
288 const turnItem: OrchestrationV2TurnItem =
289 row.role === "user"
290 ? {
291 ...baseTurnItem,
292 createdBy: "user",
293 creationSource: "server",
294 type: "user_message",
295 messageId,
296 inputIntent: "turn_start",
297 text: row.text,
298 ...(row.context_json
299 ? {
300 context: Schema.decodeUnknownSync(OrchestrationMessageContext)(
301 parseJson(row.context_json),
302 ),
303 }
304 : {}),
305 attachments,
306 }
307 : {
308 ...baseTurnItem,
309 type: "assistant_message",
310 messageId,
311 text: row.text,
312 ...(row.context_json
313 ? {
314 context: Schema.decodeUnknownSync(OrchestrationMessageContext)(
315 parseJson(row.context_json),
316 ),
317 }
318 : {}),
319 streaming: false,
320 };
321 return [
322 {
323 id: EventId.make(`${IMPORT_EVENT_PREFIX}:message:${row.message_id}`),
324 type: "message.updated",
325 threadId,
326 occurredAt: updatedAt,
327 payload: message,
328 },
329 {
330 id: EventId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${row.message_id}`),
331 type: "turn-item.updated",
332 threadId,
333 occurredAt: updatedAt,
334 payload: turnItem,
335 },
336 ];
337}
338
339function chunks<A>(items: ReadonlyArray<A>, size: number): Array<ReadonlyArray<A>> {
340 const result: Array<ReadonlyArray<A>> = [];
341 for (let index = 0; index < items.length; index += size) {
342 result.push(items.slice(index, index + size));
343 }
344 return result;
345}
346
347const make = Effect.gen(function* () {
348 const sql = yield* SqlClient.SqlClient;
349 const eventSink = yield* EventSink.EventSinkV2;
350 const transcriptImports = yield* KeyedLock.make<ThreadId>();
351
352 const listMessages = (threadId: ThreadId) =>
353 sql<LegacyMessageRow>`
354 SELECT
355 message_id,
356 thread_id,
357 role,
358 text,
359 attachments_json,
360 context_json,
361 is_streaming,
362 created_at,
363 updated_at,
364 ROW_NUMBER() OVER (
365 PARTITION BY thread_id
366 ORDER BY created_at ASC, message_id ASC
367 ) AS ordinal
368 FROM projection_thread_messages
369 WHERE thread_id = ${threadId}
370 AND role IN ('user', 'assistant')
371 ORDER BY created_at ASC, message_id ASC
372 `;
373
374 const listShellMessages = (threadId: ThreadId) =>
375 Effect.gen(function* () {
376 const latest = yield* sql<LegacyMessageRow>`
377 SELECT
378 message.message_id,
379 message.thread_id,
380 message.role,
381 message.text,
382 message.attachments_json,
383 message.context_json,
384 message.is_streaming,
385 message.created_at,
386 message.updated_at,
387 (
388 SELECT COUNT(*)
389 FROM projection_thread_messages AS earlier
390 WHERE earlier.thread_id = message.thread_id
391 AND earlier.role IN ('user', 'assistant')
392 AND (
393 earlier.created_at < message.created_at
394 OR (
395 earlier.created_at = message.created_at
396 AND earlier.message_id <= message.message_id
397 )
398 )
399 ) AS ordinal
400 FROM projection_thread_messages AS message
401 WHERE message.thread_id = ${threadId}
402 AND message.role IN ('user', 'assistant')
403 ORDER BY message.created_at DESC, message.message_id DESC
404 LIMIT 1
405 `;
406 const latestUser = yield* sql<LegacyMessageRow>`
407 SELECT
408 message.message_id,
409 message.thread_id,
410 message.role,
411 message.text,
412 message.attachments_json,
413 message.context_json,
414 message.is_streaming,
415 message.created_at,
416 message.updated_at,
417 (
418 SELECT COUNT(*)
419 FROM projection_thread_messages AS earlier
420 WHERE earlier.thread_id = message.thread_id
421 AND earlier.role IN ('user', 'assistant')
422 AND (
423 earlier.created_at < message.created_at
424 OR (
425 earlier.created_at = message.created_at
426 AND earlier.message_id <= message.message_id
427 )
428 )
429 ) AS ordinal
430 FROM projection_thread_messages AS message
431 WHERE message.thread_id = ${threadId}
432 AND message.role = 'user'
433 ORDER BY message.created_at DESC, message.message_id DESC
434 LIMIT 1
435 `;
436 return [latestUser[0], latest[0]].filter(
437 (message, index, selected): message is LegacyMessageRow =>
438 message !== undefined &&
439 selected.findIndex((candidate) => candidate?.message_id === message.message_id) === index,
440 );
441 });
442
443 const reconcileShellsBase = Effect.gen(function* () {
444 const now = DateTime.formatIso(yield* DateTime.now);
445 const repairRows = yield* sql<LegacyRepairRow>`
446 SELECT
447 thread.thread_id,
448 thread.project_id,
449 thread.title,
450 thread.model_selection_json,
451 thread.runtime_mode,
452 thread.interaction_mode,
453 thread.branch,
454 thread.worktree_path,
455 thread.created_at,
456 thread.updated_at,
457 thread.archived_at,
458 thread.settled_override,
459 thread.settled_at,
460 thread.unsettled_at,
461 thread.snoozed_until,
462 thread.snoozed_at,
463 thread.pinned_at,
464 thread.auto_settle_disabled_at,
465 thread.pin_order_key,
466 (SELECT json_group_array(json_object('host', pr.host, 'repository', pr.repository, 'number', pr.number, 'url', pr.url, 'source', pr.source, 'linkedAt', pr.linked_at, 'snapshot', json(pr.snapshot_json), 'stack', json(pr.stack_json))) FROM projection_thread_pull_requests pr WHERE pr.thread_id = thread.thread_id) AS pull_requests_json,
467 thread.linked_pull_request_json,
468 thread.branch_pull_request_json,
469 thread.active_order_key,
470 thread.deleted_at,
471 projection.payload_json
472 FROM orchestration_v2_legacy_imports AS legacy_import
473 INNER JOIN projection_threads AS thread
474 ON thread.thread_id = legacy_import.thread_id
475 INNER JOIN orchestration_v2_projection_threads AS projection
476 ON projection.thread_id = legacy_import.thread_id
477 WHERE json_type(projection.payload_json, '$.pinnedAt') IS NULL
478 OR json_type(projection.payload_json, '$.pinOrderKey') IS NULL
479 OR json_type(projection.payload_json, '$.snoozedUntil') IS NULL
480 OR json_type(projection.payload_json, '$.snoozedAt') IS NULL
481 OR json_type(projection.payload_json, '$.unsettledAt') IS NULL
482 OR json_type(projection.payload_json, '$.linkedPullRequest') IS NULL
483 OR json_type(projection.payload_json, '$.pullRequests') IS NULL
484 OR json_type(projection.payload_json, '$.branchPullRequest') IS NULL
485 OR json_type(projection.payload_json, '$.activeOrderKey') IS NULL
486 ORDER BY thread.created_at ASC, thread.thread_id ASC
487 `;
488 let repairedThreadCount = 0;
489 for (const row of repairRows) {
490 const decoded = decodeStoredThread(row.payload_json);
491 if (Option.isNone(decoded)) continue;
492 const current = decoded.value;
493 const legacy = importedThread(row);
494 const legacyPullRequests = legacy.pullRequests ?? [];
495 const repaired: OrchestrationV2AppThread = {
496 ...current,
497 pinnedAt: current.pinnedAt === undefined ? legacy.pinnedAt : current.pinnedAt,
498 autoSettleDisabledAt:
499 current.autoSettleDisabledAt === undefined
500 ? legacy.autoSettleDisabledAt
501 : current.autoSettleDisabledAt,
502 pinOrderKey: current.pinOrderKey === undefined ? legacy.pinOrderKey : current.pinOrderKey,
503 snoozedUntil:
504 current.snoozedUntil === undefined ? legacy.snoozedUntil : current.snoozedUntil,
505 snoozedAt: current.snoozedAt === undefined ? legacy.snoozedAt : current.snoozedAt,
506 unsettledAt: current.unsettledAt === undefined ? legacy.unsettledAt : current.unsettledAt,
507 linkedPullRequest:
508 current.linkedPullRequest === undefined
509 ? legacy.linkedPullRequest
510 : current.linkedPullRequest,
511 pullRequests:
512 current.pullRequests === undefined
513 ? current.linkedPullRequest === null
514 ? []
515 : legacyPullRequests.length > 0
516 ? legacyPullRequests
517 : threadPullRequestsOf({
518 linkedPullRequest:
519 current.linkedPullRequest === undefined
520 ? legacy.linkedPullRequest
521 : current.linkedPullRequest,
522 })
523 : current.pullRequests,
524 branchPullRequest:
525 current.branchPullRequest === undefined
526 ? legacy.branchPullRequest
527 : current.branchPullRequest,
528 activeOrderKey:
529 current.activeOrderKey === undefined ? legacy.activeOrderKey : current.activeOrderKey,
530 };
531 // Later schema additions can require another repair for the same thread.
532 const repairId = yield* randomUuidV4;
533 yield* eventSink.write({
534 events: [
535 {
536 id: EventId.make(
537 `${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:metadata-repair:${repairId}`,
538 ),
539 type: "thread.metadata-updated",
540 threadId: repaired.id,
541 providerInstanceId: repaired.providerInstanceId,
542 occurredAt: dateTime(now),
543 payload: repaired,
544 },
545 ],
546 });
547 repairedThreadCount += 1;
548 }
549 const rows = yield* sql<LegacyThreadRow>`
550 SELECT
551 thread.thread_id,
552 thread.project_id,
553 thread.title,
554 thread.model_selection_json,
555 thread.runtime_mode,
556 thread.interaction_mode,
557 thread.branch,
558 thread.worktree_path,
559 thread.created_at,
560 thread.updated_at,
561 thread.archived_at,
562 thread.settled_override,
563 thread.settled_at,
564 thread.unsettled_at,
565 thread.snoozed_until,
566 thread.snoozed_at,
567 thread.pinned_at,
568 thread.auto_settle_disabled_at,
569 thread.pin_order_key,
570 (SELECT json_group_array(json_object('host', pr.host, 'repository', pr.repository, 'number', pr.number, 'url', pr.url, 'source', pr.source, 'linkedAt', pr.linked_at, 'snapshot', json(pr.snapshot_json), 'stack', json(pr.stack_json))) FROM projection_thread_pull_requests pr WHERE pr.thread_id = thread.thread_id) AS pull_requests_json,
571 thread.linked_pull_request_json,
572 thread.branch_pull_request_json,
573 thread.active_order_key,
574 thread.deleted_at
575 FROM projection_threads AS thread
576 WHERE NOT EXISTS (
577 SELECT 1
578 FROM orchestration_events AS event INDEXED BY orchestration_events_v2_created_threads_idx
579 WHERE event.application_event_version = 2
580 AND event.aggregate_kind = 'thread'
581 AND event.stream_id = thread.thread_id
582 AND event.event_type = 'thread.created'
583 )
584 ORDER BY thread.created_at ASC, thread.thread_id ASC
585 `;
586 let importedThreadCount = repairedThreadCount;
587 let importedMessageCount = 0;
588 for (const row of rows) {
589 const thread = importedThread(row);
590 const previews = yield* listShellMessages(thread.id);
591 const events: Array<OrchestrationV2DomainEvent> = [
592 {
593 id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:created`),
594 type: "thread.created",
595 threadId: thread.id,
596 providerInstanceId: thread.providerInstanceId,
597 occurredAt: thread.createdAt,
598 payload: thread,
599 },
600 ...previews.flatMap(messageEvents),
601 {
602 id: EventId.make(`${IMPORT_EVENT_PREFIX}:thread:${row.thread_id}:shell`),
603 type: "thread.metadata-updated",
604 threadId: thread.id,
605 providerInstanceId: thread.providerInstanceId,
606 occurredAt: thread.updatedAt,
607 payload: thread,
608 },
609 ];
610 yield* sql.withTransaction(
611 Effect.gen(function* () {
612 yield* Effect.forEach(
613 previews,
614 (message) =>
615 sql`
616 INSERT INTO orchestration_v2_turn_item_positions (
617 thread_id,
618 turn_item_id,
619 ordinal
620 )
621 VALUES (
622 ${thread.id},
623 ${TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${message.message_id}`)},
624 ${message.ordinal}
625 )
626 ON CONFLICT(thread_id, turn_item_id) DO NOTHING
627 `,
628 { discard: true },
629 );
630 yield* eventSink.write({ events });
631 yield* sql`
632 INSERT INTO orchestration_v2_legacy_imports (
633 thread_id,
634 source_updated_at,
635 shell_imported_at,
636 transcript_imported_at,
637 imported_message_count,
638 last_error
639 )
640 VALUES (
641 ${thread.id},
642 ${row.updated_at},
643 ${now},
644 NULL,
645 ${previews.length},
646 NULL
647 )
648 ON CONFLICT(thread_id) DO NOTHING
649 `;
650 }),
651 );
652 importedThreadCount += 1;
653 importedMessageCount += previews.length;
654 }
655 return { importedThreadCount, importedMessageCount };
656 });
657
658 const reconcileShells = reconcileShellsBase.pipe(
659 Effect.mapError((cause) => new LegacyV1ThreadImportError({ operation: "import", cause })),
660 );
661
662 const pendingThreadCount = sql<{ readonly count: number }>`
663 SELECT COUNT(*) AS count
664 FROM (
665 SELECT thread.thread_id
666 FROM projection_threads AS thread
667 WHERE NOT EXISTS (
668 SELECT 1
669 FROM orchestration_events AS event INDEXED BY orchestration_events_v2_created_threads_idx
670 WHERE event.application_event_version = 2
671 AND event.aggregate_kind = 'thread'
672 AND event.stream_id = thread.thread_id
673 AND event.event_type = 'thread.created'
674 )
675 UNION
676 SELECT legacy_import.thread_id
677 FROM orchestration_v2_legacy_imports AS legacy_import
678 WHERE legacy_import.transcript_imported_at IS NULL
679 )
680 `.pipe(
681 Effect.map((rows) => rows[0]?.count ?? 0),
682 Effect.mapError(
683 (cause) => new LegacyV1ThreadImportError({ operation: "inspect pending", cause }),
684 ),
685 );
686
687 // Threads whose transcript import this process has already confirmed.
688 // `transcript_imported_at` is never reset to NULL, so a positive answer
689 // stays valid for the process lifetime; ensureTranscript runs on most
690 // thread reads and command dispatches, so skipping the lock + lookup here
691 // keeps that path off the database entirely after first confirmation.
692 const confirmedTranscriptThreadIds = new Set<ThreadId>();
693
694 const ensureTranscriptBase = (threadId: ThreadId) =>
695 transcriptImports.withLock(
696 threadId,
697 Effect.gen(function* () {
698 const imports = yield* sql<LegacyImportRow>`
699 SELECT thread_id, transcript_imported_at
700 FROM orchestration_v2_legacy_imports
701 WHERE thread_id = ${threadId}
702 LIMIT 1
703 `;
704 const imported = imports[0];
705 if (imported === undefined || imported.transcript_imported_at !== null) {
706 if (imported !== undefined) {
707 confirmedTranscriptThreadIds.add(threadId);
708 }
709 return { importedThreadCount: 0, importedMessageCount: 0 };
710 }
711 const messages = yield* listMessages(threadId);
712 const existingRows = yield* sql<{ readonly event_id: string }>`
713 SELECT event_id
714 FROM orchestration_events
715 WHERE application_event_version = 2
716 AND aggregate_kind = 'thread'
717 AND stream_id = ${threadId}
718 AND event_id LIKE ${`${IMPORT_EVENT_PREFIX}:message:%`}
719 `;
720 const existing = new Set(existingRows.map((row) => row.event_id));
721 const missing = messages.filter(
722 (message) => !existing.has(`${IMPORT_EVENT_PREFIX}:message:${message.message_id}`),
723 );
724 for (const batch of chunks(missing, TRANSCRIPT_EVENT_BATCH_SIZE / 2)) {
725 yield* Effect.forEach(
726 batch,
727 (message) =>
728 sql`
729 INSERT INTO orchestration_v2_turn_item_positions (
730 thread_id,
731 turn_item_id,
732 ordinal
733 )
734 VALUES (
735 ${threadId},
736 ${TurnItemId.make(`${IMPORT_EVENT_PREFIX}:turn-item:${message.message_id}`)},
737 ${message.ordinal}
738 )
739 ON CONFLICT(thread_id, turn_item_id) DO NOTHING
740 `,
741 { discard: true },
742 );
743 yield* eventSink.write({ events: batch.flatMap(messageEvents) });
744 yield* Effect.yieldNow;
745 }
746 const now = DateTime.formatIso(yield* DateTime.now);
747 yield* sql`
748 UPDATE orchestration_v2_legacy_imports
749 SET
750 transcript_imported_at = ${now},
751 imported_message_count = ${messages.length},
752 last_error = NULL
753 WHERE thread_id = ${threadId}
754 `;
755 confirmedTranscriptThreadIds.add(threadId);
756 return {
757 importedThreadCount: 1,
758 importedMessageCount: missing.length,
759 };
760 }),
761 );
762
763 const ensureTranscript = (threadId: ThreadId) =>
764 confirmedTranscriptThreadIds.has(threadId)
765 ? Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 })
766 : ensureTranscriptBase(threadId).pipe(
767 Effect.mapError(
768 (cause) =>
769 new LegacyV1ThreadImportError({
770 operation: "hydrate transcript for",
771 threadId,
772 cause,
773 }),
774 ),
775 );
776
777 const importPendingTranscripts = Effect.gen(function* () {
778 const rows = yield* sql<LegacyImportRow>`
779 SELECT thread_id, transcript_imported_at
780 FROM orchestration_v2_legacy_imports
781 WHERE transcript_imported_at IS NULL
782 ORDER BY shell_imported_at ASC, thread_id ASC
783 `;
784 let importedThreadCount = 0;
785 let importedMessageCount = 0;
786 for (const row of rows) {
787 const result = yield* ensureTranscript(ThreadId.make(row.thread_id)).pipe(
788 Effect.tapError((error) =>
789 Effect.logWarning("Failed to hydrate migrated v1 thread transcript", {
790 threadId: row.thread_id,
791 cause: error,
792 }),
793 ),
794 Effect.catch(() =>
795 sql`
796 UPDATE orchestration_v2_legacy_imports
797 SET last_error = 'Transcript hydration failed; retry on next open.'
798 WHERE thread_id = ${row.thread_id}
799 `.pipe(
800 Effect.as({ importedThreadCount: 0, importedMessageCount: 0 }),
801 Effect.orElseSucceed(() => ({
802 importedThreadCount: 0,
803 importedMessageCount: 0,
804 })),
805 ),
806 ),
807 );
808 importedThreadCount += result.importedThreadCount;
809 importedMessageCount += result.importedMessageCount;
810 yield* Effect.yieldNow;
811 }
812 return { importedThreadCount, importedMessageCount };
813 }).pipe(
814 Effect.catchCause((cause) =>
815 Effect.logWarning("Legacy v1 transcript background import stopped", { cause }).pipe(
816 Effect.as({ importedThreadCount: 0, importedMessageCount: 0 }),
817 ),
818 ),
819 );
820
821 return LegacyV1ThreadImporter.of({
822 pendingThreadCount,
823 reconcileShells,
824 ensureTranscript,
825 importPendingTranscripts,
826 });
827});
828
829export const layer: Layer.Layer<
830 LegacyV1ThreadImporter,
831 never,
832 EventSink.EventSinkV2 | SqlClient.SqlClient
833> = Layer.effect(LegacyV1ThreadImporter, make);