ota_db.py

"""
ota_db.py — minimal SQLite store for the single "latest OTA update" record.

Browser-back has no database otherwise (accounts live on the Messenger
backend, see auth.py). This mirrors the exact schema/functions from the
Messenger backend's proven OTA implementation, scoped down to just what
Browser-back needs.
"""

import sqlite3
import time
import os

DB_PATH = os.environ.get("OTA_DB_PATH", "ota.db")


def get_conn():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_conn()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS ota_update (
            id INTEGER PRIMARY KEY,
            update_id TEXT NOT NULL,
            runtime_version TEXT NOT NULL,
            launch_asset_json TEXT NOT NULL,
            assets_json TEXT NOT NULL,
            notes TEXT,
            created_at INTEGER NOT NULL
        )
    """)
    conn.commit()
    conn.close()


def set_latest_ota_update(update_id, runtime_version, launch_asset_json, assets_json, notes=None):
    """Always overwrites the single row (id=1) — there's only ever one
    'latest' OTA update, same pattern as the Messenger backend uses."""
    conn = get_conn()
    try:
        conn.execute("""
            INSERT INTO ota_update (id, update_id, runtime_version, launch_asset_json, assets_json, notes, created_at)
            VALUES (1, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(id) DO UPDATE SET update_id=excluded.update_id, runtime_version=excluded.runtime_version,
                launch_asset_json=excluded.launch_asset_json, assets_json=excluded.assets_json,
                notes=excluded.notes, created_at=excluded.created_at
        """, (update_id, runtime_version, launch_asset_json, assets_json, notes, int(time.time())))
        conn.commit()
        return True
    finally:
        conn.close()


def get_latest_ota_update():
    conn = get_conn()
    try:
        row = conn.execute("""
            SELECT update_id, runtime_version, launch_asset_json, assets_json, notes, created_at
            FROM ota_update WHERE id = 1
        """).fetchone()
        return dict(row) if row else None
    finally:
        conn.close()