1import {
2 normalizeThreadPullRequestKey,
3 threadPullRequestKeyOf,
4 threadPullRequestKeysEqual,
5 visibleThreadPullRequests,
6} from "../sourceControl/threadPullRequests.ts";
7import {
8 PullRequestLinkedThreadsResult,
9 PullRequestOperationError,
10 type OrchestrationV2AppThread,
11 type OrchestrationV2DomainEvent,
12 type ThreadId,
13 type ThreadPullRequestKey,
14} from "@t3tools/contracts";
15import * as DateTime from "effect/DateTime";
16import * as Effect from "effect/Effect";
17import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
18import * as Schema from "effect/Schema";
19import * as SqlClient from "effect/sql/SqlClient";
20
21const decodeLinkedThreads = Schema.decodeUnknownEffect(PullRequestLinkedThreadsResult);
22
23export const listLinkedPullRequestThreads = Effect.fn("listLinkedPullRequestThreads")(
24 function* (input: ThreadPullRequestKey) {
25 const key = normalizeThreadPullRequestKey(input);
26 const sql = yield* SqlClient.SqlClient;
27 const rows = yield* sql<{
28 id: string;
29 projectId: string;
30 title: string;
31 archivedAt: string | null;
32 settled: number;
33 host: string;
34 repository: string;
35 number: number;
36 url: string;
37 }>`
38 SELECT t.thread_id AS id, t.project_id AS "projectId", t.title,
39 t.archived_at AS "archivedAt",
40 (json_extract(t.payload_json, '$.settledAt') IS NOT NULL
41 OR json_extract(t.payload_json, '$.settledOverride') IS 'settled') AS settled,
42 json_extract(link.value, '$.host') AS host,
43 json_extract(link.value, '$.repository') AS repository,
44 json_extract(link.value, '$.number') AS number,
45 json_extract(link.value, '$.url') AS url,
46 t.updated_at AS "updatedAt"
47 FROM orchestration_v2_projection_threads AS t
48 JOIN json_each(t.payload_json, '$.pullRequests') AS link
49 WHERE json_extract(link.value, '$.number') = ${key.number}
50 AND json_extract(link.value, '$.source') != 'stack-dismissed'
51 AND t.deleted_at IS NULL
52 AND json_extract(t.payload_json, '$.lineage.sideChat') IS NOT 1
53 ORDER BY "updatedAt" DESC, id ASC
54 `.pipe(NodeSqliteClient.readOnly);
55 const threads = rows
56 .filter((row) => threadPullRequestKeysEqual(row, key))
57 .map(({ id, projectId, title, archivedAt, settled }) => ({
58 id,
59 projectId,
60 title,
61 archivedAt,
62 settled: settled === 1,
63 }));
64 return yield* decodeLinkedThreads({ threads });
65 },
66 Effect.mapError(
67 (cause) =>
68 new PullRequestOperationError({
69 operation: "linkedThreads",
70 detail: "Could not load linked threads.",
71 cause,
72 }),
73 ),
74);
75
76// Mirrors the rows and filters of `listLinkedPullRequestThreads`.
77function linkedKeys(thread: OrchestrationV2AppThread): ReadonlyArray<string> {
78 if (thread.deletedAt !== null || thread.lineage.sideChat === true) return [];
79 return visibleThreadPullRequests(thread.pullRequests ?? [])
80 .map(threadPullRequestKeyOf)
81 .toSorted();
82}
83
84function linkedThreadsView(
85 thread: OrchestrationV2AppThread,
86 keys: ReadonlyArray<string>,
87): string | null {
88 if (keys.length === 0) return null;
89 const settled = thread.settledAt !== null || thread.settledOverride === "settled";
90 return JSON.stringify([
91 keys,
92 thread.title,
93 thread.projectId,
94 thread.archivedAt !== null,
95 settled,
96 ]);
97}
98
99/**
100 * Tells how a domain event changed `pullRequests.linkedThreads` answers, so readers can refresh on
101 * change instead of polling. `reordered` means activity on a linked thread, or on one whose links
102 * are unknown since a restart, which can reorder answers. A thread's first full event after a
103 * restart counts as a change, since its earlier links are unknown.
104 */
105export function makeLinkedThreadsChangeDetector() {
106 const views = new Map<ThreadId, string | null>();
107 // Last activity of each thread with visible links.
108 const activeAt = new Map<ThreadId, number>();
109 const active = (threadId: ThreadId, at: DateTime.Utc): boolean => {
110 const previous = activeAt.get(threadId);
111 if (previous === undefined) return false;
112 const millis = DateTime.toEpochMillis(at);
113 if (millis <= previous) return false;
114 activeAt.set(threadId, millis);
115 return true;
116 };
117 return (event: OrchestrationV2DomainEvent): "changed" | "reordered" | null => {
118 // The projection stamps every non-thread event as the thread's activity. A thread not seen
119 // since startup may be linked already, since history is not replayed.
120 if (!event.type.startsWith("thread.")) {
121 if (!views.has(event.threadId)) return "reordered";
122 return active(event.threadId, event.occurredAt) ? "reordered" : null;
123 }
124 if (!("lineage" in event.payload)) return null;
125 const thread = event.payload;
126 const view = linkedThreadsView(thread, linkedKeys(thread));
127 const known = views.has(event.threadId) || event.type === "thread.created";
128 const previous = views.get(event.threadId) ?? null;
129 if (event.type === "thread.deleted") views.delete(event.threadId);
130 else views.set(event.threadId, view);
131 if (view === null) activeAt.delete(event.threadId);
132 if (!known || view !== previous) {
133 if (view !== null) activeAt.set(event.threadId, DateTime.toEpochMillis(thread.updatedAt));
134 return "changed";
135 }
136 return active(event.threadId, thread.updatedAt) ? "reordered" : null;
137 };
138}