from typing import Optional
"""VerIQ — SQLite storage (WAL mode)."""
import json
import os
import secrets
import sqlite3
import time
from pathlib import Path
DB_PATH = Path(os.environ.get("VERIQ_DB_PATH", "/var/lib/veriq/veriq.db"))
_SCHEMA = """
CREATE TABLE IF NOT EXISTS results (
result_id TEXT PRIMARY KEY,
score INTEGER NOT NULL,
percentile REAL NOT NULL,
bank_version TEXT NOT NULL,
identity TEXT,
completed_at TEXT NOT NULL,
payload TEXT NOT NULL,
signature TEXT NOT NULL,
answers TEXT NOT NULL,
flagged INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ts REAL NOT NULL,
kind TEXT NOT NULL,
meta TEXT
);
CREATE INDEX IF NOT EXISTS idx_events_kind ON events(kind);
CREATE TABLE IF NOT EXISTS accounts (
account_id TEXT PRIMARY KEY,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS identities (
provider TEXT NOT NULL,
provider_user_id TEXT NOT NULL,
username TEXT NOT NULL,
account_id TEXT NOT NULL REFERENCES accounts(account_id),
PRIMARY KEY (provider, provider_user_id)
);
"""
def _conn() -> sqlite3.Connection:
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(DB_PATH, timeout=10)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.row_factory = sqlite3.Row
return conn
def init_db() -> None:
with _conn() as c:
c.executescript(_SCHEMA)
# migration: score_json snapshot column (added 2026-10-13, iter 4)
try:
c.execute("ALTER TABLE results ADD COLUMN score_json TEXT")
except sqlite3.OperationalError:
pass # column already exists
# migration: superseded flag (auth-before-test flow, 2026-10-13)
try:
c.execute("ALTER TABLE results ADD COLUMN superseded INTEGER DEFAULT 0")
except sqlite3.OperationalError:
pass # column already exists
# migration: account_id column (multi-identity accounts, 2026-10-13)
try:
c.execute("ALTER TABLE results ADD COLUMN account_id TEXT")
except sqlite3.OperationalError:
pass # column already exists
c.execute("CREATE INDEX IF NOT EXISTS idx_results_account"
" ON results(account_id)")
migrate_results_to_accounts(conn=c)
def new_result_id() -> str:
"""8-char base62 URL-safe id with collision check."""
alphabet = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"
with _conn() as c:
while True:
rid = "".join(secrets.choice(alphabet) for _ in range(8))
row = c.execute("SELECT 1 FROM results WHERE result_id=?", (rid,)).fetchone()
if row is None:
return rid
def save_result(result_id: str, score: int, percentile: float,
bank_version: str, identity: Optional[str], completed_at: str,
payload: str, signature: str, answers: dict,
flagged: bool = False, score_json: Optional[str] = None,
account_id: Optional[str] = None) -> None:
with _conn() as c:
c.execute(
"INSERT INTO results (result_id, score, percentile, bank_version,"
" identity, completed_at, payload, signature, answers, flagged,"
" score_json, account_id)"
" VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
(result_id, score, percentile, bank_version, identity,
completed_at, payload, signature, json.dumps(answers),
int(flagged), score_json, account_id),
)
def get_result(result_id: str) -> Optional[dict]:
with _conn() as c:
row = c.execute("SELECT * FROM results WHERE result_id=?", (result_id,)).fetchone()
if row is None:
return None
return dict(row)
def supersede_identity(identity: str) -> int:
"""Mark all previous active results of this identity as superseded."""
with _conn() as c:
cur = c.execute(
"UPDATE results SET superseded=1 WHERE identity=? AND superseded=0",
(identity,))
return cur.rowcount
def log_event(kind: str, meta: Optional[dict] = None) -> None:
with _conn() as c:
c.execute("INSERT INTO events (ts, kind, meta) VALUES (?,?,?)",
(time.time(), kind, json.dumps(meta or {})))
def event_stats() -> dict:
with _conn() as c:
rows = c.execute(
"SELECT kind, COUNT(*) AS n FROM events GROUP BY kind").fetchall()
return {r["kind"]: r["n"] for r in rows}
def stats_overview() -> dict:
"""Aggregate numbers for the /stats dashboard."""
with _conn() as c:
events = {r["kind"]: r["n"] for r in c.execute(
"SELECT kind, COUNT(*) AS n FROM events GROUP BY kind").fetchall()}
total, flagged = c.execute(
"SELECT COUNT(*), COALESCE(SUM(flagged), 0) FROM results").fetchone()
avg_iq = c.execute("SELECT AVG(score) FROM results").fetchone()[0]
return {
"events": events,
"results_total": total,
"results_flagged": flagged,
"avg_iq": round(avg_iq, 1) if avg_iq is not None else None,
}
def claim_result(result_id: str, identity: str) -> bool:
"""Bind identity to an existing result. False if result_id unknown."""
with _conn() as c:
cur = c.execute("UPDATE results SET identity=? WHERE result_id=?",
(identity, result_id))
return cur.rowcount > 0
# --- accounts & identities (multi-identity accounts, 2026-10-13) ---------------
def new_account_id() -> str:
return "a" + secrets.token_hex(8)
def create_account(conn: Optional[sqlite3.Connection] = None) -> str:
"""Create an account row, return its id."""
aid = new_account_id()
sql = "INSERT INTO accounts (account_id, created_at) VALUES (?,?)"
args = (aid, time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
if conn is not None:
conn.execute(sql, args)
return aid
with _conn() as c:
c.execute(sql, args)
return aid
def get_identity(provider: str, provider_user_id: str) -> Optional[dict]:
with _conn() as c:
row = c.execute(
"SELECT * FROM identities WHERE provider=? AND provider_user_id=?",
(provider, provider_user_id)).fetchone()
return dict(row) if row else None
def create_identity(provider: str, provider_user_id: str, username: str,
account_id: str,
conn: Optional[sqlite3.Connection] = None) -> None:
sql = ("INSERT INTO identities (provider, provider_user_id, username,"
" account_id) VALUES (?,?,?,?)")
args = (provider, provider_user_id, username, account_id)
if conn is not None:
conn.execute(sql, args)
return
with _conn() as c:
c.execute(sql, args)
def update_identity_username(provider: str, provider_user_id: str,
username: str) -> None:
with _conn() as c:
c.execute("UPDATE identities SET username=? WHERE provider=?"
" AND provider_user_id=?",
(username, provider, provider_user_id))
def account_identities(account_id: str) -> list:
with _conn() as c:
rows = c.execute(
"SELECT provider, username FROM identities WHERE account_id=?"
" ORDER BY provider", (account_id,)).fetchall()
return [dict(r) for r in rows]
def _migrate_results_to_accounts(c: sqlite3.Connection) -> int:
rows = c.execute(
"SELECT DISTINCT identity FROM results"
" WHERE identity IS NOT NULL AND identity != ''"
" AND account_id IS NULL").fetchall()
n = 0
for r in rows:
identity = r["identity"]
if ":" in identity:
provider, uname = identity.split(":", 1)
uname = uname.lstrip("@")
else:
provider, uname = "x", identity
pid = identity # stable unique key (real provider_user_id unknown)
row = c.execute(
"SELECT account_id FROM identities"
" WHERE provider=? AND provider_user_id=?",
(provider, pid)).fetchone()
if row:
account_id = row["account_id"]
else:
account_id = new_account_id()
c.execute("INSERT INTO accounts (account_id, created_at)"
" VALUES (?,?)",
(account_id,
time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())))
c.execute("INSERT INTO identities (provider, provider_user_id,"
" username, account_id) VALUES (?,?,?,?)",
(provider, pid, uname, account_id))
c.execute("UPDATE results SET account_id=? WHERE identity=?"
" AND account_id IS NULL", (account_id, identity))
n += 1
return n
def migrate_results_to_accounts(conn: Optional[sqlite3.Connection] = None) -> int:
"""One-time migration: give every pre-accounts result an account.
Old rows only carry identity like "x:@user"; provider_user_id is unknown,
so the full identity string is used as a stable unique key.
"""
if conn is not None:
return _migrate_results_to_accounts(conn)
with _conn() as c:
n = _migrate_results_to_accounts(c)
c.commit()
return n
def supersede_account(account_id: str) -> int:
"""Mark all previous active results of this account as superseded."""
with _conn() as c:
cur = c.execute(
"UPDATE results SET superseded=1 WHERE account_id=?"
" AND superseded=0", (account_id,))
return cur.rowcount
def active_result_for_account(account_id: str) -> Optional[dict]:
with _conn() as c:
row = c.execute(
"SELECT * FROM results WHERE account_id=? AND superseded=0"
" ORDER BY completed_at DESC LIMIT 1", (account_id,)).fetchone()
return dict(row) if row else None
