From 5dde025b83aca89d3e66db8fcfd14215642b006f Mon Sep 17 00:00:00 2001
From: Calvin Low <calvinlow@pmadvisors.my>
Date: Wed, 15 Jul 2026 01:59:44 +0800
Subject: [PATCH] fix(history): give each thread its own SQLite connection

The daemon runs every sync through asyncio.to_thread(), so SyncHistory's single
cached connection gets handed to whichever pool thread runs the next source.
sqlite3 refuses it:

    SQLite objects created in a thread can only be used in that same thread.

The first source to finish wins the connection and succeeds; every other source
raises. Because SyncHistory.log() is called after the upload completes, the
documents do reach the knowledge base -- but the sync is recorded and reported as
an error, so the daemon's /health, its dashboard and its history all show failures
for syncs that actually worked.

Hold the connection in a threading.local() instead, so each thread gets its own.
Also open with WAL and a busy timeout, so the now-concurrent writers wait on the
brief write lock rather than raising 'database is locked'.

Fixes #39
---
 src/oikb/history.py | 36 +++++++++++++++++++++++++++---------
 1 file changed, 27 insertions(+), 9 deletions(-)

diff --git a/src/oikb/history.py b/src/oikb/history.py
index 37c2076..723c305 100644
--- a/src/oikb/history.py
+++ b/src/oikb/history.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import sqlite3
+import threading
 import time
 import uuid
 from pathlib import Path
@@ -41,14 +42,25 @@ class SyncHistory:
     def __init__(self, db_path: Path | None = None):
         self.db_path = db_path or _DEFAULT_DB
         self.db_path.parent.mkdir(parents=True, exist_ok=True)
-        self._conn: sqlite3.Connection | None = None
+        # One connection per thread. The daemon runs each sync through
+        # asyncio.to_thread(), so a single shared connection is handed to whichever
+        # pool thread happens to run the next source -- and sqlite3 refuses it with
+        # "SQLite objects created in a thread can only be used in that same thread",
+        # failing every source but the first.
+        self._local = threading.local()
 
     def _get_conn(self) -> sqlite3.Connection:
-        if self._conn is None:
-            self._conn = sqlite3.connect(str(self.db_path))
-            self._conn.row_factory = sqlite3.Row
-            self._conn.executescript(_SCHEMA)
-        return self._conn
+        conn: sqlite3.Connection | None = getattr(self._local, "conn", None)
+        if conn is None:
+            conn = sqlite3.connect(str(self.db_path), timeout=30.0)
+            conn.row_factory = sqlite3.Row
+            # WAL lets the concurrent sources read while one writes; the busy timeout
+            # makes the brief write-lock contention wait rather than raise.
+            conn.execute("PRAGMA journal_mode=WAL")
+            conn.execute("PRAGMA busy_timeout=30000")
+            conn.executescript(_SCHEMA)
+            self._local.conn = conn
+        return conn
 
     def log(
         self,
@@ -133,6 +145,12 @@ def clear(self, older_than_days: int = 30) -> int:
         return cursor.rowcount
 
     def close(self) -> None:
-        if self._conn:
-            self._conn.close()
-            self._conn = None
+        """Close this thread's connection, if it has one.
+
+        Connections are per-thread, so each thread closes its own; the others are
+        released when their thread exits.
+        """
+        conn: sqlite3.Connection | None = getattr(self._local, "conn", None)
+        if conn is not None:
+            conn.close()
+            self._local.conn = None