Spaces:

tommytracx
/

thox-openwebui

Running

fix: take sqlite off the bucket - it is the mount, not the tuning
df99538 2 months ago
45.3 kB
#!/usr/bin/env python3
"""THOX Open WebUI state sync: HF Dataset <-> /app/backend/data.
WHY THIS EXISTS
    HF Spaces have ephemeral storage. The paid persistent-storage tier is not
    available on this account (whoami-v2 reports canPay:false), so a private HF
    Dataset is the backing store instead.
HOT / COLD SPLIT
    webui.db          ~30MB, changes constantly -> snapshot+upload every cycle
    assets.tar.zst    ~1.2GB, changes rarely    -> upload only when the hash moves
    vector_db/ is ~4250 Chroma collection directories. Synced as loose files that
    is 4250 requests per pass, so it travels as a single tar.
CONSISTENCY
    webui.db is live SQLite in WAL mode. A plain copy while Open WebUI is writing
    yields a torn file, so backups go through `VACUUM INTO`, which takes an atomic
    snapshot against concurrent writers.
WHAT THIS DOES NOT GUARANTEE
    Writes made since the last successful backup are lost on an unexpected
    restart. This is weaker than a real volume; it is the ceiling on the free tier.
"""
import hashlib, json, os, pathlib, re, shutil, sqlite3, subprocess, sys, time

DATA   = pathlib.Path(os.environ.get("DATA_DIR", "/app/backend/data"))
REPO   = os.environ.get("STATE_REPO", "tommytracx/thox-openwebui-state")
TOKEN  = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
INTERVAL = int(os.environ.get("BACKUP_INTERVAL", "300"))
WORK   = pathlib.Path("/tmp/thox-state"); WORK.mkdir(parents=True, exist_ok=True)
ASSETS = ["uploads", "vector_db"]


def log(msg):
    print(f"[thox-state] {msg}", flush=True)


def _api():
    from huggingface_hub import HfApi
    return HfApi(token=TOKEN)


def _sha(path, chunk=1 << 20):
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for b in iter(lambda: fh.read(chunk), b""):
            h.update(b)
    return h.hexdigest()


def _marker(name):
    return WORK / f"{name}.sha256"


# ------------------------------------------------------- integrity / repair
# ROOT CAUSE, 2026-08-26. The Space looped "Restarting" and kept booting into
# "DATABASE IS CORRUPT". The corruption was a SYMPTOM, not the cause:
#
#   1. open-webui's scheduler polls timers once a SECOND with
#          WHERE JSON_EXTRACT(chat.meta,'$."internal"') IS 1 AND ...
#      which is an UNINDEXED scan over every chat row.
#   2. sqlite raises "malformed JSON" if a SINGLE scanned value is not valid
#      JSON, so one bad chat.meta takes down the whole scheduler -- once per
#      second, each time printing a ~100-line traceback (measured: repeating
#      at 18:43:48, :49, :50, :51, ... within a single boot).
#   3. That write+log pressure keeps the WAL permanently hot, so the checkpoint
#      loop can never TRUNCATE it. The log says exactly that:
#          [thox-state] checkpoint busy (wal=483 pages) - will retry
#   4. The bucket mount is not write-through, so a restart then races an
#      unflushed 483-page WAL and lands a torn file -> quick_check fails ->
#      "DATABASE IS CORRUPT" -> restore from the Dataset -> back to step 1.
#
# Restoring from the Dataset could NEVER break that cycle, because the poison
# is a ROW VALUE and every VACUUM INTO snapshot preserves it faithfully.
# Repairing the JSON is the fix; the rest of this section is damage control.
#
# Upstream has since reworked that query to filter on an indexed timer_at
# column first, which bounds the scan to real timers. The image pins
# open-webui:main, so a rebuild picks that up -- but the repair stays, because
# it costs nothing and does not depend on upstream's release schedule.


def _json_columns(con):
    """Discover (table, column, empty) for every column that actually holds JSON.
    Discovered rather than hard-coded: open-webui adds JSON columns across
    releases, and a repair that only knew about chat.meta would silently miss
    whichever column breaks next. `empty` preserves the column's shape, so a
    repaired array column does not come back as an object.
    """
    out = []
    tables = [r[0] for r in con.execute(
        "select name from sqlite_master where type='table' and name not like 'sqlite_%'")]
    for t in tables:
        try:
            cols = con.execute('PRAGMA table_info("{}")'.format(t)).fetchall()
        except sqlite3.Error:
            continue
        for row in cols:
            c = row[1]
            q = ('select json_type("{c}") from "{t}" where "{c}" is not null '
                 'and json_valid("{c}") '
                 'and json_type("{c}") in (\'object\',\'array\') limit 1')
            try:
                shape = con.execute(q.format(t=t, c=c)).fetchone()
            except sqlite3.Error:
                continue                       # not text, or unusable here
            if shape:
                out.append((t, c, "[]" if shape[0] == "array" else "{}"))
    return out


