1import { assistantCitationsToPlainText } from "@t3tools/shared/assistantCitations";
2import { ATTACHMENT_ONLY_BOOTSTRAP_PROMPT } from "@t3tools/shared/composerContextReferences";
3import * as Effect from "effect/Effect";
4import type * as SqlClient from "effect/sql/SqlClient";
5
6// SQLite's ltrim needs the characters matched by isAttachmentOnlyBootstrapPrompt's \s explicitly.
7const BOOTSTRAP_PREFIX_WHITESPACE =
8 "\t\n\v\f\r \u00a0\u1680\u2000\u2001\u2002\u2003\u2004\u2005\u2006\u2007\u2008\u2009\u200a\u2028\u2029\u202f\u205f\u3000\ufeff";
9
10export const nonPlaceholderMessageCondition = (sql: SqlClient.SqlClient, row = "messages") =>
11 sql.literal(`
12 CASE
13 WHEN substr(json_extract(${row}.payload_json, '$.text'), 1, 11) = 'Ultrathink:' COLLATE NOCASE
14 THEN ltrim(substr(json_extract(${row}.payload_json, '$.text'), 12), '${BOOTSTRAP_PREFIX_WHITESPACE}')
15 ELSE json_extract(${row}.payload_json, '$.text')
16 END COLLATE BINARY <> '${ATTACHMENT_ONLY_BOOTSTRAP_PROMPT.replaceAll("'", "''")}'
17`);
18
19/**
20 * Exclude attachment placeholders, generated notices and subagent threads. ACP subagents
21 * rewrite their finished answer on every streamed chunk.
22 * Migration triggers, backfill, rebuild and scans share this predicate; `row` may be `new`.
23 */
24export const indexedMessageCondition = (sql: SqlClient.SqlClient, row = "messages") =>
25 sql`
26 ${sql.literal(`
27 ${row}.streaming = 0
28 AND (
29 ${row}.role = 'assistant'
30 OR (
31 ${row}.role = 'user'
32 AND COALESCE(json_extract(${row}.payload_json, '$.createdBy'), 'user') = 'user'
33 )
34 )
35 AND NOT EXISTS (
36 SELECT 1 FROM orchestration_v2_projection_threads AS subagent_threads
37 WHERE subagent_threads.thread_id = ${row}.thread_id
38 AND json_extract(subagent_threads.payload_json, '$.lineage.relationshipToParent') = 'subagent'
39 )
40`)}
41 AND ${nonPlaceholderMessageCondition(sql, row)}
42`;
43
44// Curly apostrophes straight, as search folds a query, since iOS keyboards
45// type "’" and others type "'".
46const foldApostrophes = (text: string) => `replace(replace(${text}, '’', ''''), '‘', '''')`;
47
48const readableText = (row: string) =>
49 `COALESCE(${row}.search_text, json_extract(${row}.payload_json, '$.text'))`;
50
51/**
52 * What search_text holds for a message: its text as it reads, a quote's
53 * percent-encoded link replaced by its words and comment. NULL when that is
54 * the stored text, so most messages store nothing extra.
55 */
56export const readableSearchText = (text: string) => {
57 const readable = assistantCitationsToPlainText(text);
58 return readable === text ? null : readable;
59};
60
61/** A message's text as it reads, for thread search. */
62export const messageSearchText = (sql: SqlClient.SqlClient) =>
63 sql.literal(readableText("messages"));
64
65/**
66 * A message's stored text with apostrophes folded, what the index held before
67 * migration 81. Migration 69's backfill uses it. `payload` is the message's
68 * JSON column.
69 */
70export const indexedMessageText = (sql: SqlClient.SqlClient, payload: string) =>
71 sql.literal(foldApostrophes(`json_extract(${payload}, '$.text')`));
72
73// What the index stores: a message's readable text with apostrophes folded.
74// Migrations 69 and 74 create the triggers before migration 81 adds
75// search_text, and `t3 doctor` can rebuild a database that has not migrated,
76// so those fall back to the stored text.
77const indexedText = Effect.fn("MessageSearchIndex.indexedText")(function* (
78 sql: SqlClient.SqlClient,
79) {
80 const column = yield* sql`
81 SELECT 1 FROM pragma_table_info('orchestration_v2_projection_messages')
82 WHERE name = 'search_text'
83 `;
84 return column.length > 0
85 ? (row: string) => sql.literal(foldApostrophes(readableText(row)))
86 : (row: string) => indexedMessageText(sql, `${row}.payload_json`);
87});
88
89export const createMessageSearchTriggers = Effect.fn("MessageSearchIndex.createTriggers")(
90 function* (sql: SqlClient.SqlClient) {
91 const text = yield* indexedText(sql);
92 yield* sql`
93 CREATE TRIGGER orchestration_v2_message_search_insert
94 AFTER INSERT ON orchestration_v2_projection_messages
95 WHEN ${indexedMessageCondition(sql, "new")}
96 BEGIN
97 INSERT INTO orchestration_v2_message_search (rowid, text)
98 SELECT new.rowid, ${text("new")}
99 WHERE json_extract(new.payload_json, '$.text') IS NOT NULL;
100 END
101 `;
102 // Streaming deltas rewrite the row many times a second; skip them until the
103 // message settles. Upserts rewrite every column, so re-index only when what
104 // the index stores changed; migration 81's backfill writes only search_text.
105 // Deleting a rowid the index never held is a no-op.
106 yield* sql`
107 CREATE TRIGGER orchestration_v2_message_search_update
108 AFTER UPDATE ON orchestration_v2_projection_messages
109 WHEN (old.streaming = 0 OR new.streaming = 0)
110 AND (
111 old.streaming IS NOT new.streaming
112 OR old.role IS NOT new.role
113 OR ${text("old")} IS NOT ${text("new")}
114 OR json_extract(old.payload_json, '$.createdBy')
115 IS NOT json_extract(new.payload_json, '$.createdBy')
116 )
117 BEGIN
118 DELETE FROM orchestration_v2_message_search WHERE rowid = old.rowid AND old.streaming = 0;
119 INSERT INTO orchestration_v2_message_search (rowid, text)
120 SELECT new.rowid, ${text("new")}
121 WHERE ${indexedMessageCondition(sql, "new")}
122 AND json_extract(new.payload_json, '$.text') IS NOT NULL;
123 END
124 `;
125 yield* sql`
126 CREATE TRIGGER orchestration_v2_message_search_delete
127 AFTER DELETE ON orchestration_v2_projection_messages
128 WHEN old.streaming = 0
129 BEGIN
130 DELETE FROM orchestration_v2_message_search WHERE rowid = old.rowid;
131 END
132 `;
133 },
134);
135
136/** Replaces the search triggers with their current definition, where the index exists. */
137export const recreateMessageSearchTriggers = Effect.fn("MessageSearchIndex.recreateTriggers")(
138 function* (sql: SqlClient.SqlClient) {
139 const index = yield* sql`
140 SELECT 1 FROM sqlite_master WHERE name = 'orchestration_v2_message_search'
141 `;
142 if (index.length === 0) return;
143 yield* sql`DROP TRIGGER orchestration_v2_message_search_insert`;
144 yield* sql`DROP TRIGGER orchestration_v2_message_search_update`;
145 yield* sql`DROP TRIGGER orchestration_v2_message_search_delete`;
146 yield* createMessageSearchTriggers(sql);
147 },
148);
149
150/**
151 * Refills the search index from the messages table. The index is keyed by
152 * message rowid, so anything that may renumber or copy rows (VACUUM, a dev
153 * database copy) rebuilds it instead of trusting it. Databases from before
154 * migration 69, or on an SQLite without FTS5 trigrams, have no index.
155 */
156export const rebuildMessageSearch = Effect.fn("MessageSearchIndex.rebuild")(function* (
157 sql: SqlClient.SqlClient,
158) {
159 const index = yield* sql`
160 SELECT 1 FROM sqlite_master WHERE name = 'orchestration_v2_message_search'
161 `;
162 if (index.length === 0) return;
163 const text = yield* indexedText(sql);
164 // One transaction: an interrupt between clear and refill would leave an
165 // empty index that nothing repairs, since search only checks the table exists.
166 yield* sql.withTransaction(
167 Effect.gen(function* () {
168 yield* sql`
169 INSERT INTO orchestration_v2_message_search (orchestration_v2_message_search)
170 VALUES ('delete-all')
171 `;
172 yield* sql`
173 INSERT INTO orchestration_v2_message_search (rowid, text)
174 SELECT messages.rowid, ${text("messages")}
175 FROM orchestration_v2_projection_messages AS messages
176 WHERE ${indexedMessageCondition(sql)}
177 AND json_extract(messages.payload_json, '$.text') IS NOT NULL
178 `;
179 }),
180 );
181});