Mercurial

comparison mrjunejune/conversation_store.c @ 260:1f9877b637e9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Add Copilot-powered cyberpunk JRPG chat Integrate the production JRPG chat with Seobeo streaming, Deita persistence, and a Bazel-managed Copilot SDK and LiteLLM inference stack. Co-authored-by: Copilot <[email protected]>
author MrJuneJune <mrjunejune@users.noreply.github.com>
date Wed, 05 Aug 2026 09:19:41 -0700
parents
children 04fee26ecce0
comparison
equal deleted inserted replaced
259:667156fcd3e3 260:1f9877b637e9
1 #include "mrjunejune/conversation_store.h"
2
3 #include "deita/deita.h"
4
5 #include <fcntl.h>
6 #include <pthread.h>
7 #include <stdio.h>
8 #include <stdlib.h>
9 #include <string.h>
10 #include <unistd.h>
11
12 struct Conversation_Store {
13 Deita_Connection *p_connection;
14 pthread_mutex_t mutex;
15 };
16
17 boolean Conversation_Store_Generate_UUID(char output[37])
18 {
19 uint8 bytes[16];
20 int fd = open("/dev/urandom", O_RDONLY);
21 if (fd < 0)
22 return FALSE;
23 size_t offset = 0;
24 while (offset < sizeof(bytes))
25 {
26 ssize_t amount = read(fd, bytes + offset, sizeof(bytes) - offset);
27 if (amount <= 0)
28 {
29 close(fd);
30 return FALSE;
31 }
32 offset += (size_t)amount;
33 }
34 close(fd);
35
36 bytes[6] = (uint8)((bytes[6] & 0x0f) | 0x40);
37 bytes[8] = (uint8)((bytes[8] & 0x3f) | 0x80);
38 snprintf(
39 output,
40 37,
41 "%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-"
42 "%02x%02x%02x%02x%02x%02x",
43 bytes[0], bytes[1], bytes[2], bytes[3],
44 bytes[4], bytes[5], bytes[6], bytes[7],
45 bytes[8], bytes[9], bytes[10], bytes[11],
46 bytes[12], bytes[13], bytes[14], bytes[15]);
47 return TRUE;
48 }
49
50 static char *Conversation_Store_Copy_Text(
51 const char *value,
52 Dowa_Arena *p_arena)
53 {
54 const char *source = value ? value : "";
55 size_t length = strlen(source);
56 char *copy = Dowa_Arena_Allocate(p_arena, length + 1);
57 if (!copy)
58 return NULL;
59 memcpy(copy, source, length + 1);
60 return copy;
61 }
62
63 static boolean Conversation_Store_Exists_Locked(
64 Conversation_Store *p_store,
65 const char *conversation_id)
66 {
67 Dowa_Arena *p_arena = Dowa_Arena_Create(1024);
68 if (!p_arena)
69 return FALSE;
70 const char *parameters[] = {conversation_id};
71 Deita_Result_Set *p_result = Deita_Query_Execute_Prepared(
72 p_store->p_connection,
73 "SELECT 1 FROM conversations WHERE id = ? AND status != 'deleted'",
74 1,
75 parameters,
76 p_arena);
77 boolean exists = p_result && Deita_Result_Set_Next(p_result);
78 if (p_result)
79 Deita_Result_Set_Free(p_result);
80 Dowa_Arena_Free(p_arena);
81 return exists;
82 }
83
84 static Conversation_Store_Result Conversation_Store_Rollback(
85 Conversation_Store *p_store,
86 Conversation_Store_Result result)
87 {
88 Deita_Query_Execute_Update(p_store->p_connection, "ROLLBACK");
89 return result;
90 }
91
92 Conversation_Store *Conversation_Store_Create(const char *database_path)
93 {
94 if (!database_path)
95 return NULL;
96
97 Conversation_Store *p_store = calloc(1, sizeof(*p_store));
98 if (!p_store)
99 return NULL;
100 p_store->p_connection = Deita_Connection_Create(
101 DEITA_DATABASE_TYPE_SQLITE3,
102 database_path);
103 if (!p_store->p_connection ||
104 !Deita_Connection_Is_Open(p_store->p_connection))
105 {
106 if (p_store->p_connection)
107 Deita_Connection_Close(p_store->p_connection);
108 free(p_store);
109 return NULL;
110 }
111 if (pthread_mutex_init(&p_store->mutex, NULL) != 0)
112 {
113 Deita_Connection_Close(p_store->p_connection);
114 free(p_store);
115 return NULL;
116 }
117
118 const char *schema =
119 "PRAGMA foreign_keys = ON;"
120 "PRAGMA journal_mode = WAL;"
121 "CREATE TABLE IF NOT EXISTS conversations ("
122 "id TEXT PRIMARY KEY,"
123 "copilot_session_id TEXT NOT NULL UNIQUE,"
124 "title TEXT NOT NULL DEFAULT '',"
125 "status TEXT NOT NULL DEFAULT 'active',"
126 "created_at INTEGER NOT NULL DEFAULT (strftime('%s','now')),"
127 "updated_at INTEGER NOT NULL DEFAULT (strftime('%s','now'))"
128 ");"
129 "CREATE TABLE IF NOT EXISTS conversation_turns ("
130 "id INTEGER PRIMARY KEY AUTOINCREMENT,"
131 "conversation_id TEXT NOT NULL,"
132 "sequence INTEGER NOT NULL,"
133 "role TEXT NOT NULL,"
134 "content TEXT NOT NULL DEFAULT '',"
135 "status TEXT NOT NULL,"
136 "request_id TEXT,"
137 "error_message TEXT,"
138 "input_tokens INTEGER NOT NULL DEFAULT 0,"
139 "output_tokens INTEGER NOT NULL DEFAULT 0,"
140 "created_at INTEGER NOT NULL DEFAULT (strftime('%s','now')),"
141 "completed_at INTEGER,"
142 "FOREIGN KEY(conversation_id) REFERENCES conversations(id) "
143 "ON DELETE CASCADE,"
144 "UNIQUE(conversation_id, sequence)"
145 ");"
146 "CREATE UNIQUE INDEX IF NOT EXISTS "
147 "idx_conversation_active_request "
148 "ON conversation_turns(conversation_id) "
149 "WHERE role = 'assistant' AND status = 'active';"
150 "CREATE INDEX IF NOT EXISTS idx_conversations_updated "
151 "ON conversations(updated_at DESC);"
152 "CREATE INDEX IF NOT EXISTS idx_turns_conversation_sequence "
153 "ON conversation_turns(conversation_id, sequence);";
154 if (Deita_Query_Execute_Update(p_store->p_connection, schema) < 0)
155 {
156 Conversation_Store_Destroy(p_store);
157 return NULL;
158 }
159 if (Deita_Query_Execute_Update(
160 p_store->p_connection,
161 "UPDATE conversation_turns "
162 "SET status = 'failed', "
163 "error_message = 'Interrupted by server restart', "
164 "completed_at = strftime('%s','now') "
165 "WHERE role = 'assistant' AND status = 'active'") < 0)
166 {
167 Conversation_Store_Destroy(p_store);
168 return NULL;
169 }
170 return p_store;
171 }
172
173 void Conversation_Store_Destroy(Conversation_Store *p_store)
174 {
175 if (!p_store)
176 return;
177 if (p_store->p_connection)
178 Deita_Connection_Close(p_store->p_connection);
179 pthread_mutex_destroy(&p_store->mutex);
180 free(p_store);
181 }
182
183 Conversation_Store_Result Conversation_Store_Create_Conversation(
184 Conversation_Store *p_store,
185 const char *title,
186 char output_id[37])
187 {
188 if (!p_store || !output_id ||
189 !Conversation_Store_Generate_UUID(output_id))
190 return CONVERSATION_STORE_ERROR;
191
192 const char *parameters[] = {
193 output_id,
194 output_id,
195 title ? title : "",
196 };
197 pthread_mutex_lock(&p_store->mutex);
198 int32 result = Deita_Query_Execute_Update_Prepared(
199 p_store->p_connection,
200 "INSERT INTO conversations (id, copilot_session_id, title) "
201 "VALUES (?, ?, ?)",
202 3,
203 parameters);
204 pthread_mutex_unlock(&p_store->mutex);
205 return result < 0 ? CONVERSATION_STORE_ERROR : CONVERSATION_STORE_OK;
206 }
207
208 Conversation_Store_Result Conversation_Store_Get(
209 Conversation_Store *p_store,
210 const char *conversation_id,
211 Conversation_Record *p_record,
212 Dowa_Arena *p_arena)
213 {
214 if (!p_store || !conversation_id || !p_record || !p_arena)
215 return CONVERSATION_STORE_ERROR;
216 memset(p_record, 0, sizeof(*p_record));
217
218 const char *parameters[] = {conversation_id};
219 pthread_mutex_lock(&p_store->mutex);
220 Deita_Result_Set *p_result = Deita_Query_Execute_Prepared(
221 p_store->p_connection,
222 "SELECT id, copilot_session_id, title, status, created_at, updated_at "
223 "FROM conversations WHERE id = ? AND status != 'deleted'",
224 1,
225 parameters,
226 p_arena);
227 if (!p_result || !Deita_Result_Set_Next(p_result))
228 {
229 if (p_result)
230 Deita_Result_Set_Free(p_result);
231 pthread_mutex_unlock(&p_store->mutex);
232 return CONVERSATION_STORE_NOT_FOUND;
233 }
234 p_record->id = Conversation_Store_Copy_Text(
235 Deita_Result_Set_Get_Text(p_result, 0), p_arena);
236 p_record->copilot_session_id = Conversation_Store_Copy_Text(
237 Deita_Result_Set_Get_Text(p_result, 1), p_arena);
238 p_record->title = Conversation_Store_Copy_Text(
239 Deita_Result_Set_Get_Text(p_result, 2), p_arena);
240 p_record->status = Conversation_Store_Copy_Text(
241 Deita_Result_Set_Get_Text(p_result, 3), p_arena);
242 p_record->created_at = Deita_Result_Set_Get_Integer(p_result, 4);
243 p_record->updated_at = Deita_Result_Set_Get_Integer(p_result, 5);
244 Deita_Result_Set_Free(p_result);
245
246 p_result = Deita_Query_Execute_Prepared(
247 p_store->p_connection,
248 "SELECT id, sequence, role, content, status, request_id, "
249 "error_message, input_tokens, output_tokens, created_at, completed_at "
250 "FROM (SELECT id, sequence, role, content, status, request_id, "
251 "error_message, input_tokens, output_tokens, created_at, completed_at "
252 "FROM conversation_turns WHERE conversation_id = ? "
253 "ORDER BY sequence DESC LIMIT 20) ORDER BY sequence",
254 1,
255 parameters,
256 p_arena);
257 if (!p_result)
258 {
259 pthread_mutex_unlock(&p_store->mutex);
260 return CONVERSATION_STORE_ERROR;
261 }
262 while (p_result && Deita_Result_Set_Next(p_result))
263 {
264 Conversation_Turn turn = {0};
265 turn.id = Deita_Result_Set_Get_Integer(p_result, 0);
266 turn.sequence = Deita_Result_Set_Get_Integer(p_result, 1);
267 turn.role = Conversation_Store_Copy_Text(
268 Deita_Result_Set_Get_Text(p_result, 2), p_arena);
269 turn.content = Conversation_Store_Copy_Text(
270 Deita_Result_Set_Get_Text(p_result, 3), p_arena);
271 turn.status = Conversation_Store_Copy_Text(
272 Deita_Result_Set_Get_Text(p_result, 4), p_arena);
273 turn.request_id = Conversation_Store_Copy_Text(
274 Deita_Result_Set_Get_Text(p_result, 5), p_arena);
275 turn.error_message = Conversation_Store_Copy_Text(
276 Deita_Result_Set_Get_Text(p_result, 6), p_arena);
277 turn.input_tokens = Deita_Result_Set_Get_Integer(p_result, 7);
278 turn.output_tokens = Deita_Result_Set_Get_Integer(p_result, 8);
279 turn.created_at = Deita_Result_Set_Get_Integer(p_result, 9);
280 turn.completed_at = Deita_Result_Set_Get_Integer(p_result, 10);
281 Dowa_Array_Push_Arena(p_record->turns, turn, p_arena);
282 }
283 boolean turns_error = Deita_Result_Set_Has_Error(p_result);
284 if (p_result)
285 Deita_Result_Set_Free(p_result);
286 pthread_mutex_unlock(&p_store->mutex);
287 return turns_error ? CONVERSATION_STORE_ERROR : CONVERSATION_STORE_OK;
288 }
289
290 Conversation_Store_Result Conversation_Store_Update_Title(
291 Conversation_Store *p_store,
292 const char *conversation_id,
293 const char *title)
294 {
295 if (!p_store || !conversation_id || !title)
296 return CONVERSATION_STORE_ERROR;
297 const char *parameters[] = {title, conversation_id};
298 pthread_mutex_lock(&p_store->mutex);
299 int32 result = Deita_Query_Execute_Update_Prepared(
300 p_store->p_connection,
301 "UPDATE conversations SET title = ?, "
302 "updated_at = strftime('%s','now') "
303 "WHERE id = ? AND status != 'deleted'",
304 2,
305 parameters);
306 pthread_mutex_unlock(&p_store->mutex);
307 if (result < 0)
308 return CONVERSATION_STORE_ERROR;
309 return result == 0 ? CONVERSATION_STORE_NOT_FOUND : CONVERSATION_STORE_OK;
310 }
311
312 Conversation_Store_Result Conversation_Store_Delete(
313 Conversation_Store *p_store,
314 const char *conversation_id)
315 {
316 if (!p_store || !conversation_id)
317 return CONVERSATION_STORE_ERROR;
318 const char *parameters[] = {conversation_id};
319 pthread_mutex_lock(&p_store->mutex);
320 int32 result = Deita_Query_Execute_Update_Prepared(
321 p_store->p_connection,
322 "DELETE FROM conversations WHERE id = ?",
323 1,
324 parameters);
325 pthread_mutex_unlock(&p_store->mutex);
326 if (result < 0)
327 return CONVERSATION_STORE_ERROR;
328 return result == 0 ? CONVERSATION_STORE_NOT_FOUND : CONVERSATION_STORE_OK;
329 }
330
331 Conversation_Store_Result Conversation_Store_Begin_Turn(
332 Conversation_Store *p_store,
333 const char *conversation_id,
334 const char *request_id,
335 const char *prompt)
336 {
337 if (!p_store || !conversation_id || !request_id || !prompt)
338 return CONVERSATION_STORE_ERROR;
339
340 pthread_mutex_lock(&p_store->mutex);
341 if (Deita_Query_Execute_Update(
342 p_store->p_connection, "BEGIN IMMEDIATE") < 0)
343 {
344 pthread_mutex_unlock(&p_store->mutex);
345 return CONVERSATION_STORE_ERROR;
346 }
347 if (!Conversation_Store_Exists_Locked(p_store, conversation_id))
348 {
349 Conversation_Store_Rollback(
350 p_store, CONVERSATION_STORE_NOT_FOUND);
351 pthread_mutex_unlock(&p_store->mutex);
352 return CONVERSATION_STORE_NOT_FOUND;
353 }
354
355 Dowa_Arena *p_arena = Dowa_Arena_Create(2048);
356 const char *parameters[] = {conversation_id};
357 Deita_Result_Set *p_result = Deita_Query_Execute_Prepared(
358 p_store->p_connection,
359 "SELECT COALESCE(MAX(sequence), 0), "
360 "SUM(CASE WHEN role = 'assistant' AND status = 'active' "
361 "THEN 1 ELSE 0 END) "
362 "FROM conversation_turns WHERE conversation_id = ?",
363 1,
364 parameters,
365 p_arena);
366 if (!p_result || !Deita_Result_Set_Next(p_result))
367 {
368 if (p_result)
369 Deita_Result_Set_Free(p_result);
370 Dowa_Arena_Free(p_arena);
371 Conversation_Store_Rollback(p_store, CONVERSATION_STORE_ERROR);
372 pthread_mutex_unlock(&p_store->mutex);
373 return CONVERSATION_STORE_ERROR;
374 }
375 int64 next_sequence = Deita_Result_Set_Get_Integer(p_result, 0) + 1;
376 int64 active_count = Deita_Result_Set_Get_Integer(p_result, 1);
377 Deita_Result_Set_Free(p_result);
378 Dowa_Arena_Free(p_arena);
379 if (active_count > 0)
380 {
381 Conversation_Store_Rollback(p_store, CONVERSATION_STORE_CONFLICT);
382 pthread_mutex_unlock(&p_store->mutex);
383 return CONVERSATION_STORE_CONFLICT;
384 }
385
386 char user_sequence[32];
387 char assistant_sequence[32];
388 snprintf(user_sequence, sizeof(user_sequence), "%lld",
389 (long long)next_sequence);
390 snprintf(assistant_sequence, sizeof(assistant_sequence), "%lld",
391 (long long)(next_sequence + 1));
392 const char *user_parameters[] = {
393 conversation_id, user_sequence, prompt, request_id,
394 };
395 if (Deita_Query_Execute_Update_Prepared(
396 p_store->p_connection,
397 "INSERT INTO conversation_turns "
398 "(conversation_id, sequence, role, content, status, request_id, "
399 "completed_at) VALUES (?, ?, 'user', ?, 'complete', ?, "
400 "strftime('%s','now'))",
401 4,
402 user_parameters) < 0)
403 {
404 Conversation_Store_Rollback(p_store, CONVERSATION_STORE_ERROR);
405 pthread_mutex_unlock(&p_store->mutex);
406 return CONVERSATION_STORE_ERROR;
407 }
408 const char *assistant_parameters[] = {
409 conversation_id, assistant_sequence, request_id,
410 };
411 if (Deita_Query_Execute_Update_Prepared(
412 p_store->p_connection,
413 "INSERT INTO conversation_turns "
414 "(conversation_id, sequence, role, status, request_id) "
415 "VALUES (?, ?, 'assistant', 'active', ?)",
416 3,
417 assistant_parameters) < 0 ||
418 Deita_Query_Execute_Update_Prepared(
419 p_store->p_connection,
420 "UPDATE conversations SET updated_at = strftime('%s','now') "
421 "WHERE id = ?",
422 1,
423 parameters) < 0 ||
424 Deita_Query_Execute_Update(
425 p_store->p_connection, "COMMIT") < 0)
426 {
427 Conversation_Store_Rollback(p_store, CONVERSATION_STORE_ERROR);
428 pthread_mutex_unlock(&p_store->mutex);
429 return CONVERSATION_STORE_ERROR;
430 }
431 pthread_mutex_unlock(&p_store->mutex);
432 return CONVERSATION_STORE_OK;
433 }
434
435 Conversation_Store_Result Conversation_Store_Complete_Turn(
436 Conversation_Store *p_store,
437 const char *conversation_id,
438 const char *request_id,
439 const char *content,
440 int64 input_tokens,
441 int64 output_tokens)
442 {
443 if (!p_store || !conversation_id || !request_id || !content)
444 return CONVERSATION_STORE_ERROR;
445 char input_value[32];
446 char output_value[32];
447 snprintf(input_value, sizeof(input_value), "%lld", (long long)input_tokens);
448 snprintf(output_value, sizeof(output_value), "%lld", (long long)output_tokens);
449 const char *parameters[] = {
450 content, input_value, output_value, conversation_id, request_id,
451 };
452 pthread_mutex_lock(&p_store->mutex);
453 int32 result = Deita_Query_Execute_Update_Prepared(
454 p_store->p_connection,
455 "UPDATE conversation_turns SET content = ?, status = 'complete', "
456 "input_tokens = ?, output_tokens = ?, "
457 "completed_at = strftime('%s','now') "
458 "WHERE conversation_id = ? AND request_id = ? "
459 "AND role = 'assistant' AND status = 'active'",
460 5,
461 parameters);
462 pthread_mutex_unlock(&p_store->mutex);
463 if (result < 0)
464 return CONVERSATION_STORE_ERROR;
465 return result == 0 ? CONVERSATION_STORE_NOT_FOUND : CONVERSATION_STORE_OK;
466 }
467
468 Conversation_Store_Result Conversation_Store_Fail_Turn(
469 Conversation_Store *p_store,
470 const char *conversation_id,
471 const char *request_id,
472 const char *error_message,
473 boolean aborted)
474 {
475 if (!p_store || !conversation_id || !request_id)
476 return CONVERSATION_STORE_ERROR;
477 const char *parameters[] = {
478 aborted ? "aborted" : "failed",
479 error_message ? error_message : "",
480 conversation_id,
481 request_id,
482 };
483 pthread_mutex_lock(&p_store->mutex);
484 int32 result = Deita_Query_Execute_Update_Prepared(
485 p_store->p_connection,
486 "UPDATE conversation_turns SET status = ?, error_message = ?, "
487 "completed_at = strftime('%s','now') "
488 "WHERE conversation_id = ? AND request_id = ? "
489 "AND role = 'assistant' AND status = 'active'",
490 4,
491 parameters);
492 pthread_mutex_unlock(&p_store->mutex);
493 if (result < 0)
494 return CONVERSATION_STORE_ERROR;
495 return result == 0 ? CONVERSATION_STORE_NOT_FOUND : CONVERSATION_STORE_OK;
496 }