def json_repair(db_path, columns=None):
    """Neutralise every value JSON_EXTRACT() would reject. Returns rows fixed.
    json_valid(x)=0 is EXACTLY the set json_extract() raises on -- verified
    against sqlite 3.53: the empty string, whitespace, bare words, truncated
    objects and blobs all report 0 and all raise. SQL NULL reports NULL (not 0)
    and is deliberately left alone, because json_extract(NULL, ...) returns
    NULL rather than raising.
    """
    con = sqlite3.connect(str(db_path), timeout=30)
    fixed = 0
    try:
        cols = columns if columns is not None else _json_columns(con)
        for t, c, empty in cols:
            sel = ('select count(*) from "{t}" where "{c}" is not null '
                   'and json_valid("{c}")=0')
            ids = ('select rowid from "{t}" where "{c}" is not null '
                   'and json_valid("{c}")=0 limit 20')
            upd = ('update "{t}" set "{c}"=? where "{c}" is not null '
                   'and json_valid("{c}")=0')
            try:
                bad = con.execute(sel.format(t=t, c=c)).fetchone()[0]
                if not bad:
                    continue
                # Log identity BEFORE overwriting: without this, the next
                # occurrence is exactly as blind as this one was.
                for (rid,) in con.execute(ids.format(t=t, c=c)):
                    log("REPAIR {}.{} rowid={} held invalid JSON -> {}".format(
                        t, c, rid, empty))
                con.execute(upd.format(t=t, c=c), (empty,))
                fixed += bad
            except sqlite3.Error as exc:
                log("repair skipped {}.{}: {}".format(t, c, exc))
        if fixed:
            con.commit()
            log("json repair: neutralised {} invalid value(s)".format(fixed))
    finally:
        con.close()
    return fixed


def integrity(db_path):
    """True if the database is both structurally sound AND actually readable.
    TWO checks, because on 2026-08-26 the pragma alone was not enough. Boot
    logged `existing database on the bucket - no seed`, and six minutes later the
    running app was failing every ~11s with `database disk image is malformed` on
    the automations query. A gate that clears a database the app cannot read is
    worse than no gate, because it routes around the recovery path.
      1. PRAGMA integrity_check, not quick_check. quick_check skips per-page
         validation and the index/table consistency checks, so it is a strict
         subset. The performance argument for it does not survive measurement:
         on the real 106.4 MB database the two came out at 0.12s and 0.13s, so
         there is nothing to trade off. (Note: 14 random page-corruption trials
         did NOT produce a case where quick_check passed and integrity_check
         failed, so this is chosen for being a superset at equal cost -- it is
         not evidence about which one missed the live fault.)
      2. A smoke read of the tables open-webui touches on every poll. This is
         the check that would definitely have caught the incident, because it
         fails the same way the app fails: by reading the same pages. A pragma
         reasons about structure; only a real query proves a real query works.
    """
    pragma = os.environ.get("INTEGRITY_PRAGMA", "integrity_check")
    if pragma not in ("integrity_check", "quick_check"):
        pragma = "integrity_check"
    try:
        con = sqlite3.connect("file:{}?mode=ro".format(db_path), uri=True, timeout=120)
    except Exception as exc:
        log("integrity: cannot open ({}: {})".format(type(exc).__name__, exc))
        return False
    try:
        try:
            verdict = con.execute("PRAGMA {}".format(pragma)).fetchone()[0]
        except Exception as exc:
            log("integrity: {} could not run ({}: {})".format(
                pragma, type(exc).__name__, exc))
            return False
        if verdict != "ok":
            log("{} FAILED: {}".format(pragma, str(verdict)[:300]))
            return False

        # Tables the scheduler and the UI read constantly. A table that is not
        # present yet is fine (fresh database, migrations pending); a table that
        # is present and unreadable is not.
        present = {r[0] for r in con.execute(
            "select name from sqlite_master where type='table'")}
        for table in ("chat", "automation", "chat_message", "config", "user", "folder"):
            if table not in present:
                continue
            try:
                con.execute('select count(*) from "{}"'.format(table)).fetchone()
            except Exception as exc:
                log("integrity: {} is unreadable ({}: {}) - treating as corrupt".format(
                    table, type(exc).__name__, exc))
                return False
        return True
    finally:
        con.close()


