1import {
2 IsoDateTime,
3 OrchestrationThreadSearchSource,
4 type OrchestrationSearchThreadsInput,
5 type OrchestrationSearchThreadsResult,
6 ProjectId,
7 ThreadId,
8} from "@t3tools/contracts";
9import * as Context from "effect/Context";
10import * as Effect from "effect/Effect";
11import * as Layer from "effect/Layer";
12import * as Schema from "effect/Schema";
13import * as SqlClient from "effect/sql/SqlClient";
14import * as SqlSchema from "effect/sql/SqlSchema";
15import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
16
17/** Carries no query text: search input is user content. */
18export class ThreadSearchError extends Schema.TaggedError<ThreadSearchError>()(
19 "ThreadSearchError",
20 {
21 operation: Schema.Literals(["query", "decode"]),
22 cause: Schema.Defect(),
23 },
24) {
25 override get message(): string {
26 return `Thread search ${this.operation} failed.`;
27 }
28}
29
30const SearchRequest = Schema.Struct({ pattern: Schema.String, limit: Schema.Int });
31const SearchRow = Schema.Struct({
32 threadId: ThreadId,
33 projectId: ProjectId,
34 source: OrchestrationThreadSearchSource,
35 matchText: Schema.String,
36 messageCreatedAt: Schema.NullOr(IsoDateTime),
37});
38
39function escapeLikePattern(value: string): string {
40 return value.replaceAll("!", "!!").replaceAll("%", "!%").replaceAll("_", "!_");
41}
42
43function foldAsciiCase(value: string): string {
44 return value.replace(/[A-Z]/g, (character) => character.toLowerCase());
45}
46
47/** At most 240 characters, centred near the first match. */
48function buildSearchSnippet(text: string, query: string): string {
49 const normalizedText = text.replace(/\s+/g, " ").trim();
50 if (normalizedText.length <= 240) {
51 return normalizedText;
52 }
53 const normalizedQuery = foldAsciiCase(query.replace(/\s+/g, " ").trim());
54 const matchIndex = foldAsciiCase(normalizedText).indexOf(normalizedQuery);
55 const bodyLength = 236;
56 const idealStart = Math.max(0, matchIndex - 72);
57 const start = Math.min(idealStart, normalizedText.length - bodyLength);
58 const end = Math.min(normalizedText.length, start + bodyLength);
59 return `${start > 0 ? "…" : ""}${normalizedText.slice(start, end)}${
60 end < normalizedText.length ? "…" : ""
61 }`;
62}
63
64/**
65 * Searches the finished user and assistant messages of active V2 threads in
66 * active projects. Legacy V1 transcripts that have not been imported yet are
67 * not searched.
68 */
69export class ThreadSearch extends Context.Service<
70 ThreadSearch,
71 {
72 readonly search: (
73 input: OrchestrationSearchThreadsInput,
74 ) => Effect.Effect<OrchestrationSearchThreadsResult, ThreadSearchError>;
75 }
76>()("t3/orchestration-v2/ThreadSearch") {}
77
78/** @public Service construction is part of the canonical Effect module API. */
79export const make = Effect.gen(function* () {
80 const sql = yield* SqlClient.SqlClient;
81
82 // One best match per thread: user messages outrank assistant ones, then the
83 // newest message wins. Threads order by match kind, then recency.
84 const searchRows = SqlSchema.findAll({
85 Request: SearchRequest,
86 Result: SearchRow,
87 execute: ({ pattern, limit }) => sql`
88 WITH candidate AS (
89 SELECT
90 threads.thread_id,
91 threads.project_id,
92 messages.role,
93 json_extract(messages.payload_json, '$.text') AS match_text,
94 messages.created_at AS message_created_at,
95 messages.message_id,
96 threads.updated_at AS thread_updated_at
97 FROM orchestration_v2_projection_messages AS messages
98 INNER JOIN orchestration_v2_projection_threads AS threads
99 ON threads.thread_id = messages.thread_id
100 INNER JOIN projection_projects AS projects
101 ON projects.project_id = threads.project_id
102 WHERE threads.deleted_at IS NULL
103 AND threads.archived_at IS NULL
104 AND projects.deleted_at IS NULL
105 AND messages.streaming = 0
106 AND messages.role IN ('user', 'assistant')
107 AND json_extract(messages.payload_json, '$.text') LIKE ${pattern} ESCAPE '!'
108 ),
109 ranked AS (
110 SELECT
111 thread_id,
112 project_id,
113 role AS source,
114 match_text,
115 message_created_at,
116 CASE role WHEN 'user' THEN 0 ELSE 1 END AS match_rank,
117 thread_updated_at,
118 ROW_NUMBER() OVER (
119 PARTITION BY thread_id
120 ORDER BY
121 CASE role WHEN 'user' THEN 0 ELSE 1 END ASC,
122 message_created_at DESC,
123 message_id ASC
124 ) AS thread_match_rank
125 FROM candidate
126 )
127 SELECT
128 thread_id AS "threadId",
129 project_id AS "projectId",
130 source,
131 match_text AS "matchText",
132 message_created_at AS "messageCreatedAt"
133 FROM ranked
134 WHERE thread_match_rank = 1
135 ORDER BY match_rank ASC, thread_updated_at DESC, thread_id ASC
136 LIMIT ${limit}
137 `,
138 });
139
140 const search: ThreadSearch["Service"]["search"] = Effect.fn("ThreadSearch.search")(
141 function* (input) {
142 // The scan reads every message, so run it on the reader worker rather
143 // than the event loop.
144 const rows = yield* sql
145 .withTransaction(
146 searchRows({
147 pattern: `%${escapeLikePattern(input.query)}%`,
148 limit: input.limit ?? 50,
149 }),
150 )
151 .pipe(
152 NodeSqliteClient.readOnly,
153 Effect.mapError(
154 (cause) =>
155 new ThreadSearchError({
156 operation: Schema.isSchemaError(cause) ? "decode" : "query",
157 cause,
158 }),
159 ),
160 );
161 return {
162 matches: rows.map((row) => ({
163 threadId: row.threadId,
164 projectId: row.projectId,
165 source: row.source,
166 snippet: buildSearchSnippet(row.matchText, input.query),
167 messageCreatedAt: row.messageCreatedAt,
168 })),
169 };
170 },
171 );
172
173 return ThreadSearch.of({ search });
174});
175
176export const layer = Layer.effect(ThreadSearch, make);