prevent sqlite request initialization locks

apps/api/app/main.py

@@ -45,6 +45,8 @@ WEBSITE_SCAN_CACHE_SECONDS = 3600
 CONTACT_EXTRACTION_PAGE_SIZE = 100
 SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "credential", "private_key"}
 CHILD_TABLES = {"contacts": ("name", "email", "phone", "title", "do_not_contact"), "domains": ("domain", "kind"), "websites": ("url", "website_class"), "evidence": ("kind", "url", "claim"), "notes": ("body",)}
+_DB_INIT_LOCK = threading.Lock()
+_INITIALIZED_DATABASES: set[str] = set()
 
 def redact(value):
     if isinstance(value, dict): return {k: ("[REDACTED]" if str(k).lower() in SECRET_KEYS or any(s in str(k).lower() for s in ("password", "token", "secret", "api_key")) else redact(v)) for k,v in value.items()}
@@ -65,7 +67,16 @@ def verify_password(password, encoded_hash, encoded_salt):
     except (TypeError, ValueError): return False
 
 def connect(db_path: str) -> sqlite3.Connection:
-    db = sqlite3.connect(db_path); db.row_factory = sqlite3.Row; db.execute("PRAGMA foreign_keys = ON"); db.executescript(SCHEMA.read_text())
+    normalized=os.path.abspath(db_path)
+    with _DB_INIT_LOCK:
+        if normalized not in _INITIALIZED_DATABASES:
+            initialized=_initialize_database(normalized); initialized.close(); _INITIALIZED_DATABASES.add(normalized)
+    db=sqlite3.connect(normalized, timeout=10); db.row_factory=sqlite3.Row
+    db.execute("PRAGMA foreign_keys=ON"); db.execute("PRAGMA busy_timeout=10000")
+    return db
+
+def _initialize_database(db_path: str) -> sqlite3.Connection:
+    db = sqlite3.connect(db_path, timeout=10); db.row_factory = sqlite3.Row; db.execute("PRAGMA busy_timeout=10000"); db.execute("PRAGMA journal_mode=WAL"); db.execute("PRAGMA foreign_keys = ON"); db.executescript(SCHEMA.read_text())
     # Upgrade databases created by Phase 1/2 without destroying data.
     cols = {r[1] for r in db.execute("PRAGMA table_info(businesses)")}
     for col, definition in (("verified", "INTEGER NOT NULL DEFAULT 0"), ("verified_at", "TEXT"), ("updated_at", "TEXT"), ("province", "TEXT NOT NULL DEFAULT ''"), ("city", "TEXT NOT NULL DEFAULT ''"), ("suburb", "TEXT NOT NULL DEFAULT ''"), ("merge_status", "TEXT NOT NULL DEFAULT 'active'"), ("merged_into_id", "INTEGER"), ("review_status", "TEXT NOT NULL DEFAULT 'pending'"), ("assigned_to", "TEXT NOT NULL DEFAULT ''"), ("review_metadata_json", "TEXT NOT NULL DEFAULT '{}'")):
@@ -175,7 +186,9 @@ class ApiHandler(BaseHTTPRequestHandler):
     def send_json(self, status, payload, extra_headers=None):
         body = json.dumps(payload, sort_keys=True, default=str).encode(); self.send_response(status); self.send_header("Content-Type","application/json; charset=utf-8"); self.send_header("Cache-Control","no-store, private"); self.send_header("Pragma","no-cache"); self.send_header("Vary","Cookie, Origin"); self.send_header("Access-Control-Allow-Origin",os.environ.get("CORS_ORIGINS","http://localhost:8080")); self.send_header("Access-Control-Allow-Credentials","true"); self.send_header("Access-Control-Allow-Methods","GET, POST, PATCH, OPTIONS"); self.send_header("Access-Control-Allow-Headers","Content-Type")
         for k,v in (extra_headers or {}).items(): self.send_header(k,v)
-        self.send_header("Content-Length",str(len(body))); self.end_headers(); self.wfile.write(body)
+        self.send_header("Content-Length",str(len(body))); self.end_headers()
+        try: self.wfile.write(body)
+        except BrokenPipeError: return
     def read_json(self):
         try:
             value=json.loads(self.rfile.read(int(self.headers.get("Content-Length","0"))) or b"{}"); return value if isinstance(value,dict) else {}