def _logical_salvage(src, dst):
    """Rebuild a database by re-INSERTing every row that can still be read.
    NOT Connection.backup(): that is a PAGE-level copy, so it reproduces the
    damaged pages verbatim and the result fails quick_check exactly like the
    input -- measured, which is why this exists. Rebuilding through SQL means
    the output is written from scratch by sqlite itself and is structurally
    sound by construction; only unreadable ROWS are lost.
    """
    s = sqlite3.connect("file:{}?mode=ro".format(src), uri=True, timeout=30)
    d = sqlite3.connect(str(dst))
    kept = dropped = 0
    rebuilt, failed = [], []
    try:
        schema = s.execute(
            "select type,name,sql from sqlite_master where sql is not null").fetchall()
        tables = [(n, q) for t, n, q in schema if t == "table" and not n.startswith("sqlite_")]
        rest = [(n, q) for t, n, q in schema if t in ("index", "view", "trigger")
                and not n.startswith("sqlite_")]

        for name, ddl in tables:
            try:
                d.execute(ddl)
            except sqlite3.Error as exc:
                log("salvage: cannot create {}: {}".format(name, exc))
                failed.append(name)
                continue
            rebuilt.append(name)
            ncols = len(s.execute('PRAGMA table_info("{}")'.format(name)).fetchall())
            ins = 'insert or ignore into "{}" values ({})'.format(
                name, ",".join(["?"] * ncols))
            try:
                rows = s.execute('select * from "{}"'.format(name)).fetchall()
            except sqlite3.Error:
                # A damaged page killed the bulk read. Walk the table in small
                # rowid windows so one bad page costs a window, not the table.
                rows = []
                try:
                    hi = s.execute('select max(rowid) from "{}"'.format(name)).fetchone()[0] or 0
                except sqlite3.Error:
                    hi = 0
                step = 200
                for lo in range(0, hi + 1, step):
                    try:
                        rows.extend(s.execute(
                            'select * from "{}" where rowid>=? and rowid<?'.format(name),
                            (lo, lo + step)).fetchall())
                    except sqlite3.Error:
                        dropped += step
            for r in rows:
                try:
                    d.execute(ins, r)
                    kept += 1
                except sqlite3.Error:
                    dropped += 1
        d.commit()
        for name, ddl in rest:
            try:
                d.execute(ddl)
            except sqlite3.Error:
                pass                      # an index we cannot rebuild is not fatal
        d.commit()
    finally:
        s.close()
        d.close()
    log("logical salvage kept {} row(s) across {} table(s), lost ~{}".format(
        kept, len(rebuilt), dropped))
    if failed:
        log("logical salvage COULD NOT REBUILD: {}".format(", ".join(failed)))
    return kept > 0


# Tables open-webui cannot start without. A salvage that dropped one of these is
# not a recovery, however clean the file it produced looks: on 2026-08-26 19:46
# the logical salvage returned a structurally PERFECT database that passed
# integrity_check and every readability probe, and the app died on
#     sqlite3.OperationalError: no such table: config
# integrity() tolerates ABSENT tables on purpose -- a fresh database has none and
# alembic is about to create them -- so absence has to be judged separately, and
# only where we are validating a replacement rather than a first boot.
_CORE_TABLES = ("config", "user", "auth", "chat", "alembic_version")


def servable(db_path):
    """True if this file is sound AND complete enough for open-webui to boot."""
    if not integrity(db_path):
        return False
    try:
        con = sqlite3.connect("file:{}?mode=ro".format(db_path), uri=True, timeout=60)
    except Exception:
        return False
    try:
        present = {r[0] for r in con.execute(
            "select name from sqlite_master where type='table'")}
    finally:
        con.close()
    missing = [t for t in _CORE_TABLES if t not in present]
    if missing:
        log("NOT SERVABLE: {} is missing core table(s): {}".format(
            db_path, ", ".join(missing)))
        return False
    return True


def populated(db_path):
    """True if this is OUR database, not a freshly-migrated blank one.
    servable() is not enough here and the distinction cost an outage. After the
    2026-08-26 19:50 boot served a blank 4096-byte database, ALEMBIC BUILT THE
    WHOLE SCHEMA into it -- config, user, auth, chat, alembic_version all present
    and all EMPTY. So the file now passes every structural test there is while
    containing none of Tommy's data, and a presence check would happily serve it
    forever.
    `user` is the marker: open-webui cannot be used at all without an account, so
    a database with zero users is either brand new or has lost everything. On
    this Space the Dataset is the source of truth, so either way the right move
    is to restore rather than serve a blank site.
    """
    if not servable(db_path):
        return False
    try:
        con = sqlite3.connect("file:{}?mode=ro".format(db_path), uri=True, timeout=60)
    except Exception:
        return False
    try:
        users = con.execute("select count(*) from user").fetchone()[0]
        chats = con.execute("select count(*) from chat").fetchone()[0]
    except sqlite3.Error as exc:
        log("populated: cannot count ({}) - treating as unpopulated".format(exc))
        return False
    finally:
        con.close()
    if users == 0:
        log("UNPOPULATED: {} has a complete schema but 0 users ({} chats) - "
            "this is a blank database, not ours".format(db_path, chats))
        return False
    return True


def recover(src, dst):
    """Salvage a damaged database into a fresh one. True if dst came out usable.
    Prefers sqlite's own .recover, which walks the b-tree pages directly and
    rebuilds from whatever is still readable -- strictly more than SQL can
    reach. The Dockerfile installs the sqlite3 CLI for exactly this; the
    logical fallback keeps the path working if it is ever missing.
    """
    src, dst = str(src), str(dst)
    pathlib.Path(dst).unlink(missing_ok=True)
    if shutil.which("sqlite3"):
        try:
            dump = subprocess.run(["sqlite3", src, ".recover"],
                                  capture_output=True, timeout=900)
            if dump.returncode == 0 and dump.stdout:
                load = subprocess.run(["sqlite3", dst], input=dump.stdout,
                                      capture_output=True, timeout=900)
                # .recover output routinely contains statements that fail to
                # replay; a non-zero exit does not mean nothing landed. Judge
                # the RESULT, not the exit code.
                if integrity(dst):
                    log(".recover salvaged {} (loader rc={})".format(src, load.returncode))
                    return True
        except Exception as exc:
            log(".recover failed ({}: {}) - trying logical salvage".format(
                type(exc).__name__, exc))
    else:
        log("sqlite3 CLI absent - using logical salvage")
    pathlib.Path(dst).unlink(missing_ok=True)
    try:
        if _logical_salvage(src, dst) and integrity(dst):
            return True
    except Exception as exc:
        log("logical salvage failed: {}: {}".format(type(exc).__name__, exc))
    pathlib.Path(dst).unlink(missing_ok=True)
    return False


