From ea875cc7afe6df4cb94ec87fbdcea48bf96015a4 Mon Sep 17 00:00:00 2001
From: JohnBaumb <80135794+JohnBaumb@users.noreply.github.com>
Date: Sun, 19 Apr 2026 14:31:33 -0700
Subject: [PATCH] fix: throttle auth last_used_at config writes

Every authenticated API request previously called config_mgr.set(api_keys),
which rewrites the entire app config blob to SQLite. Under load this caused
significant write amplification and lock contention.

Persistence of last_used_at is now throttled per key hash to once every
15 minutes. The in-memory timestamp on the matched key is still updated
immediately, so reads within the same process see the live value; only
the on-disk persistence is throttled.
---
 api/auth.py | 33 +++++++++++++++++++++++++++++----
 1 file changed, 29 insertions(+), 4 deletions(-)

diff --git a/api/auth.py b/api/auth.py
index 3ddecc42..c6f0fb33 100644
--- a/api/auth.py
+++ b/api/auth.py
@@ -4,8 +4,9 @@ API key authentication for the SoulSync public API.
 
 import hashlib
 import secrets
+import threading
 import uuid
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from functools import wraps
 
 from flask import request, current_app
@@ -13,6 +14,27 @@ from flask import request, current_app
 from .helpers import api_error
 
 
+# Throttle persistence of `last_used_at` so every authenticated request
+# does not rewrite the full app config. Maps key_hash -> last-persisted datetime.
+_USAGE_WRITE_INTERVAL = timedelta(minutes=15)
+_last_persisted_usage: dict[str, datetime] = {}
+_usage_lock = threading.Lock()
+
+
+def _should_persist_usage(key_hash: str, now: datetime) -> bool:
+    """Return True if `last_used_at` for the given key should be written to disk.
+
+    Thread-safe: tracks the last write per key hash in memory and only returns
+    True once per `_USAGE_WRITE_INTERVAL`.
+    """
+    with _usage_lock:
+        previous = _last_persisted_usage.get(key_hash)
+        if previous is None or (now - previous) >= _USAGE_WRITE_INTERVAL:
+            _last_persisted_usage[key_hash] = now
+            return True
+        return False
+
+
 def generate_api_key(label=""):
     """Generate a new API key.
 
@@ -67,9 +89,12 @@ def require_api_key(f):
         if not matched:
             return api_error("INVALID_KEY", "Invalid API key.", 403)
 
-        # Update last-used timestamp (best-effort)
-        matched["last_used_at"] = datetime.now(timezone.utc).isoformat()
-        config_mgr.set("api_keys", stored_keys)
+        # Update last-used timestamp (best-effort, throttled to avoid rewriting
+        # the full app config on every authenticated request).
+        now = datetime.now(timezone.utc)
+        matched["last_used_at"] = now.isoformat()
+        if _should_persist_usage(key_hash, now):
+            config_mgr.set("api_keys", stored_keys)
 
         return f(*args, **kwargs)