"""SQLite storage for the hive (swarm address book).
Single-writer, single-process, tiny — SQLite is deliberately enough here
(the hive is a notebook, not a Navi database). ``online`` is derived at read
time: a machine is considered present if it announced within the TTL window.
"""
from __future__ import annotations
import json
import sqlite3
import time
from pathlib import Path
_SCHEMA = """
CREATE TABLE IF NOT EXISTS machines (
instance_id TEXT PRIMARY KEY,
name TEXT NOT NULL,
host TEXT,
port INTEGER,
version TEXT,
meta TEXT,
first_seen REAL NOT NULL,
last_seen REAL NOT NULL
);
-- Append-only config history. A rollback is a new row, never a rewrite, so
-- "what was the fleet told, and when" is always answerable — including for a
-- document that turned out to be bad.
CREATE TABLE IF NOT EXISTS config_versions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
created_at REAL NOT NULL,
created_by TEXT,
note TEXT,
layers TEXT NOT NULL,
hash TEXT NOT NULL
);
-- The last reply from each instance, per version. This is the tracking the
-- owner asked for: who has it, who staged it, who rejected it and why.
CREATE TABLE IF NOT EXISTS config_acks (
instance_id TEXT NOT NULL,
version INTEGER NOT NULL,
hash TEXT NOT NULL,
status TEXT NOT NULL,
restart_required TEXT,
last_error TEXT,
acked_at REAL NOT NULL,
PRIMARY KEY (instance_id, version)
);
"""
class HiveDB:
def __init__(self, path: str | Path = "hive.db") -> None:
self._path = str(path)
self._conn = sqlite3.connect(self._path, check_same_thread=False)
self._conn.row_factory = sqlite3.Row
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.executescript(_SCHEMA)
self._conn.commit()
def close(self) -> None:
self._conn.close()
def upsert(
self,
*,
instance_id: str,
name: str,
host: str | None,
port: int,
version: str,
meta: dict,
) -> None:
now = time.time()
self._conn.execute(
"""
INSERT INTO machines
(instance_id, name, host, port, version, meta, first_seen, last_seen)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(instance_id) DO UPDATE SET
name = excluded.name,
host = COALESCE(excluded.host, machines.host),
port = excluded.port,
version = excluded.version,
meta = excluded.meta,
last_seen = excluded.last_seen
""",
(
instance_id,
name,
host,
port,
version,
json.dumps(meta),
now,
now,
),
)
self._conn.commit()
def list_peers(self, ttl_sec: int) -> list[dict]:
now = time.time()
rows = self._conn.execute(
"SELECT * FROM machines ORDER BY name COLLATE NOCASE"
).fetchall()
peers = []
for row in rows:
try:
meta = json.loads(row["meta"] or "{}")
except json.JSONDecodeError:
meta = {}
peers.append(
{
"name": row["name"],
"instance_id": row["instance_id"],
"host": row["host"],
"port": row["port"],
"address": f"{row['host']}:{row['port']}" if row["host"] else None,
"version": row["version"],
"online": (now - row["last_seen"]) <= ttl_sec,
"first_seen": row["first_seen"],
"last_seen": row["last_seen"],
"meta": meta,
}
)
return peers
# ── config documents ────────────────────────────────────────────────────
def add_config(
self,
*,
layers: list[dict],
doc_hash: str,
note: str = "",
created_by: str = "",
) -> dict:
now = time.time()
cur = self._conn.execute(
"""
INSERT INTO config_versions (created_at, created_by, note, layers, hash)
VALUES (?, ?, ?, ?, ?)
""",
(now, created_by, note, json.dumps(layers), doc_hash),
)
self._conn.commit()
return self.get_config(int(cur.lastrowid)) # type: ignore[arg-type]
def _version_row(self, row: sqlite3.Row) -> dict:
try:
layers = json.loads(row["layers"] or "[]")
except json.JSONDecodeError:
layers = []
return {
"id": row["id"],
"created_at": row["created_at"],
"created_by": row["created_by"],
"note": row["note"],
"layers": layers,
"hash": row["hash"],
}
def get_config(self, version_id: int) -> dict | None:
row = self._conn.execute(
"SELECT * FROM config_versions WHERE id = ?", (version_id,)
).fetchone()
return self._version_row(row) if row else None
def latest_config(self) -> dict | None:
row = self._conn.execute(
"SELECT * FROM config_versions ORDER BY id DESC LIMIT 1"
).fetchone()
return self._version_row(row) if row else None
def config_history(self, limit: int = 50) -> list[dict]:
rows = self._conn.execute(
"""
SELECT id, created_at, created_by, note, hash, layers
FROM config_versions ORDER BY id DESC LIMIT ?
""",
(max(1, min(limit, 500)),),
).fetchall()
out = []
for row in rows:
try:
count = len(json.loads(row["layers"] or "[]"))
except json.JSONDecodeError:
count = 0
out.append(
{
"id": row["id"],
"created_at": row["created_at"],
"created_by": row["created_by"],
"note": row["note"],
"hash": row["hash"],
"layers": count,
}
)
return out
# ── acks ────────────────────────────────────────────────────────────────
def record_ack(
self,
*,
instance_id: str,
version: int,
hash: str,
status: str,
restart_required: list[str] | None = None,
last_error: str | None = None,
) -> None:
self._conn.execute(
"""
INSERT INTO config_acks
(instance_id, version, hash, status, restart_required, last_error, acked_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(instance_id, version) DO UPDATE SET
hash = excluded.hash,
status = excluded.status,
restart_required = excluded.restart_required,
last_error = excluded.last_error,
acked_at = excluded.acked_at
""",
(
instance_id,
version,
hash,
status,
json.dumps(restart_required or []),
last_error,
time.time(),
),
)
self._conn.commit()
def get_acks(self, version: int) -> dict[str, dict]:
rows = self._conn.execute(
"SELECT * FROM config_acks WHERE version = ?", (version,)
).fetchall()
out: dict[str, dict] = {}
for row in rows:
try:
restart = json.loads(row["restart_required"] or "[]")
except json.JSONDecodeError:
restart = []
out[row["instance_id"]] = {
"status": row["status"],
"hash": row["hash"],
"restart_required": restart,
"last_error": row["last_error"],
"acked_at": row["acked_at"],
}
return out