# --------------------------------------------------------- schema reconcile
# A stamped alembic revision is NOT proof the DDL landed. On SQLite alembic runs
# with "Will assume non-transactional DDL", and a revision it has stamped never
# runs again. Observed 2026-08-26 19:04 on this Space: the log says
#
#     Running upgrade 6d09d1bf1f23 -> d4c1a8e37b62, add chat timer_at and ...
#
# and the app then failed once a SECOND with `no such column: chat.timer_at`
# (391 scheduler errors / 47555 log lines in seven minutes). The most likely
# mechanism is this repo's own checkpoint loop: a SECOND process issuing
# wal_checkpoint(TRUNCATE) against a bucket-mounted database while alembic was
# mid-migration. checkpoint_loop now waits for the backend before its first
# cycle so it cannot overlap DDL again -- this is the backstop for the damage
# already done, and for any future repeat.
#
# Deliberately NOT a schema manager. It reconciles exactly the divergences we
# have actually observed, and defers to alembic whenever alembic still has the
# migration pending.

# revision -> columns that revision is responsible for adding.
# `pending` lists the revisions from which alembic will still RUN that
# migration; if the database is stamped at one of those, we must keep our hands
# off and let alembic do its job.
_EXPECTED_SCHEMA = [
    {
        "revision": "d4c1a8e37b62",
        "pending": {"f0bd01a18a3d", "1ce6ade7d93b", "6d09d1bf1f23"},
        "table": "chat",
        "column": "timer_at",
        "ddl": 'ALTER TABLE "chat" ADD COLUMN "timer_at" BIGINT',
        "extra": ['CREATE INDEX IF NOT EXISTS "timer_at_idx" ON "chat" ("timer_at") '
                  "WHERE timer_at IS NOT NULL"],
    },
]


def reconcile_schema(db_path=None):
    """Add columns a stamped-but-unapplied migration owes us. Returns count applied.
    ALTER TABLE ADD COLUMN with no default is O(1) in sqlite and safe against a
    running app: existing rows read NULL, which is exactly what the migration
    would have produced.
    """
    db = pathlib.Path(db_path or (DATA / "webui.db"))
    if not db.exists():
        return 0
    applied = 0
    try:
        con = sqlite3.connect(str(db), timeout=60)
    except sqlite3.Error as exc:
        log("reconcile: cannot open database: {}".format(exc))
        return 0
    try:
        try:
            row = con.execute("select version_num from alembic_version").fetchone()
        except sqlite3.Error:
            return 0                      # not an open-webui database yet
        stamped = row[0] if row else None
        for want in _EXPECTED_SCHEMA:
            if stamped in want["pending"]:
                # alembic has not reached this revision yet; it will add the
                # column itself, and pre-creating it would make op.add_column
                # fail with "duplicate column name" and abort the migration.
                continue
            try:
                cols = [r[1] for r in con.execute(
                    'PRAGMA table_info("{}")'.format(want["table"])).fetchall()]
            except sqlite3.Error:
                continue
            if not cols or want["column"] in cols:
                continue
            log("RECONCILE {}.{} is missing though alembic is stamped at {} "
                "(revision {} should have added it) - applying".format(
                    want["table"], want["column"], stamped, want["revision"]))
            try:
                con.execute(want["ddl"])
                for extra in want["extra"]:
                    try:
                        con.execute(extra)
                    except sqlite3.Error as exc:
                        log("reconcile: index skipped: {}".format(exc))
                con.commit()
                applied += 1
            except sqlite3.Error as exc:
                log("reconcile: {}.{} failed: {}".format(
                    want["table"], want["column"], exc))
    finally:
        con.close()
    return applied


def _wait_for_backend(timeout=900):
    """Block until open-webui answers, so nothing writes during its migrations.
    The checkpoint loop is a SECOND writer against a database on a mount whose
    locking cannot be relied on. Overlapping it with alembic's DDL is how a
    migration gets stamped without landing. Waiting costs one checkpoint cycle
    and removes the whole race.
    """
    import urllib.request
    url = "http://127.0.0.1:{}/health".format(os.environ.get("PORT", "8080"))
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(url, timeout=5) as r:
                if r.status == 200:
                    log("backend is up - checkpoint loop may now write")
                    return True
        except Exception:
            pass
        time.sleep(5)
    log("backend did not answer in {}s - starting checkpoint loop anyway".format(timeout))
    return False


def disable_automations(db_path=None):
    """Set automations.enable=false in the config table. Returns True if it stuck.
    WHY THIS IS A BOOT STEP AND NOT A SETTING TOMMY TOGGLES: he cannot reach the
    setting. The 122 restored automations start firing the moment the app is up,
    which on this bucket mount corrupts the database within minutes, and Open
    WebUI answers a failed user lookup with the SAME message as a wrong password
    -- "The email or password provided is incorrect" -- so the corruption locks
    him out of the UI he would use to turn them off.
    Written straight to the config table, like seed_connections/seed_stt: these
    are PersistentConfig values, so an environment variable seeds only a FIRST
    boot and this Space always has a database.
    Fully reversible: Settings -> Automations, or drop THOX_DISABLE_AUTOMATIONS
    and redeploy. Nothing is deleted -- all 122 automations are still there.
    """
    db = pathlib.Path(db_path or (DATA / "webui.db"))
    if not db.exists():
        log("automations: no database yet - nothing to disable")
        return False
    try:
        con = sqlite3.connect(str(db), timeout=60)
    except sqlite3.Error as exc:
        log("automations: cannot open database ({})".format(exc))
        return False
    try:
        row = con.execute(
            "select value from config where key='automations.enable'").fetchone()
        if row is not None and str(row[0]).strip().lower() in ("false", '"false"'):
            log("automations: already disabled")
            return True
        if row is None:
            cols = [c[1] for c in con.execute('PRAGMA table_info("config")').fetchall()]
            if "updated_at" in cols:
                con.execute("insert into config(key,value,updated_at) "
                            "values('automations.enable','false',strftime('%s','now'))")
            else:
                con.execute("insert into config(key,value) "
                            "values('automations.enable','false')")
        else:
            con.execute("update config set value='false' "
                        "where key='automations.enable'")
        con.commit()
        log("DISABLED the automation scheduler (automations.enable=false) - "
            "122 automations are retained, re-enable in Settings > Automations")
        return True
    except sqlite3.Error as exc:
        log("automations: could not disable ({}) - continuing".format(exc))
        return False
    finally:
        con.close()


def prepare(db_path=None):
    """Make the live database safe to serve, immediately before open-webui opens it.
    Three things the app cannot do for itself:
      1. Neutralise invalid JSON. open-webui's scheduler scans chat.meta once a
         second and dies on the first value json_extract() rejects, so a single
         poisoned row must never survive into serving.
      2. Put the file in WAL mode. VACUUM INTO writes its output in DELETE mode,
         so a database that arrived via restore() starts non-WAL and the
         checkpoint loop has nothing to truncate until open-webui gets round to
         switching it.
      3. Truncate whatever WAL is already on disk, so the first checkpoint cycle
         does not inherit a backlog from the boot that just died.
    Never raises: a boot must always reach `exec start.sh`. An unserved Space is
    a worse outcome than an unrepaired one.
    """
    db = pathlib.Path(db_path or (DATA / "webui.db"))
    if not db.exists():
        log("prepare: no database yet - open-webui will create one")
        return 0
    fixed = 0
    try:
        fixed = json_repair(db)
    except Exception as exc:
        log("prepare: repair failed ({}: {}) - continuing".format(type(exc).__name__, exc))
    # Safe to run before alembic: reconcile_schema defers whenever alembic still
    # has the migration pending, so it can only ever ADD a column that alembic
    # has already claimed to have added and did not.
    try:
        reconcile_schema(db)
    except Exception as exc:
        log("prepare: reconcile failed ({}: {}) - continuing".format(type(exc).__name__, exc))
    try:
        con = sqlite3.connect(str(db), timeout=60)
        try:
            # DELETE, not WAL. WAL depends on a coherent -shm shared-memory
            # file, which a FUSE/object-storage mount does not provide; that is
            # what kept corrupting this database. Must agree with
            # DATABASE_ENABLE_SQLITE_WAL in the Dockerfile, or open-webui simply
            # switches it back on its first connection.
            want = os.environ.get("THOX_JOURNAL_MODE", "DELETE")
            mode = con.execute("PRAGMA journal_mode={}".format(want)).fetchone()[0]
            if mode.lower() == "wal":
                con.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone()
            log("prepare: journal_mode={}, WAL truncated, {} value(s) repaired".format(
                mode, fixed))
        finally:
            con.close()
    except Exception as exc:
        log("prepare: WAL setup failed ({}: {}) - continuing".format(
            type(exc).__name__, exc))
    return fixed


def install(candidate, target):
    """Atomically move candidate onto target, clearing stale WAL sidecars.
    os.replace() is the entire point. The previous code did shutil.copy2()
    STRAIGHT ONTO the live path, so a container killed mid-copy -- or a full
    bucket -- left a half-written webui.db that the next boot correctly judged
    corrupt: a restore that could manufacture the very corruption it was sent
    to repair. A rename is atomic, so the file is either the old one or the new
    one and never a prefix of either.
    """
    target = pathlib.Path(target)
    os.replace(str(candidate), str(target))
    for sidecar in ("-wal", "-shm"):
        pathlib.Path(str(target) + sidecar).unlink(missing_ok=True)


# ---------------------------------------------------------------- restore
def restore():
    """Install the Dataset snapshot at DATA/webui.db, verifying before it lands.
    ORDER MATTERS: download -> integrity-check -> salvage if needed -> repair
    JSON -> atomic rename. The old flow copied the download straight onto the
    live path and only then counted chats, so a bad snapshot became the live
    database and a killed copy left a truncated one. Nothing is published here
    until it has passed quick_check.
    """
    if not TOKEN:
        log("no HF_TOKEN -> starting with an empty database (nothing restored)")
        return False
    from huggingface_hub import hf_hub_download
    DATA.mkdir(parents=True, exist_ok=True)
    ok = False

    try:
        p = hf_hub_download(REPO, "webui.db", repo_type="dataset", token=TOKEN,
                            local_dir=str(WORK / "dl"))
        # Stage inside DATA so the final os.replace() is a same-filesystem
        # rename. Staging in /tmp would make it a cross-device copy, which is
        # exactly the non-atomic step this is here to avoid.
        staged = DATA / "webui.db.incoming"
        staged.unlink(missing_ok=True)
        shutil.copy2(p, staged)

        if not integrity(staged):
            # The snapshot itself is damaged. Blindly installing it is how a
            # restore turns one bad boot into a permanent loop.
            log("SNAPSHOT FAILS quick_check - attempting .recover before install")
            salvaged = DATA / "webui.db.recovered"
            if recover(staged, salvaged):
                staged.unlink(missing_ok=True)
                staged = salvaged
            else:
                staged.unlink(missing_ok=True)
                salvaged.unlink(missing_ok=True)
                raise RuntimeError("snapshot is corrupt and unrecoverable")

        json_repair(staged)
        n = sqlite3.connect(str(staged)).execute(
            "select count(*) from chat").fetchone()[0]
        size = staged.stat().st_size
        install(staged, DATA / "webui.db")
        _marker("webui.db").write_text(_sha(DATA / "webui.db"))
        log(f"restored webui.db ({size/1e6:.1f} MB, {n} chats)")
        ok = True
    except Exception as exc:                     # first boot, or repo still empty
        log(f"no webui.db restored ({type(exc).__name__}: {exc}) -> fresh database")

    # Assets live on the bucket via symlink now, so they survive a restart and do
    # not need re-extracting. Only pull the 1.2 GB tar when they are genuinely
    # absent, or every boot pays five minutes to rewrite files it already has.
    try:
        have = any((DATA / a).is_dir() and any((DATA / a).iterdir()) for a in ASSETS)
    except Exception:
        have = False
    if have:
        log("assets already present (bucket-backed) - not re-extracting")
        return True

    try:
        p = hf_hub_download(REPO, "assets.tar.zst", repo_type="dataset", token=TOKEN,
                            local_dir=str(WORK / "dl"))
        subprocess.run(["tar", "--use-compress-program=unzstd", "-xf", p, "-C", str(DATA)],
                       check=True)
        _marker("assets").write_text(_sha(p))
        log(f"restored assets.tar.zst -> {', '.join(ASSETS)}")
        ok = True
    except Exception as exc:
        log(f"no assets restored ({type(exc).__name__}) -> knowledge bases will be empty")

    return ok


# ---------------------------------------------------------------- backup
# Credential shapes that must never reach the state Dataset.
_SECRET_RX = re.compile("|".join([
    r"sk-proj-[A-Za-z0-9_-]{20,}",
    r"sk-ant-[A-Za-z0-9_-]{20,}",
    r"sk-[A-Za-z0-9]{32,}",
    r"hf_[A-Za-z0-9]{30,}",
    r"nvapi-[A-Za-z0-9_-]{20,}",
    r"AIza[A-Za-z0-9_-]{30,}",
    r"gsk_[A-Za-z0-9]{40,}",
    r"[0-9a-f]{32}[.][A-Za-z0-9_-]{16,}",
]))


def _scrub(db_path):
    """Blank every credential in `config` before the snapshot leaves the Space.
    seed_connections.py writes HF_TOKEN into openai.api_keys so Open WebUI can
    reach the private model Spaces. Without this, the backup loop would copy that
    token -- and any provider key a user has entered -- straight into the Dataset.
    Seeding is idempotent and re-runs on every boot, so connection keys never need
    to be persisted; they are rebuilt from the Space secret instead.
    """
    con = sqlite3.connect(db_path)
    try:
        n = 0
        # 1. Structural pass. Every element of these arrays IS a credential by
        #    definition of the field, so blank them wholesale. Pattern matching
        #    alone is a denylist and misses any provider whose key prefix is not
        #    enumerated below (xai-, vendor-specific, THOX-internal, ...).
        for key in ("openai.api_keys", "ollama.api_keys"):
            try:
                row = con.execute("select value from config where key=?", (key,)).fetchone()
            except sqlite3.Error:
                return                    # no config table: nothing to scrub
            if not row:
                continue
            try:
                arr = json.loads(row[0])
            except (TypeError, json.JSONDecodeError):
                continue
            if isinstance(arr, list) and any(arr):
                con.execute("update config set value=? where key=?",
                            (json.dumps([""] * len(arr)), key))
                n += 1

        # 2. Pattern pass, for credentials embedded in other config values
        #    (audio.*, image_generation.*, tool servers, nested api_configs).
        for k, v in con.execute("select key,value from config").fetchall():
            s = str(v)
            if _SECRET_RX.search(s):
                con.execute("update config set value=? where key=?",
                            (_SECRET_RX.sub("", s), k))
                n += 1
        if n:
            con.commit()
            log(f"scrubbed credentials from {n} config key(s) before upload")
    finally:
        con.close()


def _snapshot_db():
    """Atomic, self-consistent copy of a live WAL database, with secrets removed."""
    out = WORK / "webui-snap.db"
    out.unlink(missing_ok=True)
    con = sqlite3.connect(f"file:{DATA/'webui.db'}?mode=ro", uri=True)
    try:
        con.execute("VACUUM INTO ?", (str(out),))
    finally:
        con.close()
    _scrub(out)
    # Repair on the way OUT as well as on the way in. The 2026-08-26 incident
    # showed the snapshot is not just a backup, it is the thing a corrupt boot
    # restores FROM -- so a poisoned value that reaches the Dataset is a poisoned
    # value every future recovery faithfully reinstalls. Cleaning the snapshot
    # here means the archive can never re-seed the fault it was meant to survive.
    json_repair(out)
    return out


def _remote_size(path_in_repo="webui.db"):
    """Bytes of the file currently in the Dataset, or None if unknown."""
    try:
        info = _api().get_paths_info(REPO, [path_in_repo], repo_type="dataset")
        for f in info:
            if getattr(f, "size", None):
                return f.size
    except Exception as exc:
        log("could not size the remote snapshot ({})".format(type(exc).__name__))
    return None


def backup_db():
    if not TOKEN or not (DATA / "webui.db").exists():
        return False
    # Check the LIVE database before snapshotting it. _snapshot_db() scrubs
    # credentials out of `config`, so on a blank database it raises `no such
    # table: config` and the archive loop merely logs "will retry next cycle".
    # That accident is not a safety mechanism -- refuse deliberately, and say so.
    if not populated(DATA / "webui.db"):
        log("REFUSING TO ARCHIVE: the live database is blank or unservable - "
            "keeping the remote copy untouched")
        return False
    snap = _snapshot_db()

    # THE ARCHIVE IS THE LAST COPY. Do not let it be overwritten by a database
    # that is empty or gutted. On 2026-08-26 a blank 4096-byte webui.db was
    # served for ~an hour; had the archive fired, it would have replaced the
    # only good snapshot -- 417 chats -- with nothing, turning a recoverable
    # incident into permanent loss. A backup that can destroy the thing it is
    # backing up is worse than no backup.
    if not servable(snap):
        log("REFUSING TO ARCHIVE: snapshot is not servable - keeping the remote copy")
        return False
    # Close it explicitly: this runs every archive cycle for the life of the
    # container, and an unclosed handle per cycle is a leak (and on Windows it
    # blocks the next cycle from replacing the snapshot file at all).
    _c = sqlite3.connect(str(snap))
    try:
        chats = _c.execute("select count(*) from chat").fetchone()[0]
    finally:
        _c.close()
    if chats == 0:
        log("REFUSING TO ARCHIVE: snapshot has 0 chats - keeping the remote copy")
        return False
    remote = _remote_size()
    local = snap.stat().st_size
    shrink_ok = os.environ.get("ARCHIVE_ALLOW_SHRINK") == "1"
    if remote and local < remote * 0.5 and not shrink_ok:
        log("REFUSING TO ARCHIVE: snapshot is {:.1f} MB vs {:.1f} MB in the Dataset "
            "(>50% smaller). Set ARCHIVE_ALLOW_SHRINK=1 if this is intended."
            .format(local / 1e6, remote / 1e6))
        return False

    digest = _sha(snap)
    if _marker("webui.db").exists() and _marker("webui.db").read_text().strip() == digest:
        return False                              # unchanged since last push
    _api().upload_file(path_or_fileobj=str(snap), path_in_repo="webui.db",
                       repo_id=REPO, repo_type="dataset",
                       commit_message=f"state: webui.db {digest[:12]}")
    _marker("webui.db").write_text(digest)
    log(f"backed up webui.db ({snap.stat().st_size/1e6:.1f} MB, {digest[:12]})")
    return True


def backup_assets():
    if not TOKEN:
        return False
    present = [a for a in ASSETS if (DATA / a).exists()]
    if not present:
        return False
    tar = WORK / "assets.tar.zst"
    tar.unlink(missing_ok=True)
    subprocess.run(f"tar -C {DATA} -cf - {' '.join(present)} | zstd -3 -T0 -q -o {tar}",
                   shell=True, check=True)
    digest = _sha(tar)
    if _marker("assets").exists() and _marker("assets").read_text().strip() == digest:
        tar.unlink(missing_ok=True)
        return False
    _api().upload_file(path_or_fileobj=str(tar), path_in_repo="assets.tar.zst",
                       repo_id=REPO, repo_type="dataset",
                       commit_message=f"state: assets {digest[:12]}")
    _marker("assets").write_text(digest)
    log(f"backed up assets.tar.zst ({tar.stat().st_size/1e6:.0f} MB, {digest[:12]})")
    tar.unlink(missing_ok=True)
    return True


def checkpoint_loop():
    """Fold the WAL into webui.db periodically so the BUCKET copy stays current.
    WHY THIS IS NECESSARY (measured 2026-08-22):
        The bucket mount is NOT write-through. Writes accumulate in webui.db-wal
        inside the container and only reach the bucket around shutdown. That
        creates a RACE on restart: a new instance can mount and read the older
        main .db file before the previous instance's flush lands, silently
        rolling back recent writes. Observed directly - 20 freshly-written
        access grants were present in the bucket copy while the freshly-restarted
        app reported none of them.
        Persistence was therefore non-deterministic, which is worse than a known
        loss window because it looks correct most of the time.
    `PRAGMA wal_checkpoint(TRUNCATE)` forces the WAL contents into the main
    database file and resets the WAL, so the file the next boot reads is already
    up to date and there is nothing racing to be flushed.
    """
    interval = int(os.environ.get("CHECKPOINT_INTERVAL", "120"))
    db = DATA / "webui.db"
    log(f"checkpoint loop every {interval}s (wal_checkpoint(TRUNCATE) + json repair)")
    # A boot-time repair alone is not enough: the poison that caused the
    # 2026-08-26 incident was written by the RUNNING app roughly seven minutes
    # after a verified-clean restore. Sweeping here bounds the damage to one
    # cycle instead of one restart, so the once-per-second scheduler traceback
    # storm never gets long enough to keep the WAL hot.
    #
    # chat.meta is swept every cycle because it is the column the scheduler
    # scans once a second; the full discovery sweep is hourly because it walks
    # every table and does not need to be hot.
    hot = [("chat", "meta", "{}")]
    cycles = 0
    # Do not write while open-webui is migrating. See _wait_for_backend.
    _wait_for_backend()
    # First pass after the backend is up: alembic has finished, so a column the
    # running code needs and does not have is an unambiguous divergence.
    try:
        reconcile_schema(db)
    except Exception as exc:
        log("reconcile failed: {}: {}".format(type(exc).__name__, exc))
    while True:
        time.sleep(interval)
        cycles += 1
        try:
            if not db.exists():
                continue
            json_repair(db, columns=None if cycles % 30 == 0 else hot)
            con = sqlite3.connect(str(db), timeout=30)
            try:
                # Only meaningful in WAL mode. Under DELETE (the configuration
                # this bucket mount requires) every commit already lands in the
                # main file, so checkpointing is a pointless extra WRITER
                # against a database on a mount that does not like writers.
                if con.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal":
                    busy, logf, ckpt = con.execute(
                        "PRAGMA wal_checkpoint(TRUNCATE)").fetchone()
                    if busy:
                        log(f"checkpoint busy (wal={logf} pages) - will retry")
            finally:
                con.close()
        except Exception as exc:
            log(f"checkpoint failed, retrying next cycle: {type(exc).__name__}: {exc}")


def archive_loop():
    """One-way ARCHIVE: bucket -> Dataset. Never restores, never seeds.
    WHY THIS IS NOT THE DESIGN WE RETIRED. The failure earlier was two WRITERS
    disagreeing about where truth lives. This is a pure sink: /data is the only
    source of truth, boot never reads the Dataset unless /data is empty, and
    nothing here ever writes to /data. A sink has no opinion to disagree with.
    Credentials are scrubbed on the way out, so the archive stays consistent with
    the sanitize-wins policy on the Dataset even though the bucket holds live keys.
    """
    interval = int(os.environ.get("ARCHIVE_INTERVAL", "3600"))
    log(f"archive loop every {interval}s (db each cycle, assets every 24th) -> {REPO}")
    i = 0
    while True:
        time.sleep(interval)
        i += 1
        try:
            backup_db()
            if i % 24 == 0:
                backup_assets()
        except Exception as exc:
            log(f"archive failed, will retry next cycle: {type(exc).__name__}: {exc}")


def loop():
    log(f"backup loop every {INTERVAL}s (db each cycle, assets every 12th)")
    i = 0
    while True:
        time.sleep(INTERVAL)
        i += 1
        try:
            backup_db()
            if i % 12 == 0:
                backup_assets()
        except Exception as exc:                  # never let a sync fault kill the app
            log(f"backup failed, will retry next cycle: {type(exc).__name__}: {exc}")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "restore"
    table = {
        "restore": restore,
        "archive-loop": archive_loop,
        "checkpoint-loop": checkpoint_loop,
        "backup": lambda: (backup_db(), backup_assets()),
        "backup-db": backup_db,
        "backup-assets": backup_assets,
        "loop": loop,
        # Boot-path and hand-run maintenance. `repair` prints the number of
        # neutralised values so a boot log states plainly whether the database
        # arrived poisoned.
        "repair": lambda: print(json_repair(sys.argv[2] if len(sys.argv) > 2
                                            else DATA / "webui.db")),
        "prepare": lambda: prepare(sys.argv[2] if len(sys.argv) > 2 else None),
        "disable-automations": lambda: disable_automations(
            sys.argv[2] if len(sys.argv) > 2 else None),
        "wait-backend": lambda: _wait_for_backend(),
        "populated": lambda: sys.exit(
            0 if populated(sys.argv[2] if len(sys.argv) > 2 else DATA / "webui.db") else 1),
        "servable": lambda: sys.exit(
            0 if servable(sys.argv[2] if len(sys.argv) > 2 else DATA / "webui.db") else 1),
        "reconcile": lambda: print(reconcile_schema(
            sys.argv[2] if len(sys.argv) > 2 else None)),
        "integrity": lambda: sys.exit(
            0 if integrity(sys.argv[2] if len(sys.argv) > 2
                           else DATA / "webui.db") else 1),
        "recover": lambda: sys.exit(0 if recover(sys.argv[2], sys.argv[3]) else 1),
    }
    table[cmd]()