admin/felhom.eu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2752e12accdbfb6a4e73e35367e3962145888944
felhom.eu/hub/internal/store/store.go
T
admin 2752e12acc hub v0.60.1: host deletion demotes escrow custody (never destroys) + customer-delete purge point + S6b obsolete 
- DeleteHost(deleteEscrow) demotes current host_escrow into host_escrow_superseded (copy-before-delete, same tx), spares existing; one shared demoteCurrentEscrowTx (reused by SaveHostEscrow). F-14 provenance/gate unchanged.
- DeleteCustomerConfig now purges both escrow tables for all the customer's hosts incl. already-deleted (F-14 provenance UNION) — the one true purge point.
- Wording: checkbox/refusal/Danger-zone → demotion. S6b OBSOLETE. Red-proofs TestDeleteHost_Demotes + TestDeleteCustomer_Purges + wording guard.
2026-07-17 11:25:38 +02:00

2798 lines
108 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package store
import (
"database/sql"
"encoding/json"
"errors"
"fmt"
"log"
"strconv"
"time"
"gitea.dooplex.hu/admin/felhom-hub/internal/semver"
_ "modernc.org/sqlite"
)
// Store handles SQLite persistence for customer reports.
type Store struct {
db *sql.DB
logger *log.Logger
// defaultMinControllerVersion is the config/env-supplied global FLOOR fallback (Phase 2 managed
// updates). Used only when neither a per-customer override nor a hub_settings row is set.
defaultMinControllerVersion string
}
// SetDefaultMinControllerVersion sets the config/env-supplied global floor fallback. Called once at
// startup from main (DEFAULT_MIN_CONTROLLER_VERSION). A hub_settings row, when present, takes
// precedence over this value (see GetGlobalMinControllerVersion).
func (s *Store) SetDefaultMinControllerVersion(v string) {
s.defaultMinControllerVersion = v
}
// CustomerSummary holds the latest status for a customer (for dashboard).
type CustomerSummary struct {
CustomerID string
CustomerName string
ControllerVersion string
ReceivedAt time.Time
HealthStatus string
CPUPercent float64
MemoryPercent float64
ContainerTotal int
ContainerRunning int
BackupLastSnapshot *time.Time
ReportJSON string
ControllerURL string
// Computed fields (not stored)
TimeSinceReport time.Duration
DiskSummary string
}
// New creates a new store and initializes the schema.
func New(dbPath string, logger *log.Logger) (*Store, error) {
db, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL&_busy_timeout=5000")
if err != nil {
return nil, fmt.Errorf("opening database: %w", err)
}
s := &Store{db: db, logger: logger}
if err := s.migrate(); err != nil {
db.Close()
return nil, fmt.Errorf("migrating database: %w", err)
}
return s, nil
}
func (s *Store) migrate() error {
_, err := s.db.Exec(`
CREATE TABLE IF NOT EXISTS reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id TEXT NOT NULL,
received_at DATETIME NOT NULL DEFAULT (datetime('now')),
report_json TEXT NOT NULL,
health_status TEXT,
cpu_percent REAL,
memory_percent REAL,
container_total INTEGER,
container_running INTEGER,
backup_last_snapshot DATETIME,
controller_version TEXT
);
CREATE INDEX IF NOT EXISTS idx_reports_customer
ON reports(customer_id, received_at DESC);
CREATE TABLE IF NOT EXISTS customer_notifications (
customer_id TEXT PRIMARY KEY,
email TEXT NOT NULL DEFAULT '',
enabled_events TEXT NOT NULL DEFAULT '[]',
created_at DATETIME DEFAULT (datetime('now')),
updated_at DATETIME DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS notification_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id TEXT NOT NULL,
event_type TEXT NOT NULL,
severity TEXT NOT NULL,
message TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending',
error_message TEXT,
created_at DATETIME DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_notification_log_customer
ON notification_log(customer_id, created_at DESC);
CREATE TABLE IF NOT EXISTS customer_configs (
customer_id TEXT PRIMARY KEY,
customer_name TEXT NOT NULL DEFAULT '',
domain TEXT NOT NULL DEFAULT '',
email TEXT NOT NULL DEFAULT '',
retrieval_password TEXT NOT NULL,
api_key TEXT NOT NULL,
config_json TEXT NOT NULL DEFAULT '{}',
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
`)
if err != nil {
return err
}
// v0.1.8: add controller_url column (idempotent — ignore error if already exists)
s.db.Exec("ALTER TABLE reports ADD COLUMN controller_url TEXT")
// v0.2.1: add status column to customer_configs (idempotent)
s.db.Exec("ALTER TABLE customer_configs ADD COLUMN status TEXT NOT NULL DEFAULT 'active'")
// v0.15.0: per-customer minimum controller version (the managed-update FLOOR). Empty = no
// per-customer override → the effective floor falls back to the global default (hub_settings /
// config). Idempotent.
s.db.Exec("ALTER TABLE customer_configs ADD COLUMN min_controller_version TEXT NOT NULL DEFAULT ''")
// v0.26.0: per-customer config_version — a monotonic counter bumped on every config save. The
// report ACK advertises it; the controller compares it against its last-applied version and, on a
// change, re-pulls controller.yaml + self-restarts (pull-based config delivery — no inbound). It is
// a STORED COUNTER, never a hash of the rendered YAML (configgen emits a fresh session_secret +
// timestamp every call, so a content hash would change spuriously). Idempotent. Existing rows seed
// at 1, so an already-running box records that as its baseline on its next report without restarting.
s.db.Exec("ALTER TABLE customer_configs ADD COLUMN config_version INTEGER NOT NULL DEFAULT 1")
// v0.51.0 (DR-tier-by-default): per-customer DR-tier flag. NEW customers default ON (the
// create handler sets it); the column default 0 is the LEGACY initialization — an existing
// customer is only flipped ON by the one-time backfill (initialize from reality: its host
// already carries an ENABLED pbs_dr descriptor; never auto-cascade a legacy box). The
// backfill runs exactly once — only when the ALTER actually added the column — and is
// DEFERRED to the end of migrate(): the hosts table it reads is created further down (a
// fresh DB reaches the ALTER before the CREATE).
_, drTierAlterErr := s.db.Exec("ALTER TABLE customer_configs ADD COLUMN dr_tier INTEGER NOT NULL DEFAULT 0")
// v0.15.0: hub_settings — a tiny key/value table for operator-set globals that must survive
// restarts (currently only the global controller-version floor). The config/env DEFAULT_MIN_
// CONTROLLER_VERSION is the FALLBACK; a row here (set via the operator UI) overrides it.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS hub_settings (
key TEXT PRIMARY KEY,
value TEXT NOT NULL DEFAULT '',
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
`)
if err != nil {
return err
}
// v0.3.0: events table for hub-native monitoring
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id TEXT NOT NULL,
event_type TEXT NOT NULL,
severity TEXT NOT NULL DEFAULT 'info',
message TEXT NOT NULL DEFAULT '',
details_json TEXT NOT NULL DEFAULT '{}',
source TEXT NOT NULL DEFAULT 'controller',
created_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_events_customer_created
ON events(customer_id, created_at DESC);
CREATE INDEX IF NOT EXISTS idx_events_type
ON events(event_type, created_at DESC);
`)
if err != nil {
return err
}
// v0.3.0: add cooldown_hours to customer_notifications (idempotent)
s.db.Exec("ALTER TABLE customer_notifications ADD COLUMN cooldown_hours INTEGER DEFAULT 6")
// v0.3.0: add channel column to notification_log (idempotent)
s.db.Exec("ALTER TABLE notification_log ADD COLUMN channel TEXT NOT NULL DEFAULT 'customer'")
// v0.4.0: app telemetry tables
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS app_telemetry (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id TEXT NOT NULL,
app_name TEXT NOT NULL,
display_name TEXT NOT NULL DEFAULT '',
reported_at DATETIME NOT NULL,
memory_current_mb REAL DEFAULT 0,
memory_avg_mb REAL DEFAULT 0,
memory_peak_mb REAL DEFAULT 0,
cpu_avg_percent REAL DEFAULT 0,
catalog_estimate TEXT DEFAULT '',
catalog_limit TEXT DEFAULT '',
log_errors INTEGER DEFAULT 0,
log_warnings INTEGER DEFAULT 0,
containers_json TEXT DEFAULT '[]',
issues_json TEXT DEFAULT '[]'
);
CREATE INDEX IF NOT EXISTS idx_app_telemetry_lookup
ON app_telemetry(app_name, reported_at);
CREATE INDEX IF NOT EXISTS idx_app_telemetry_customer
ON app_telemetry(customer_id, app_name, reported_at);
CREATE INDEX IF NOT EXISTS idx_app_telemetry_prune
ON app_telemetry(reported_at);
CREATE TABLE IF NOT EXISTS app_log_issues (
id INTEGER PRIMARY KEY AUTOINCREMENT,
app_name TEXT NOT NULL,
fingerprint TEXT NOT NULL,
severity TEXT NOT NULL,
message TEXT NOT NULL,
first_seen DATETIME NOT NULL,
last_seen DATETIME NOT NULL,
occurrence_count INTEGER DEFAULT 1,
affected_customers TEXT DEFAULT '[]',
UNIQUE(app_name, fingerprint)
);
CREATE INDEX IF NOT EXISTS idx_app_log_issues_app
ON app_log_issues(app_name, last_seen DESC);
`)
if err != nil {
return err
}
// Phase-1 retire (2026-06-16): the Infra Backup mechanism is gone. It pushed
// plaintext customer secrets to the hub (the app-secret encryption key, restic
// password, Cloudflare tokens — a zero-knowledge violation) and had been dead since
// slice 8C. Drop both tables and VACUUM so the freed pages — which still hold the
// plaintext — are physically reclaimed from the DB file, not merely delinked.
// Gated on existence so ordinary restarts don't pay the VACUUM cost.
var infraTables int
s.db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type='table'
AND name IN ('infra_backup_versions','infra_backups')`).Scan(&infraTables)
if infraTables > 0 {
if _, err = s.db.Exec(`DROP TABLE IF EXISTS infra_backup_versions;
DROP TABLE IF EXISTS infra_backups;`); err != nil {
return fmt.Errorf("dropping retired infra_backup tables: %w", err)
}
// VACUUM rewrites the database file, discarding the freed (plaintext) pages.
if _, err = s.db.Exec(`VACUUM`); err != nil {
return fmt.Errorf("vacuum after infra_backup drop: %w", err)
}
// WAL checkpoint(TRUNCATE) so no plaintext lingers in the -wal sidecar either.
s.db.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`)
if s.logger != nil {
s.logger.Printf("[INFO] Retired infra-backup: dropped %d table(s) and VACUUMed to reclaim plaintext pages", infraTables)
}
}
// v0.7.0: host-domain (slice 3). Purely additive — the controller path
// (reports/customer_configs) is untouched; the schema cutover is slice 10.
// Columns marked INERT exist now so slice 10 needs no ALTER; nothing reads or
// writes them this slice.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS hosts (
host_id TEXT PRIMARY KEY,
customer_id TEXT NOT NULL,
api_key TEXT NOT NULL,
agent_version TEXT NOT NULL DEFAULT '',
last_report_at DATETIME,
desired_json TEXT NOT NULL DEFAULT '{}',
desired_generation INTEGER NOT NULL DEFAULT 0,
dr_record_json TEXT NOT NULL DEFAULT '{}',
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_hosts_customer ON hosts(customer_id);
CREATE TABLE IF NOT EXISTS guests (
guest_id TEXT PRIMARY KEY,
customer_id TEXT NOT NULL,
host_id TEXT NOT NULL,
vmid INTEGER NOT NULL,
display_name TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL DEFAULT 'unknown',
controller_version TEXT NOT NULL DEFAULT '',
last_seen_at DATETIME,
api_key TEXT NOT NULL DEFAULT '',
desired_spec_json TEXT NOT NULL DEFAULT '{}',
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_guests_host ON guests(host_id);
CREATE INDEX IF NOT EXISTS idx_guests_customer ON guests(customer_id);
CREATE TABLE IF NOT EXISTS host_reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
host_id TEXT NOT NULL,
customer_id TEXT NOT NULL,
received_at DATETIME NOT NULL DEFAULT (datetime('now')),
report_json TEXT NOT NULL,
agent_version TEXT,
cpu_percent REAL,
memory_percent REAL,
disk_percent REAL,
guest_total INTEGER,
guest_running INTEGER,
cloudflared_status TEXT
);
CREATE INDEX IF NOT EXISTS idx_host_reports_host ON host_reports(host_id, received_at DESC);
CREATE INDEX IF NOT EXISTS idx_host_reports_customer ON host_reports(customer_id, received_at DESC);
-- host_escrow (slice 7, doc 03 §8a): the OPAQUE R-wrapped PBS-key escrow blob. The hub
-- stores the ciphertext bytes against the host and NEVER decrypts them (it has no recovery
-- code). One row per host; a re-upload (rotation) is last-write-wins. Restore-mode serving
-- (handing the blob back to a re-enrolling box) is slice 10.
CREATE TABLE IF NOT EXISTS host_escrow (
host_id TEXT PRIMARY KEY,
blob BLOB NOT NULL,
key_fingerprint TEXT NOT NULL DEFAULT '',
posture TEXT NOT NULL DEFAULT '',
created_at DATETIME NOT NULL,
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
-- host_escrow_superseded (v0.60.0, offsite-continuity): RETAINED old escrow blobs. Viktor's
-- ruling (data protection first): when a NEW escrow blob supersedes an old one whose sealed
-- restic-password sha DIFFERS, the old row is COPIED here BEFORE host_escrow is overwritten —
-- so the old passphrase stays customer-R-recoverable (turns the reinstall-orphan incident from
-- "history destroyed" into "history recoverable with the recovery code"). Append-only; the hub
-- never decrypts; NO pruning (the blobs are tiny + R-encrypted; custody unchanged). The ACK and
-- restore-serving read host_escrow (the CURRENT row) — never this table.
CREATE TABLE IF NOT EXISTS host_escrow_superseded (
id INTEGER PRIMARY KEY AUTOINCREMENT,
host_id TEXT NOT NULL,
blob BLOB NOT NULL,
key_fingerprint TEXT NOT NULL DEFAULT '',
posture TEXT NOT NULL DEFAULT '',
created_at DATETIME NOT NULL,
restic_pw_sha256 TEXT NOT NULL DEFAULT '',
superseded_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_host_escrow_superseded_host ON host_escrow_superseded(host_id);
-- signed_jobs (slice 10A): the per-host queue of OPAQUE operator-signed destructive-op
-- blobs. The hub STORES + SERVES them; it never forges one (there is no signing key
-- hub-side) and never executes them (execution + signature verification is slice 10B).
-- HasSignedOps on the control envelope is "this host has >=1 pending job". A job is opaque
-- bytes (the signed-op envelope the agent verifies); the hub treats it as a blob.
CREATE TABLE IF NOT EXISTS signed_jobs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
host_id TEXT NOT NULL,
job_id TEXT NOT NULL,
blob BLOB NOT NULL,
created_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE INDEX IF NOT EXISTS idx_signed_jobs_host ON signed_jobs(host_id, id);
`)
if err != nil {
return err
}
// Slice 10D (DR capstone) — additive columns on existing tables (fire-and-forget; a duplicate
// column on re-run is ignored). `recovery_mode_until` gates restore-directive serving + re-enroll
// (NULL/past = off; future = recovery mode active, auto-expires). host_escrow gains the IDENTITY
// blob (age-wrapped {tunnel_token, pbs_token}) + the NON-secret DR directive (pbs repo/namespace,
// expected key fingerprint, tunnel id) — the hub serves these only in recovery mode; no usable
// secret is hub-held (the blobs need R, which the hub never has).
s.db.Exec(`ALTER TABLE hosts ADD COLUMN recovery_mode_until DATETIME`)
s.db.Exec(`ALTER TABLE host_escrow ADD COLUMN identity_blob BLOB`)
s.db.Exec(`ALTER TABLE host_escrow ADD COLUMN directive_json TEXT NOT NULL DEFAULT '{}'`)
// SLICE 3 (escrow auto-confirm) — sha256 hex of the offsite restic repo password sealed in the
// identity blob. The hash of a 256-bit random secret is non-reversible/non-brute-forceable — safe to
// store and serve; it lets the controller VERIFY "the escrow covers the CURRENT repo password"
// instead of trusting blob-presence. NULL/'' = a legacy or password-less blob (never auto-confirms).
s.db.Exec(`ALTER TABLE host_escrow ADD COLUMN restic_pw_sha256 TEXT`)
// v0.57.0 (2.3, escrow honesty on offsite re-issue) — stale_at is set when the offsite repo
// password is re-issued: the blob then seals a password that no longer opens the repo, so the
// hub must stop advertising "ceremony done" and withhold the (now non-matching) restic_pw_sha256
// from the auto-confirm ACK. NULL = current; a fresh ceremony (SaveHostEscrow) clears it.
s.db.Exec(`ALTER TABLE host_escrow ADD COLUMN stale_at DATETIME`)
// dr_recipe (SPIKE-dr-recipe-2026-06-16): the secret-free DR reconstruction recipe, stored
// PLAINTEXT (it has NO secrets — the clean inverse of the retired infra_backup). Two halves keyed
// by customer: the agent's storage/guest/PBS half (host_half_json, from the host-report) and the
// controller's customer/apps half (app_half_json, from the controller report); the hub assembles
// them on read. DEDICATED table, separate from the opaque host_escrow. One row per customer;
// each half is last-write-wins and preserves the other.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS dr_recipe (
customer_id TEXT PRIMARY KEY,
recipe_version INTEGER NOT NULL DEFAULT 1,
host_id TEXT NOT NULL DEFAULT '',
host_half_json TEXT NOT NULL DEFAULT '',
app_half_json TEXT NOT NULL DEFAULT '',
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
`)
if err != nil {
return err
}
// S1 offsite connectivity (doc 06 §3.2): the WG endpoint record + peer registry. The hub is the
// source of truth; the endpoint converges on it (SSH push, internal/wgsync). No `status` column
// on wg_peers — presence in the table IS the desired state; the doc-06 `status` field belongs to
// the S2 host-join. assigned_ip stores the BARE IP (no /32 — that's presentation, appended by
// the API layer and the sync payload). UNIQUE(assigned_ip) is the allocator's race backstop.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS wg_endpoints (
endpoint_id TEXT PRIMARY KEY,
dns_name TEXT NOT NULL,
wg_port INTEGER NOT NULL,
server_pubkey TEXT NOT NULL,
tunnel_subnet TEXT NOT NULL,
pbs_tunnel_ip TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS wg_peers (
pubkey TEXT PRIMARY KEY,
assigned_ip TEXT NOT NULL UNIQUE,
host_id TEXT NOT NULL DEFAULT '',
note TEXT NOT NULL DEFAULT '',
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
-- S2: one BOUND peer per host (partial index — S1's unbound admin/test rows unaffected).
CREATE UNIQUE INDEX IF NOT EXISTS idx_wg_peers_host ON wg_peers(host_id) WHERE host_id != '';
`)
if err != nil {
return err
}
// host_recovery (TASK G1): the break-glass root@pam console credential, vaulted at rest and
// operator-retrievable. UNLIKE host_escrow (opaque, hub-can't-open), this IS a hub-held secret the
// operator retrieves to reach the PVE web console (pveproxy — a failure domain distinct from sshd)
// when both the sshd path AND the agent-independent auto-heal have failed. One row per host,
// last-write-wins (day-0 sets it; --rotate re-sets). Never in desired-state, never logged.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS host_recovery (
host_id TEXT PRIMARY KEY,
username TEXT NOT NULL,
secret TEXT NOT NULL,
set_at DATETIME NOT NULL DEFAULT (datetime('now')),
updated_at DATETIME NOT NULL DEFAULT (datetime('now'))
);
`)
if err != nil {
return err
}
// offsite provisioning (SLICE 1): the ONE-TIME transient storage-box/subaccount password. The hub
// generates it at provision time, delivers it to the controller EXACTLY ONCE (consume endpoint), then
// it is dead — the controller installs its own key + the hub resets the box password. It is transient
// custody, NOT the ConfigJSON (which is served every pull). One row per customer; consumed_at marks it
// spent. Never logged, never in any served config. (No hub-side at-rest cipher exists; the DB file is
// 0600 and the value is short-lived + single-use.)
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS one_time_secrets (
customer_id TEXT PRIMARY KEY,
value TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
consumed_at DATETIME
);
`)
if err != nil {
return err
}
// PBS DR tier (SLICE 1, v0.44.0): the one-time PBS token secret, HOST-scoped — the sibling of
// one_time_secrets (customer/controller custody) for host/agent custody. The hub receives the
// secret over the tenantsync channel, stores it here, and the AGENT consumes it exactly once
// (POST /api/v1/hosts/{id}/pbs/consume-token, per-host key). consumed_at marks it spent; a
// re-issue supersedes any unconsumed value. Never logged, never in desired-state/ConfigJSON.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS host_pbs_secrets (
host_id TEXT PRIMARY KEY,
value TEXT NOT NULL,
created_at DATETIME NOT NULL DEFAULT (datetime('now')),
consumed_at DATETIME
);
`)
if err != nil {
return err
}
// v0.50.0 — customer-claim password arc (DRILL-day0-vm F-4): one row per customer holding the
// ACTIVE claim/reset code state. code_hash is bcrypt(code) — the plaintext exists ONLY inside
// the email send (same custody rule as the retrieval passphrase). generation is monotonic: a
// resend/reset rotates the code (generation+1) and the controller refuses codes of an already-
// consumed generation. claimed_at is set once (first successful claim) and NEVER cleared by a
// rotation — a reset code on a claimed box must not un-claim it. emailed_at records the last
// send; reset_day/reset_count are the hub-side 3/day reset-request limiter.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS customer_claims (
customer_id TEXT PRIMARY KEY,
code_hash TEXT NOT NULL,
generation INTEGER NOT NULL DEFAULT 1,
issued_at DATETIME NOT NULL,
emailed_at DATETIME,
claimed_at DATETIME,
reset_day TEXT NOT NULL DEFAULT '',
reset_count INTEGER NOT NULL DEFAULT 0
);
`)
if err != nil {
return err
}
// v0.43.0 — remote app-log diagnostics. Additive columns on app_log_issues:
// context = JSON array of ±5 redacted lines around the FIRST occurrence (first capture
// wins — stable repro context, no churn); context_customer = whose box it came from
// (provenance); dismissed_at = dismissal instead of futile deletion (the controller
// re-sends recurring issues — a delete is undone minutes later; a dismissal stays until
// a genuinely NEW occurrence with last_seen > dismissed_at resurfaces the row).
s.db.Exec(`ALTER TABLE app_log_issues ADD COLUMN context TEXT`)
s.db.Exec(`ALTER TABLE app_log_issues ADD COLUMN context_customer TEXT`)
s.db.Exec(`ALTER TABLE app_log_issues ADD COLUMN dismissed_at DATETIME`)
// log_tail_requests: the operator's pending "send me this app's logs" intents — the
// pull-based ACK flag (the hub NEVER connects into a box; the controller sees the flag
// in its report ACK and ships the tail on its next cycle). One active request per
// (customer, app) — a re-click refreshes requested_at. Cleared when the tail arrives
// (consume-once). app_log_tails: the received tails, transient — keep the last 2 per
// (customer, app), older pruned at insert.
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS log_tail_requests (
customer_id TEXT NOT NULL,
app_name TEXT NOT NULL,
requested_at DATETIME NOT NULL,
PRIMARY KEY (customer_id, app_name)
);
CREATE TABLE IF NOT EXISTS app_log_tails (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id TEXT NOT NULL,
app_name TEXT NOT NULL,
collected_at DATETIME NOT NULL,
received_at DATETIME NOT NULL,
lines_json TEXT NOT NULL
);
`)
if err != nil {
return err
}
// Component log bundles (v0.46.0 observability, see logbundle.go): the operator's
// pending pull intents per (scope, component) — scope = customer_id for the
// controller (report ACK channel) / host_id for the agent (heartbeat envelope) —
// plus the received gzip bundles (72 h TTL, purge on the 60 s sweep; a secret-gate
// hit stores a BLOCKED flag row with no payload).
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS log_bundle_requests (
scope_id TEXT NOT NULL,
component TEXT NOT NULL,
requested_at DATETIME NOT NULL,
PRIMARY KEY (scope_id, component)
);
CREATE TABLE IF NOT EXISTS log_bundles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
scope_id TEXT NOT NULL,
component TEXT NOT NULL,
collected_at DATETIME NOT NULL,
received_at DATETIME NOT NULL,
size_bytes INTEGER NOT NULL,
gz BLOB,
blocked INTEGER NOT NULL DEFAULT 0,
blocked_reason TEXT NOT NULL DEFAULT ''
);
`)
if err != nil {
return err
}
// v0.53.0 — host-deletion provenance (F-14, operator ruling 2026-07-13): one row per DeleteHost,
// written INSIDE the delete transaction. escrow_acked records whether the host was removed
// through the escrow-ack flow (the operator explicitly acknowledged destroying a PRESENT escrow
// row — acknowledged key destruction). The PBS-DR enable path may auto-re-issue a surviving ep0
// tenancy ONLY when the customer's most recent record here has escrow_acked=1; no record (all
// pre-v0.53.0 deletions — deliberately NO backfill) or an un-acked record keeps the manual
// re-issue path the only one (never-silently-re-key law).
_, err = s.db.Exec(`
CREATE TABLE IF NOT EXISTS host_deletions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
host_id TEXT NOT NULL,
customer_id TEXT NOT NULL,
deleted_at DATETIME NOT NULL DEFAULT (datetime('now')),
escrow_acked INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_host_deletions_customer ON host_deletions(customer_id, id DESC);
`)
if err != nil {
return err
}
// v0.51.0 dr_tier one-time legacy backfill — see the ALTER above; runs last so every table
// it touches (hosts, customer_configs) exists on a fresh DB too (where it finds nothing).
if drTierAlterErr == nil {
if err := s.backfillDRTierFromDescriptors(); err != nil {
return fmt.Errorf("dr_tier backfill: %w", err)
}
}
return nil
}
// NotificationPrefs holds per-customer notification preferences.
type NotificationPrefs struct {
CustomerID string
Email string
EnabledEvents []string
CooldownHours int
}
// GetNotificationPrefs returns notification preferences for a customer.
func (s *Store) GetNotificationPrefs(customerID string) (*NotificationPrefs, error) {
var email, eventsJSON string
var cooldownHours int
err := s.db.QueryRow(
"SELECT email, enabled_events, COALESCE(cooldown_hours, 6) FROM customer_notifications WHERE customer_id = ?",
customerID,
).Scan(&email, &eventsJSON, &cooldownHours)
if err != nil {
if err == sql.ErrNoRows {
return nil, nil
}
return nil, err
}
var events []string
if err := json.Unmarshal([]byte(eventsJSON), &events); err != nil {
s.logger.Printf("[WARN] Corrupt enabled_events JSON for %s: %v", customerID, err)
}
if cooldownHours <= 0 {
cooldownHours = 6
}
return &NotificationPrefs{
CustomerID: customerID,
Email: email,
EnabledEvents: events,
CooldownHours: cooldownHours,
}, nil
}
// SaveNotificationPrefs creates or updates notification preferences for a customer.
func (s *Store) SaveNotificationPrefs(customerID, email string, enabledEvents []string, cooldownHours int) error {
eventsJSON, _ := json.Marshal(enabledEvents)
if cooldownHours <= 0 {
cooldownHours = 6
}
_, err := s.db.Exec(`
INSERT INTO customer_notifications (customer_id, email, enabled_events, cooldown_hours, updated_at)
VALUES (?, ?, ?, ?, datetime('now'))
ON CONFLICT(customer_id) DO UPDATE SET
email = excluded.email,
enabled_events = excluded.enabled_events,
cooldown_hours = excluded.cooldown_hours,
updated_at = datetime('now')`,
customerID, email, string(eventsJSON), cooldownHours,
)
return err
}
// LogNotification records a notification attempt.
func (s *Store) LogNotification(customerID, eventType, severity, message, status, errorMsg, channel string) error {
if channel == "" {
channel = "customer"
}
_, err := s.db.Exec(`
INSERT INTO notification_log (customer_id, event_type, severity, message, status, error_message, channel)
VALUES (?, ?, ?, ?, ?, ?, ?)`,
customerID, eventType, severity, message, status, errorMsg, channel,
)
return err
}
// NotificationLogEntry represents a single notification log record.
type NotificationLogEntry struct {
EventType string
Severity string
Message string
Status string // "sent", "skipped", "failed"
ErrorMessage string
Channel string // "operator" or "customer"
CreatedAt time.Time
}
// GetRecentNotifications returns the most recent notification log entries for a customer.
func (s *Store) GetRecentNotifications(customerID string, limit int) ([]NotificationLogEntry, error) {
rows, err := s.db.Query(`
SELECT event_type, severity, message, status, COALESCE(error_message, ''), COALESCE(channel, 'customer'), created_at
FROM notification_log
WHERE customer_id = ?
ORDER BY created_at DESC
LIMIT ?`, customerID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
var entries []NotificationLogEntry
for rows.Next() {
var e NotificationLogEntry
var createdAt, errorMsg string
if err := rows.Scan(&e.EventType, &e.Severity, &e.Message, &e.Status, &errorMsg, &e.Channel, &createdAt); err != nil {
return nil, err
}
e.CreatedAt = parseSQLiteTime(createdAt)
e.ErrorMessage = errorMsg
entries = append(entries, e)
}
return entries, rows.Err()
}
// SaveReport stores a new report. The reportJSON should be the raw JSON payload.
func (s *Store) SaveReport(customerID string, reportJSON []byte) error {
// Parse denormalized fields from the JSON
var parsed struct {
ControllerVersion string `json:"controller_version"`
ControllerURL string `json:"controller_url"`
System struct {
CPUPercent float64 `json:"cpu_percent"`
MemoryPercent float64 `json:"memory_percent"`
} `json:"system"`
Containers struct {
Total int `json:"total"`
Running int `json:"running"`
} `json:"containers"`
Backup struct {
LastSnapshot *time.Time `json:"last_snapshot"`
} `json:"backup"`
Health struct {
Status string `json:"status"`
} `json:"health"`
}
if err := json.Unmarshal(reportJSON, &parsed); err != nil {
s.logger.Printf("[WARN] Cannot parse report fields for denormalization: %v", err)
}
var backupSnapshot *string
if parsed.Backup.LastSnapshot != nil {
t := parsed.Backup.LastSnapshot.Format(time.RFC3339)
backupSnapshot = &t
}
_, err := s.db.Exec(`
INSERT INTO reports (customer_id, report_json, health_status, cpu_percent,
memory_percent, container_total, container_running,
backup_last_snapshot, controller_version, controller_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
customerID, string(reportJSON),
parsed.Health.Status, parsed.System.CPUPercent,
parsed.System.MemoryPercent, parsed.Containers.Total,
parsed.Containers.Running, backupSnapshot,
parsed.ControllerVersion, parsed.ControllerURL,
)
return err
}
// GetCustomers returns the latest report summary for each customer.
func (s *Store) GetCustomers() ([]CustomerSummary, error) {
rows, err := s.db.Query(`
SELECT r.customer_id, r.received_at, r.report_json,
r.health_status, r.cpu_percent, r.memory_percent,
r.container_total, r.container_running,
r.backup_last_snapshot, r.controller_version, r.controller_url
FROM reports r
INNER JOIN (
SELECT customer_id, MAX(received_at) as max_time
FROM reports
GROUP BY customer_id
) latest ON r.customer_id = latest.customer_id
AND r.received_at = latest.max_time
ORDER BY r.customer_id`)
if err != nil {
return nil, err
}
defer rows.Close()
var customers []CustomerSummary
for rows.Next() {
var c CustomerSummary
var receivedAt string
var backupSnapshot sql.NullString
var controllerURL sql.NullString
if err := rows.Scan(&c.CustomerID, &receivedAt, &c.ReportJSON,
&c.HealthStatus, &c.CPUPercent, &c.MemoryPercent,
&c.ContainerTotal, &c.ContainerRunning,
&backupSnapshot, &c.ControllerVersion, &controllerURL); err != nil {
return nil, err
}
c.ReceivedAt = parseSQLiteTime(receivedAt)
c.TimeSinceReport = time.Since(c.ReceivedAt)
if backupSnapshot.Valid {
t, err := time.Parse(time.RFC3339, backupSnapshot.String)
if err == nil {
c.BackupLastSnapshot = &t
}
}
if controllerURL.Valid {
c.ControllerURL = controllerURL.String
}
// Parse customer_name from JSON
var report struct {
CustomerName string `json:"customer_name"`
}
if err := json.Unmarshal([]byte(c.ReportJSON), &report); err != nil {
s.logger.Printf("[WARN] Cannot parse customer_name from report JSON for %s: %v", c.CustomerID, err)
}
c.CustomerName = report.CustomerName
// Parse disk summary
c.DiskSummary = parseDiskSummary(c.ReportJSON)
customers = append(customers, c)
}
return customers, rows.Err()
}
// GetCustomer returns the latest report for a specific customer.
func (s *Store) GetCustomer(customerID string) (*CustomerSummary, error) {
row := s.db.QueryRow(`
SELECT customer_id, received_at, report_json,
health_status, cpu_percent, memory_percent,
container_total, container_running,
backup_last_snapshot, controller_version, controller_url
FROM reports
WHERE customer_id = ?
ORDER BY received_at DESC
LIMIT 1`, customerID)
var c CustomerSummary
var receivedAt string
var backupSnapshot sql.NullString
var controllerURL sql.NullString
if err := row.Scan(&c.CustomerID, &receivedAt, &c.ReportJSON,
&c.HealthStatus, &c.CPUPercent, &c.MemoryPercent,
&c.ContainerTotal, &c.ContainerRunning,
&backupSnapshot, &c.ControllerVersion, &controllerURL); err != nil {
if err == sql.ErrNoRows {
return nil, nil
}
return nil, err
}
c.ReceivedAt = parseSQLiteTime(receivedAt)
c.TimeSinceReport = time.Since(c.ReceivedAt)
if backupSnapshot.Valid {
t, err := time.Parse(time.RFC3339, backupSnapshot.String)
if err == nil {
c.BackupLastSnapshot = &t
}
}
if controllerURL.Valid {
c.ControllerURL = controllerURL.String
}
var report struct {
CustomerName string `json:"customer_name"`
}
json.Unmarshal([]byte(c.ReportJSON), &report)
c.CustomerName = report.CustomerName
c.DiskSummary = parseDiskSummary(c.ReportJSON)
return &c, nil
}
// GetCustomerHistory returns report history for a customer.
func (s *Store) GetCustomerHistory(customerID string, since time.Duration) ([]CustomerSummary, error) {
cutoff := time.Now().Add(-since).Format("2006-01-02 15:04:05")
rows, err := s.db.Query(`
SELECT customer_id, received_at, report_json,
health_status, cpu_percent, memory_percent,
container_total, container_running,
backup_last_snapshot, controller_version, controller_url
FROM reports
WHERE customer_id = ? AND received_at >= ?
ORDER BY received_at DESC`, customerID, cutoff)
if err != nil {
return nil, err
}
defer rows.Close()
var history []CustomerSummary
for rows.Next() {
var c CustomerSummary
var receivedAt string
var backupSnapshot sql.NullString
var controllerURL sql.NullString
if err := rows.Scan(&c.CustomerID, &receivedAt, &c.ReportJSON,
&c.HealthStatus, &c.CPUPercent, &c.MemoryPercent,
&c.ContainerTotal, &c.ContainerRunning,
&backupSnapshot, &c.ControllerVersion, &controllerURL); err != nil {
return nil, err
}
c.ReceivedAt = parseSQLiteTime(receivedAt)
c.TimeSinceReport = time.Since(c.ReceivedAt)
if backupSnapshot.Valid {
t, err := time.Parse(time.RFC3339, backupSnapshot.String)
if err == nil {
c.BackupLastSnapshot = &t
}
}
if controllerURL.Valid {
c.ControllerURL = controllerURL.String
}
history = append(history, c)
}
return history, rows.Err()
}
// (Infra-backup store methods retired 2026-06-16. The tables are dropped + VACUUMed
// in migrate(); the push/get/versions handlers and the operator panel are gone.)
// Prune deletes reports older than the given number of days.
func (s *Store) Prune(maxDays int) (int64, error) {
cutoff := time.Now().AddDate(0, 0, -maxDays).Format("2006-01-02 15:04:05")
res, err := s.db.Exec("DELETE FROM reports WHERE received_at < ?", cutoff)
if err != nil {
return 0, err
}
n, _ := res.RowsAffected()
// v0.7.0: prune the parallel host-domain report stream, same retention.
if hres, herr := s.db.Exec("DELETE FROM host_reports WHERE received_at < ?", cutoff); herr == nil {
hn, _ := hres.RowsAffected()
n += hn
}
return n, nil
}
// Close closes the database connection.
func (s *Store) Close() error {
return s.db.Close()
}
// backfillDRTierFromDescriptors is the ONE-TIME legacy initialization for the v0.51.0 dr_tier
// column (called only when the ALTER just added it): a customer whose host already carries an
// ENABLED pbs_dr descriptor in desired_json gets dr_tier=1 — initialize from reality; everyone
// else stays 0 (never auto-cascade a legacy box). JSON is inspected in Go, not by LIKE, so field
// order/whitespace can't fool it.
func (s *Store) backfillDRTierFromDescriptors() error {
rows, err := s.db.Query(`SELECT customer_id, desired_json FROM hosts WHERE desired_json LIKE '%pbs_dr%'`)
if err != nil {
return err
}
defer rows.Close()
enabled := map[string]bool{}
for rows.Next() {
var customerID, desired string
if err := rows.Scan(&customerID, &desired); err != nil {
return err
}
var doc struct {
PBSDR *struct {
Enabled bool `json:"enabled"`
} `json:"pbs_dr"`
}
if json.Unmarshal([]byte(desired), &doc) == nil && doc.PBSDR != nil && doc.PBSDR.Enabled {
enabled[customerID] = true
}
}
if err := rows.Err(); err != nil {
return err
}
for customerID := range enabled {
if _, err := s.db.Exec(`UPDATE customer_configs SET dr_tier = 1 WHERE customer_id = ?`, customerID); err != nil {
return err
}
}
return nil
}
// CustomerConfig holds a pre-provisioned customer configuration.
type CustomerConfig struct {
CustomerID string
CustomerName string
Domain string
Email string
RetrievalPassword string
APIKey string
ConfigJSON string // JSON object with customer-specific override fields
Status string // "active" or "blocked"
// MinControllerVersion is the per-customer minimum controller version (managed-update FLOOR
// override). Empty = use the global default. Set/cleared via the operator UI.
MinControllerVersion string
// ConfigVersion is the monotonic config counter (bumped on every SaveCustomerConfig). The report
// ACK advertises it; the controller re-pulls + self-restarts when it changes. Never a YAML hash.
ConfigVersion int
// DRTier (v0.51.0, DR-tier-by-default) is the per-customer DR-tier flag — the operator INTENT
// the pbsdr cascade converges toward. Default ON for NEW customers (create handler); legacy
// rows were initialized from reality by the one-time backfill (enabled descriptor → ON).
// OFF = zero Felhom-side cost (no ep0 namespace is provisioned) and offsite provisioning is
// refused (the escrow ceremony depends on the PBS key — drill F-6, closed by policy).
DRTier bool
CreatedAt time.Time
UpdatedAt time.Time
}
// SaveCustomerConfig creates or updates a customer configuration. Every save BUMPS config_version
// (new rows start at 1; updates increment) — this is the load-bearing signal that drives the
// controller's pull-based config-refresh via the report ACK. Bumping here covers every field that
// feeds the generated controller.yaml (identity + the config_json overrides). NOTE: the floor
// (min_controller_version), block/unblock status, and retrieval-password regen are deliberately NOT
// config.yaml content and intentionally do NOT bump it (they have their own signals or none).
func (s *Store) SaveCustomerConfig(cfg *CustomerConfig) error {
_, err := s.db.Exec(`
INSERT INTO customer_configs (customer_id, customer_name, domain, email,
retrieval_password, api_key, config_json, min_controller_version, dr_tier, config_version, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, datetime('now'))
ON CONFLICT(customer_id) DO UPDATE SET
customer_name = excluded.customer_name,
domain = excluded.domain,
email = excluded.email,
retrieval_password = excluded.retrieval_password,
api_key = excluded.api_key,
config_json = excluded.config_json,
min_controller_version = excluded.min_controller_version,
dr_tier = excluded.dr_tier,
config_version = customer_configs.config_version + 1,
updated_at = datetime('now')`,
cfg.CustomerID, cfg.CustomerName, cfg.Domain, cfg.Email,
cfg.RetrievalPassword, cfg.APIKey, cfg.ConfigJSON, cfg.MinControllerVersion, cfg.DRTier,
)
return err
}
// GetCustomerConfig returns a customer configuration by ID, or nil if not found.
func (s *Store) GetCustomerConfig(customerID string) (*CustomerConfig, error) {
var cfg CustomerConfig
var createdAt, updatedAt string
err := s.db.QueryRow(`
SELECT customer_id, customer_name, domain, email,
retrieval_password, api_key, config_json, status, min_controller_version, dr_tier, config_version, created_at, updated_at
FROM customer_configs WHERE customer_id = ?`,
customerID,
).Scan(&cfg.CustomerID, &cfg.CustomerName, &cfg.Domain, &cfg.Email,
&cfg.RetrievalPassword, &cfg.APIKey, &cfg.ConfigJSON, &cfg.Status, &cfg.MinControllerVersion,
&cfg.DRTier, &cfg.ConfigVersion, &createdAt, &updatedAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
cfg.CreatedAt = parseSQLiteTime(createdAt)
cfg.UpdatedAt = parseSQLiteTime(updatedAt)
return &cfg, nil
}
// SaveOneTimeSecret stores (last-write-wins) the one-time transient offsite password for a customer,
// resetting the consumed flag (a fresh provision supersedes any prior unconsumed value). Never logged.
func (s *Store) SaveOneTimeSecret(customerID, value string) error {
_, err := s.db.Exec(`
INSERT INTO one_time_secrets (customer_id, value, created_at, consumed_at)
VALUES (?, ?, datetime('now'), NULL)
ON CONFLICT(customer_id) DO UPDATE SET value = excluded.value, created_at = datetime('now'), consumed_at = NULL`,
customerID, value)
return err
}
// ConsumeOneTimeSecret returns the customer's one-time offsite password and marks it consumed in the SAME
// transaction (single use). A second call — or a call when none is stored — returns ("", sql.ErrNoRows).
// The value is never logged.
func (s *Store) ConsumeOneTimeSecret(customerID string) (string, error) {
tx, err := s.db.Begin()
if err != nil {
return "", err
}
defer tx.Rollback()
var value string
err = tx.QueryRow(`SELECT value FROM one_time_secrets WHERE customer_id = ? AND consumed_at IS NULL`, customerID).Scan(&value)
if err != nil {
return "", err // sql.ErrNoRows when absent OR already consumed
}
if _, err := tx.Exec(`UPDATE one_time_secrets SET consumed_at = datetime('now') WHERE customer_id = ?`, customerID); err != nil {
return "", err
}
if err := tx.Commit(); err != nil {
return "", err
}
return value, nil
}
// ListCustomerConfigs returns all customer configurations ordered by ID.
func (s *Store) ListCustomerConfigs() ([]CustomerConfig, error) {
rows, err := s.db.Query(`
SELECT customer_id, customer_name, domain, email,
retrieval_password, api_key, config_json, status, min_controller_version, dr_tier, config_version, created_at, updated_at
FROM customer_configs ORDER BY customer_id`)
if err != nil {
return nil, err
}
defer rows.Close()
var configs []CustomerConfig
for rows.Next() {
var cfg CustomerConfig
var createdAt, updatedAt string
if err := rows.Scan(&cfg.CustomerID, &cfg.CustomerName, &cfg.Domain, &cfg.Email,
&cfg.RetrievalPassword, &cfg.APIKey, &cfg.ConfigJSON, &cfg.Status, &cfg.MinControllerVersion,
&cfg.DRTier, &cfg.ConfigVersion, &createdAt, &updatedAt); err != nil {
return nil, err
}
cfg.CreatedAt = parseSQLiteTime(createdAt)
cfg.UpdatedAt = parseSQLiteTime(updatedAt)
configs = append(configs, cfg)
}
return configs, rows.Err()
}
// DeleteCustomerConfig deletes a customer configuration AND purges the customer's escrow custody
// (v0.60.1). The customer Danger-zone Delete is the ONE true purge point for recovery-key custody:
// host deletion only DEMOTES a blob to retained custody (never destroys), so removing the customer is
// the deliberate, acknowledged point where that retained custody is permanently removed. In one tx it
// deletes host_escrow AND host_escrow_superseded for ALL the customer's hosts — INCLUDING hosts
// already deleted (whose demoted blobs survive in host_escrow_superseded), resolved via the F-14
// host_deletions provenance so a host-delete-then-customer-delete ordering leaves nothing orphaned.
// The broader offboarding lifecycle (Hetzner sub-account, WG peer, Storage-Box data, the host rows
// themselves) is NOT this method — see the delete/re-create rehearsal (ROADMAP R-3).
func (s *Store) DeleteCustomerConfig(customerID string) error {
tx, err := s.db.Begin()
if err != nil {
return err
}
defer tx.Rollback()
// Current hosts' escrow.
if _, err := tx.Exec(`DELETE FROM host_escrow WHERE host_id IN (SELECT host_id FROM hosts WHERE customer_id = ?)`, customerID); err != nil {
return fmt.Errorf("DeleteCustomerConfig %s: purge host_escrow: %w", customerID, err)
}
// Retained (superseded) blobs for BOTH current and already-deleted hosts of this customer.
if _, err := tx.Exec(`
DELETE FROM host_escrow_superseded WHERE host_id IN (
SELECT host_id FROM hosts WHERE customer_id = ?
UNION
SELECT host_id FROM host_deletions WHERE customer_id = ?
)`, customerID, customerID); err != nil {
return fmt.Errorf("DeleteCustomerConfig %s: purge host_escrow_superseded: %w", customerID, err)
}
if _, err := tx.Exec(`DELETE FROM customer_configs WHERE customer_id = ?`, customerID); err != nil {
return fmt.Errorf("DeleteCustomerConfig %s: delete config: %w", customerID, err)
}
return tx.Commit()
}
// GetCustomerConfigByAPIKey looks up a customer config by its unique API key.
// Returns nil if no matching key is found.
func (s *Store) GetCustomerConfigByAPIKey(apiKey string) (*CustomerConfig, error) {
var cfg CustomerConfig
var createdAt, updatedAt string
err := s.db.QueryRow(`
SELECT customer_id, customer_name, domain, email,
retrieval_password, api_key, config_json, status, min_controller_version, dr_tier, config_version, created_at, updated_at
FROM customer_configs WHERE api_key = ?`,
apiKey,
).Scan(&cfg.CustomerID, &cfg.CustomerName, &cfg.Domain, &cfg.Email,
&cfg.RetrievalPassword, &cfg.APIKey, &cfg.ConfigJSON, &cfg.Status, &cfg.MinControllerVersion,
&cfg.DRTier, &cfg.ConfigVersion, &createdAt, &updatedAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
cfg.CreatedAt = parseSQLiteTime(createdAt)
cfg.UpdatedAt = parseSQLiteTime(updatedAt)
return &cfg, nil
}
// ── Customer-claim password arc (v0.50.0, DRILL-day0-vm F-4) ──────────────────────────────
// ClaimState is one customer's active claim/reset-code state. CodeHash is bcrypt(code) — the
// store NEVER holds a plaintext code. Claimed means the customer has completed the claim (set
// their own password) at least once; rotations never clear it.
type ClaimState struct {
CustomerID string
CodeHash string
Generation int
IssuedAt time.Time
EmailedAt *time.Time
ClaimedAt *time.Time
ResetDay string
ResetCount int
}
// Claimed reports whether the claim has been completed at least once.
func (c *ClaimState) Claimed() bool { return c != nil && c.ClaimedAt != nil }
// GetClaim returns the customer's claim state, or nil when none exists.
func (s *Store) GetClaim(customerID string) (*ClaimState, error) {
var cs ClaimState
var issuedAt string
var emailedAt, claimedAt sql.NullString
err := s.db.QueryRow(`
SELECT customer_id, code_hash, generation, issued_at, emailed_at, claimed_at, reset_day, reset_count
FROM customer_claims WHERE customer_id = ?`, customerID,
).Scan(&cs.CustomerID, &cs.CodeHash, &cs.Generation, &issuedAt, &emailedAt, &claimedAt, &cs.ResetDay, &cs.ResetCount)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
cs.IssuedAt = parseSQLiteTime(issuedAt)
if emailedAt.Valid {
t := parseSQLiteTime(emailedAt.String)
cs.EmailedAt = &t
}
if claimedAt.Valid {
t := parseSQLiteTime(claimedAt.String)
cs.ClaimedAt = &t
}
return &cs, nil
}
// RotateClaimCode installs a fresh code hash: creates the row (generation 1) or rotates it
// (generation+1, new issued_at, emailed_at cleared until the send is confirmed). claimed_at is
// deliberately PRESERVED — rotating a claimed customer's code (a password reset) never un-claims
// the box. Returns the new generation.
func (s *Store) RotateClaimCode(customerID, codeHash string) (int, error) {
_, err := s.db.Exec(`
INSERT INTO customer_claims (customer_id, code_hash, generation, issued_at)
VALUES (?, ?, 1, datetime('now'))
ON CONFLICT(customer_id) DO UPDATE SET
code_hash = excluded.code_hash,
generation = customer_claims.generation + 1,
issued_at = datetime('now'),
emailed_at = NULL`,
customerID, codeHash)
if err != nil {
return 0, err
}
var gen int
if err := s.db.QueryRow(`SELECT generation FROM customer_claims WHERE customer_id = ?`, customerID).Scan(&gen); err != nil {
return 0, err
}
return gen, nil
}
// MarkClaimEmailed records that the active code was delivered to the registered address.
func (s *Store) MarkClaimEmailed(customerID string) error {
_, err := s.db.Exec(`UPDATE customer_claims SET emailed_at = datetime('now') WHERE customer_id = ?`, customerID)
return err
}
// MarkClaimed records the first successful claim (idempotent — an already-set claimed_at stays).
// Returns true when this call performed the unclaimed→claimed transition.
func (s *Store) MarkClaimed(customerID string) (bool, error) {
res, err := s.db.Exec(`UPDATE customer_claims SET claimed_at = datetime('now') WHERE customer_id = ? AND claimed_at IS NULL`, customerID)
if err != nil {
return false, err
}
n, _ := res.RowsAffected()
return n > 0, nil
}
// BumpResetCount enforces the hub-side reset-request limiter: increments today's counter and
// returns the post-increment count (the caller compares against the daily cap). The day key
// rolls over automatically (UTC date).
func (s *Store) BumpResetCount(customerID string) (int, error) {
day := time.Now().UTC().Format("2006-01-02")
_, err := s.db.Exec(`
UPDATE customer_claims SET
reset_count = CASE WHEN reset_day = ? THEN reset_count + 1 ELSE 1 END,
reset_day = ?
WHERE customer_id = ?`, day, day, customerID)
if err != nil {
return 0, err
}
var count int
if err := s.db.QueryRow(`SELECT reset_count FROM customer_claims WHERE customer_id = ?`, customerID).Scan(&count); err != nil {
return 0, err
}
return count, nil
}
// SetCustomerConfigStatus sets the status (active/blocked) for a customer config.
func (s *Store) SetCustomerConfigStatus(customerID, status string) error {
_, err := s.db.Exec(`
UPDATE customer_configs SET status = ?, updated_at = datetime('now')
WHERE customer_id = ?`,
status, customerID,
)
return err
}
// SetMinControllerVersion sets (or clears, with "") the per-customer controller-version floor
// override. The customer config must already exist.
func (s *Store) SetMinControllerVersion(customerID, version string) error {
_, err := s.db.Exec(`
UPDATE customer_configs SET min_controller_version = ?, updated_at = datetime('now')
WHERE customer_id = ?`,
version, customerID,
)
return err
}
// GetGlobalMinControllerVersion returns the operator-set global floor from hub_settings if present,
// else the config/env-supplied default. Empty string = no global floor.
func (s *Store) GetGlobalMinControllerVersion() string {
var v string
err := s.db.QueryRow(`SELECT value FROM hub_settings WHERE key = 'min_controller_version'`).Scan(&v)
if err == nil && v != "" {
return v
}
// No DB override (missing row or empty value) → fall back to the config/env default.
return s.defaultMinControllerVersion
}
// GlobalFloorResolution is the full picture of the effective global floor for the operator UI: the
// resolved value + WHICH source won + both raw inputs. It makes the "a manifest save silently armed
// a live floor" incident (publish-train 0.81/0.113) permanently visible — the operator can see the
// DB override vs the env fallback at a glance.
type GlobalFloorResolution struct {
Effective string // the value GetGlobalMinControllerVersion returns ("" = no floor)
Source string // "db" | "env" | "none"
DBValue string // the hub_settings row value ("" = no row / cleared)
EnvValue string // the DEFAULT_MIN_CONTROLLER_VERSION fallback
}
// ResolveGlobalFloor reports the effective floor AND its source (DB hub_settings row vs the env
// default). Mirrors GetGlobalMinControllerVersion's precedence exactly — do not fork the rule.
func (s *Store) ResolveGlobalFloor() GlobalFloorResolution {
res := GlobalFloorResolution{EnvValue: s.defaultMinControllerVersion}
var db string
if err := s.db.QueryRow(`SELECT value FROM hub_settings WHERE key = 'min_controller_version'`).Scan(&db); err == nil {
res.DBValue = db
}
switch {
case res.DBValue != "":
res.Effective, res.Source = res.DBValue, "db"
case res.EnvValue != "":
res.Effective, res.Source = res.EnvValue, "env"
default:
res.Source = "none"
}
return res
}
// SetGlobalMinControllerVersion persists the operator-set global floor (overriding the config/env
// default). Pass "" to clear the override and fall back to the default.
func (s *Store) SetGlobalMinControllerVersion(version string) error {
_, err := s.db.Exec(`
INSERT INTO hub_settings (key, value, updated_at)
VALUES ('min_controller_version', ?, datetime('now'))
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = datetime('now')`,
version,
)
return err
}
// ArtifactManifest is the operator-vouched current artifact set (agent binary + golden archive)
// served to the host-bootstrap script so it can verify-before-install. The hub is the TRUST ROOT
// for these checksums (a different root than Gitea, which only STORES the bytes): the script fetches
// each artifact from Gitea with the config-retrieve git token, then checks its sha256 against the
// value recorded here before installing/using it. Empty fields = nothing published yet (the script
// then falls back to the local golden / fails clearly on a missing binary).
type ArtifactManifest struct {
AgentVersion string `json:"agent_version"`
AgentSHA256 string `json:"agent_sha256"`
GoldenVersion string `json:"golden_version"`
GoldenSHA256 string `json:"golden_sha256"`
// MinAgent is the MINIMUM host-agent version this golden's controller requires (the controller
// CHANGELOG `MinAgent:` value the operator vouches at manifest time). Empty = an UNCOUPLED
// release: no per-box agent gating. When set, the hub HOLDS the controller-version floor for any
// box whose agent is below it (Part D) — mechanising the publish-train "agent BEFORE controller
// floor" rule instead of leaving it to operator discipline.
MinAgent string `json:"min_agent"`
}
// hub_settings keys for the artifact manifest (BUNDLE slice). Stored as discrete key/value rows in
// the existing hub_settings table — same mechanism as the controller-version floor, so it survives
// restarts and needs no schema change.
const (
settingArtifactAgentVersion = "artifact_agent_version"
settingArtifactAgentSHA256 = "artifact_agent_sha256"
settingArtifactGoldenVersion = "artifact_golden_version"
settingArtifactGoldenSHA256 = "artifact_golden_sha256"
settingArtifactMinAgent = "artifact_min_agent"
)
// settingOperatorPasswordHash is the hub_settings key for the operator login password bcrypt hash,
// set via the Configuration UI (v0.54.0). When present it OVERRIDES the config/env seed
// (auth.password_hash in hub.yaml) — the same DB-override-wins precedence as the controller-version
// floor. The ConfigMap value stays the break-glass fallback: clear this row (or edit the manifest +
// redeploy) to reset a lost password.
const settingOperatorPasswordHash = "operator_password_hash"
// GetOperatorPasswordHash returns the UI-set operator password bcrypt hash, or "" when none has been
// set (the config/env seed is then authoritative).
func (s *Store) GetOperatorPasswordHash() string { return s.getSetting(settingOperatorPasswordHash) }
// SetOperatorPasswordHash persists a new operator password bcrypt hash set via the Configuration UI.
func (s *Store) SetOperatorPasswordHash(hash string) error {
return s.setSetting(settingOperatorPasswordHash, hash)
}
// getSetting reads a single hub_settings value ("" if the row is absent).
func (s *Store) getSetting(key string) string {
var v string
if err := s.db.QueryRow(`SELECT value FROM hub_settings WHERE key = ?`, key).Scan(&v); err != nil {
return ""
}
return v
}
// setSetting upserts a single hub_settings value.
func (s *Store) setSetting(key, value string) error {
_, err := s.db.Exec(`
INSERT INTO hub_settings (key, value, updated_at)
VALUES (?, ?, datetime('now'))
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = datetime('now')`,
key, value,
)
return err
}
// GetArtifactManifest returns the operator-recorded current artifact set. All-empty when nothing
// has been published yet.
func (s *Store) GetArtifactManifest() ArtifactManifest {
return ArtifactManifest{
AgentVersion: s.getSetting(settingArtifactAgentVersion),
AgentSHA256: s.getSetting(settingArtifactAgentSHA256),
GoldenVersion: s.getSetting(settingArtifactGoldenVersion),
GoldenSHA256: s.getSetting(settingArtifactGoldenSHA256),
MinAgent: s.getSetting(settingArtifactMinAgent),
}
}
// SetArtifactManifest persists the operator-recorded current artifact set (all four fields). Each
// field is stored independently so a partial form submission (e.g. agent only) still round-trips.
func (s *Store) SetArtifactManifest(m ArtifactManifest) error {
if err := s.setSetting(settingArtifactAgentVersion, m.AgentVersion); err != nil {
return err
}
if err := s.setSetting(settingArtifactAgentSHA256, m.AgentSHA256); err != nil {
return err
}
if err := s.setSetting(settingArtifactGoldenVersion, m.GoldenVersion); err != nil {
return err
}
if err := s.setSetting(settingArtifactGoldenSHA256, m.GoldenSHA256); err != nil {
return err
}
return s.setSetting(settingArtifactMinAgent, m.MinAgent)
}
// EffectiveMinControllerVersion resolves the floor that actually applies to a customer: the
// per-customer override when set (non-empty), otherwise the global floor (hub_settings → config/env
// default). Returns "" when no floor applies at all (Phase 2 inert for that customer).
func (s *Store) EffectiveMinControllerVersion(customerID string) string {
cfg, err := s.GetCustomerConfig(customerID)
if err == nil && cfg != nil && cfg.MinControllerVersion != "" {
return cfg.MinControllerVersion
}
return s.GetGlobalMinControllerVersion()
}
// ManagedFloorDecision is the per-box outcome of the MinAgent conditional floor (Part D): the floor
// to actually serve this customer's controller, whether it is being HELD (and why), and the inputs.
type ManagedFloorDecision struct {
Floor string // the controller-version floor to SERVE ("" = serve none)
Held bool // true = the floor is withheld because the box's agent is below MinAgent
AgentVersion string // the box's reported agent version ("" = unknown → held when MinAgent is set)
MinAgent string // the manifest's MinAgent for the current golden ("" = uncoupled, no gating)
}
// ResolveManagedFloor decides the controller-version floor to serve a customer, HOLDING it when the
// golden the floor points at requires a newer host agent than the box currently runs (Part D — the
// hub-enforced "agent BEFORE controller floor" rule). Logic:
// - effective floor "" → nothing to serve (no floor configured);
// - manifest MinAgent "" → UNCOUPLED release: serve the floor as-is (no agent gating);
// - agent_version known AND ≥ MinAgent → serve the floor;
// - agent_version below MinAgent, OR unknown/unparseable → HOLD (serve no directive) + flag.
// A held box is VISIBLE (the dashboard renders the reason), never silently stale.
func (s *Store) ResolveManagedFloor(customerID string) ManagedFloorDecision {
d := ManagedFloorDecision{Floor: s.EffectiveMinControllerVersion(customerID)}
if d.Floor == "" {
return d
}
d.MinAgent = s.GetArtifactManifest().MinAgent
if d.MinAgent == "" {
return d // uncoupled release — no agent gate
}
if h, err := s.GetHostByCustomer(customerID); err == nil && h != nil {
d.AgentVersion = h.AgentVersion
}
if d.AgentVersion != "" && semver.Valid(d.AgentVersion) && semver.Valid(d.MinAgent) &&
semver.Compare(d.AgentVersion, d.MinAgent) >= 0 {
return d // agent is new enough — serve the floor
}
// Agent too old, or unknown/unparseable → hold the floor (never push a controller past its agent).
d.Held = true
d.Floor = ""
return d
}
// IsCustomerBlocked returns true if the customer config has status "blocked".
func (s *Store) IsCustomerBlocked(customerID string) bool {
var status string
err := s.db.QueryRow(
"SELECT status FROM customer_configs WHERE customer_id = ?",
customerID,
).Scan(&status)
return err == nil && status == "blocked"
}
// UpdateRetrievalPassword updates the retrieval password for a customer config.
func (s *Store) UpdateRetrievalPassword(customerID, newPassword string) error {
_, err := s.db.Exec(`
UPDATE customer_configs SET retrieval_password = ?, updated_at = datetime('now')
WHERE customer_id = ?`,
newPassword, customerID,
)
return err
}
// --- Event system ---
// Event represents a single event record.
type Event struct {
ID int64
CustomerID string
EventType string
Severity string // "info", "warning", "error"
Message string
DetailsJSON string // raw JSON
Source string // "controller" or "hub"
CreatedAt time.Time
}
// SaveEvent inserts a new event and returns its ID.
func (s *Store) SaveEvent(customerID, eventType, severity, message, detailsJSON, source string) (int64, error) {
if detailsJSON == "" {
detailsJSON = "{}"
}
if source == "" {
source = "controller"
}
res, err := s.db.Exec(`
INSERT INTO events (customer_id, event_type, severity, message, details_json, source)
VALUES (?, ?, ?, ?, ?, ?)`,
customerID, eventType, severity, message, detailsJSON, source,
)
if err != nil {
return 0, err
}
return res.LastInsertId()
}
// GetRecentEvents returns the most recent events for a customer, newest first.
func (s *Store) GetRecentEvents(customerID string, limit int) ([]Event, error) {
rows, err := s.db.Query(`
SELECT id, customer_id, event_type, severity, message, details_json, source, created_at
FROM events
WHERE customer_id = ?
ORDER BY created_at DESC
LIMIT ?`, customerID, limit)
if err != nil {
return nil, err
}
defer rows.Close()
return scanEvents(rows)
}
// GetEventsByType returns events of a specific type for a customer since a given time.
func (s *Store) GetEventsByType(customerID, eventType string, since time.Time) ([]Event, error) {
rows, err := s.db.Query(`
SELECT id, customer_id, event_type, severity, message, details_json, source, created_at
FROM events
WHERE customer_id = ? AND event_type = ? AND created_at >= ?
ORDER BY created_at DESC`,
customerID, eventType, since.UTC().Format("2006-01-02 15:04:05"))
if err != nil {
return nil, err
}
defer rows.Close()
return scanEvents(rows)
}
// GetLatestEventByType returns the most recent event of a given type for a customer.
func (s *Store) GetLatestEventByType(customerID, eventType string) (*Event, error) {
var e Event
var createdAt string
err := s.db.QueryRow(`
SELECT id, customer_id, event_type, severity, message, details_json, source, created_at
FROM events
WHERE customer_id = ? AND event_type = ?
ORDER BY created_at DESC
LIMIT 1`, customerID, eventType,
).Scan(&e.ID, &e.CustomerID, &e.EventType, &e.Severity, &e.Message, &e.DetailsJSON, &e.Source, &createdAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
e.CreatedAt = parseSQLiteTime(createdAt)
return &e, nil
}
// GetAllRecentEvents returns the most recent events across all customers.
func (s *Store) GetAllRecentEvents(limit int) ([]Event, error) {
rows, err := s.db.Query(`
SELECT id, customer_id, event_type, severity, message, details_json, source, created_at
FROM events
ORDER BY created_at DESC
LIMIT ?`, limit)
if err != nil {
return nil, err
}
defer rows.Close()
return scanEvents(rows)
}
// CountEventsBySeverity returns a count of events per severity for a customer since a given time.
func (s *Store) CountEventsBySeverity(customerID string, since time.Time) (map[string]int, error) {
rows, err := s.db.Query(`
SELECT severity, COUNT(*) FROM events
WHERE customer_id = ? AND created_at >= ?
GROUP BY severity`,
customerID, since.UTC().Format("2006-01-02 15:04:05"))
if err != nil {
return nil, err
}
defer rows.Close()
counts := make(map[string]int)
for rows.Next() {
var sev string
var count int
if err := rows.Scan(&sev, &count); err != nil {
return nil, err
}
counts[sev] = count
}
return counts, rows.Err()
}
// PruneEvents deletes events older than the given number of days.
func (s *Store) PruneEvents(maxDays int) (int64, error) {
cutoff := time.Now().AddDate(0, 0, -maxDays).UTC().Format("2006-01-02 15:04:05")
res, err := s.db.Exec("DELETE FROM events WHERE created_at < ?", cutoff)
if err != nil {
return 0, err
}
return res.RowsAffected()
}
// GetActiveCustomerIDs returns customer IDs from customer_configs where status is 'active'.
func (s *Store) GetActiveCustomerIDs() ([]string, error) {
rows, err := s.db.Query("SELECT customer_id FROM customer_configs WHERE status = 'active'")
if err != nil {
return nil, err
}
defer rows.Close()
var ids []string
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
return nil, err
}
ids = append(ids, id)
}
return ids, rows.Err()
}
// Ping verifies the database is accessible.
func (s *Store) Ping() error {
var n int
return s.db.QueryRow("SELECT 1").Scan(&n)
}
func scanEvents(rows *sql.Rows) ([]Event, error) {
var events []Event
for rows.Next() {
var e Event
var createdAt string
if err := rows.Scan(&e.ID, &e.CustomerID, &e.EventType, &e.Severity, &e.Message, &e.DetailsJSON, &e.Source, &createdAt); err != nil {
return nil, err
}
e.CreatedAt = parseSQLiteTime(createdAt)
events = append(events, e)
}
return events, rows.Err()
}
// parseSQLiteTime tries multiple formats that modernc.org/sqlite may return.
func parseSQLiteTime(s string) time.Time {
formats := []string{
"2006-01-02 15:04:05", // SQLite datetime('now')
"2006-01-02T15:04:05Z", // RFC3339 without fractional
time.RFC3339, // 2006-01-02T15:04:05Z07:00
time.RFC3339Nano, // with fractional seconds
"2006-01-02 15:04:05+00:00", // with explicit UTC offset
"2006-01-02 15:04:05.999999999", // with fractional, no TZ
}
for _, f := range formats {
if t, err := time.Parse(f, s); err == nil {
return t
}
}
// Last resort: if string is non-empty, log it for debugging
if s != "" {
log.Printf("[WARN] Could not parse timestamp: %q", s)
}
return time.Time{} // zero time
}
func parseDiskSummary(reportJSON string) string {
var report struct {
Storage []struct {
Mount string `json:"mount"`
Percent float64 `json:"percent"`
} `json:"storage"`
}
// ignore parse errors — show "–" on failure
json.Unmarshal([]byte(reportJSON), &report) //nolint:errcheck
var parts []string
for _, s := range report.Storage {
parts = append(parts, fmt.Sprintf("%.0f%%", s.Percent))
}
if len(parts) == 0 {
return "–"
}
result := parts[0]
for _, p := range parts[1:] {
result += "/" + p
}
return result
}
// ---- v0.7.0: host-domain (slice 3) ----
// Additive store surface for the agent's host-report stream. The controller-path
// methods above are untouched.
// Host is one customer agent. Mixes operator-intent columns (Desired*, DRRecord —
// INERT until slice 10) with box-reported reality (AgentVersion, LastReportAt).
type Host struct {
HostID string
CustomerID string
APIKey string
AgentVersion string
LastReportAt *time.Time
DesiredJSON string
DesiredGeneration int64
DRRecordJSON string
RecoveryModeUntil *time.Time // slice 10D: recovery mode active until this time (nil/past = off)
CreatedAt time.Time
UpdatedAt time.Time
}
// InRecoveryMode reports whether the host is currently in recovery mode (set + not expired).
func (h *Host) InRecoveryMode(now time.Time) bool {
return h.RecoveryModeUntil != nil && now.Before(*h.RecoveryModeUntil)
}
// Guest is one controller LXC. Reality columns are report-driven; APIKey and
// DesiredSpecJSON are INERT until slice 10 and must survive report upserts.
type Guest struct {
GuestID string
CustomerID string
HostID string
VMID int
DisplayName string
Status string
ControllerVersion string
LastSeenAt *time.Time
APIKey string
DesiredSpecJSON string
CreatedAt time.Time
UpdatedAt time.Time
}
// HostReportDenorm are the denormalized fields pulled from a host-report for the
// dashboard / staleness, mirroring the reports table's denorm pattern.
type HostReportDenorm struct {
AgentVersion string
CPUPercent float64
MemoryPercent float64
DiskPercent float64
GuestTotal int
GuestRunning int
CloudflaredStatus string
}
// HostStaleRow is the minimal per-host recency row the dead-man's-switch reads.
type HostStaleRow struct {
HostID string
CustomerID string
LastReportAt time.Time
}
// GuestID derives the interim guest primary key from host + vmid. The hub owns the
// id scheme (locked decision 3) so the slice-10 swap to durable ids is hub-only.
func GuestID(hostID string, vmid int) string {
return hostID + "/" + strconv.Itoa(vmid)
}
func scanHost(scan func(dest ...any) error) (*Host, error) {
var h Host
var lastReport, recoveryUntil sql.NullString
var createdAt, updatedAt string
err := scan(&h.HostID, &h.CustomerID, &h.APIKey, &h.AgentVersion, &lastReport,
&h.DesiredJSON, &h.DesiredGeneration, &h.DRRecordJSON, &recoveryUntil, &createdAt, &updatedAt)
if err != nil {
return nil, err
}
if lastReport.Valid {
t := parseSQLiteTime(lastReport.String)
h.LastReportAt = &t
}
if recoveryUntil.Valid && recoveryUntil.String != "" {
t := parseSQLiteTime(recoveryUntil.String)
h.RecoveryModeUntil = &t
}
h.CreatedAt = parseSQLiteTime(createdAt)
h.UpdatedAt = parseSQLiteTime(updatedAt)
return &h, nil
}
const hostSelectCols = `host_id, customer_id, api_key, agent_version, last_report_at,
desired_json, desired_generation, dr_record_json, recovery_mode_until, created_at, updated_at`
// GetHostByAPIKey looks up a host by its per-host hub key. Returns nil (no error)
// if no match — parallels GetCustomerConfigByAPIKey.
func (s *Store) GetHostByAPIKey(apiKey string) (*Host, error) {
h, err := scanHost(s.db.QueryRow(`SELECT `+hostSelectCols+` FROM hosts WHERE api_key = ?`, apiKey).Scan)
if err == sql.ErrNoRows {
return nil, nil
}
return h, err
}
// GetHost looks up a host by id. Returns nil (no error) if not found.
func (s *Store) GetHost(hostID string) (*Host, error) {
h, err := scanHost(s.db.QueryRow(`SELECT `+hostSelectCols+` FROM hosts WHERE host_id = ?`, hostID).Scan)
if err == sql.ErrNoRows {
return nil, nil
}
return h, err
}
// GetHostByCustomer returns the customer's host, or nil (no error) if none exists.
// Backs the passphrase-authed host-enroll mint-once-reuse path (Day-0 option C): on
// the second enroll the existing credential is reused, not re-minted. A customer is
// expected to have at most one host in the Day-0 model; if more than one ever exists,
// the most-recently-updated wins (we never mint a duplicate on a reuse). Uses the
// idx_hosts_customer index. Mirrors GetHostByAPIKey's nil-on-not-found contract.
func (s *Store) GetHostByCustomer(customerID string) (*Host, error) {
h, err := scanHost(s.db.QueryRow(`SELECT `+hostSelectCols+
` FROM hosts WHERE customer_id = ? ORDER BY updated_at DESC LIMIT 1`, customerID).Scan)
if err == sql.ErrNoRows {
return nil, nil
}
return h, err
}
// ListHostsByCustomer returns the customer's hosts ordered by host_id (v0.47.0 — the
// customer page's Host tab is a LIST by design: 1 host today, N for a later HA cluster).
// Uses the idx_hosts_customer index.
func (s *Store) ListHostsByCustomer(customerID string) ([]Host, error) {
rows, err := s.db.Query(`SELECT `+hostSelectCols+
` FROM hosts WHERE customer_id = ? ORDER BY host_id`, customerID)
if err != nil {
return nil, err
}
defer rows.Close()
var hosts []Host
for rows.Next() {
h, err := scanHost(rows.Scan)
if err != nil {
return nil, err
}
hosts = append(hosts, *h)
}
return hosts, rows.Err()
}
// ListHosts returns all hosts (debug / host-domain views).
func (s *Store) ListHosts() ([]Host, error) {
rows, err := s.db.Query(`SELECT ` + hostSelectCols + ` FROM hosts ORDER BY host_id`)
if err != nil {
return nil, err
}
defer rows.Close()
var hosts []Host
for rows.Next() {
h, err := scanHost(rows.Scan)
if err != nil {
return nil, err
}
hosts = append(hosts, *h)
}
return hosts, rows.Err()
}
// ErrHostEscrowPresent is returned by DeleteHost when the host still has a key-escrow row
// and the caller did not explicitly acknowledge deleting it (fail-safe-to-refuse — an escrow
// blob may be the ONLY remaining path to a customer's backup keys).
var ErrHostEscrowPresent = errors.New("host has key escrow; deletion requires the explicit escrow acknowledgement")
// HostArtifacts summarizes what a host deletion would remove — counts/booleans ONLY (the
// impact preview must never carry a secret or blob).
type HostArtifacts struct {
Guests int
Reports int
LogBundles int // log_bundles rows with scope_id == host_id (the agent channel ONLY)
EscrowPresent bool
WGPeerBound bool
PBSSecretPresent bool
RecoveryPresent bool
}
// CountHostArtifacts reports the per-table blast radius of deleting a host (v0.47.0 stale
// host removal). LogBundles counts ONLY host-scoped rows — customer-scoped bundles (the
// controller channel, scope_id == customer_id) belong to the customer and are never touched.
func (s *Store) CountHostArtifacts(hostID string) (HostArtifacts, error) {
var a HostArtifacts
counts := []struct {
dst *int
query string
}{
{&a.Guests, `SELECT COUNT(*) FROM guests WHERE host_id = ?`},
{&a.Reports, `SELECT COUNT(*) FROM host_reports WHERE host_id = ?`},
{&a.LogBundles, `SELECT COUNT(*) FROM log_bundles WHERE scope_id = ?`},
}
for _, c := range counts {
if err := s.db.QueryRow(c.query, hostID).Scan(c.dst); err != nil {
return a, err
}
}
flags := []struct {
dst *bool
query string
}{
{&a.EscrowPresent, `SELECT EXISTS(SELECT 1 FROM host_escrow WHERE host_id = ?)`},
{&a.WGPeerBound, `SELECT EXISTS(SELECT 1 FROM wg_peers WHERE host_id = ?)`},
{&a.PBSSecretPresent, `SELECT EXISTS(SELECT 1 FROM host_pbs_secrets WHERE host_id = ?)`},
{&a.RecoveryPresent, `SELECT EXISTS(SELECT 1 FROM host_recovery WHERE host_id = ?)`},
}
for _, f := range flags {
var n int
if err := s.db.QueryRow(f.query, hostID).Scan(&n); err != nil {
return a, err
}
*f.dst = n != 0
}
return a, nil
}
// DeleteHost removes a host and every host-scoped artifact in ONE transaction (v0.47.0
// stale host removal). The online-gate lives in the web handler — the store deletes what
// it is told to. Guards:
// - empty hostID → refused (would DELETE the '' scope rows);
// - escrow present without deleteEscrow → ErrHostEscrowPresent, the tx never starts.
//
// The wg_peers delete is INSIDE the tx on purpose — a crash between a host delete and a
// separate peer delete would strand a bound peer the reconciler keeps pushing. The wgsync
// reconciler's 5-minute declarative full-list push converges the endpoint after the row
// disappears — no bump, no reconciler change. log_bundle rows die by scope_id == host_id
// (agent channel); customer-scoped bundles (scope_id == customer_id) are NOT touched.
//
// v0.53.0 (F-14 provenance): every delete also writes a host_deletions row IN THE SAME tx.
// escrow_acked = deleteEscrow AND an escrow row was actually present — "removed through the
// escrow-ack flow" means the operator acknowledged the host removal and the current escrow blob was
// DEMOTED to retained custody (v0.60.1: moved into host_escrow_superseded, not destroyed), not
// merely that the checkbox was ticked over nothing. The flag's F-14 gate semantics are unchanged.
func (s *Store) DeleteHost(hostID string, deleteEscrow bool) error {
if hostID == "" {
return fmt.Errorf("DeleteHost: empty host_id")
}
var escrowPresent int
if err := s.db.QueryRow(`SELECT EXISTS(SELECT 1 FROM host_escrow WHERE host_id = ?)`, hostID).Scan(&escrowPresent); err != nil {
return fmt.Errorf("DeleteHost %s: escrow check: %w", hostID, err)
}
if !deleteEscrow && escrowPresent != 0 {
return ErrHostEscrowPresent
}
tx, err := s.db.Begin()
if err != nil {
return fmt.Errorf("DeleteHost %s: begin: %w", hostID, err)
}
defer tx.Rollback()
// Provenance first (reads the host row this tx is about to delete). A host_id that has no
// row deletes nothing anyway — skip the record rather than inventing an empty customer_id.
var customerID string
switch err := tx.QueryRow(`SELECT customer_id FROM hosts WHERE host_id = ?`, hostID).Scan(&customerID); err {
case nil:
acked := 0
if deleteEscrow && escrowPresent != 0 {
acked = 1
}
if _, err := tx.Exec(`INSERT INTO host_deletions (host_id, customer_id, escrow_acked) VALUES (?, ?, ?)`,
hostID, customerID, acked); err != nil {
return fmt.Errorf("DeleteHost %s: provenance record: %w", hostID, err)
}
case sql.ErrNoRows:
// no host row — fall through, the deletes below are no-ops
default:
return fmt.Errorf("DeleteHost %s: customer lookup: %w", hostID, err)
}
// v0.60.1: host deletion is a LIFECYCLE event — the current escrow blob is DEMOTED to retained
// custody (copied into host_escrow_superseded, copy-BEFORE-delete in this same tx), NEVER
// destroyed; existing superseded rows are spared. No operator path through host lifecycle can
// lose a blob. The customer Danger-zone Delete is the one true purge point (deleteCustomer).
if deleteEscrow {
if _, derr := demoteCurrentEscrowTx(tx, hostID); derr != nil {
return fmt.Errorf("DeleteHost %s: demote escrow to retained custody: %w", hostID, derr)
}
if _, derr := tx.Exec(`DELETE FROM host_escrow WHERE host_id = ?`, hostID); derr != nil {
return fmt.Errorf("DeleteHost %s: remove current escrow row: %w", hostID, derr)
}
}
stmts := []string{
`DELETE FROM guests WHERE host_id = ?`,
`DELETE FROM host_reports WHERE host_id = ?`,
`DELETE FROM signed_jobs WHERE host_id = ?`,
`DELETE FROM host_recovery WHERE host_id = ?`,
`DELETE FROM host_pbs_secrets WHERE host_id = ?`,
`DELETE FROM log_bundle_requests WHERE scope_id = ?`,
`DELETE FROM log_bundles WHERE scope_id = ?`,
`DELETE FROM wg_peers WHERE host_id = ?`,
`DELETE FROM hosts WHERE host_id = ?`,
}
for _, q := range stmts {
if _, err := tx.Exec(q, hostID); err != nil {
return fmt.Errorf("DeleteHost %s: %q: %w", hostID, q, err)
}
}
return tx.Commit()
}
// HostDeletion is one host-removal provenance record (v0.53.0, F-14). EscrowAcked means the
// operator removed the host through the escrow-ack flow — the current key custody was DEMOTED to
// retained custody (v0.60.1), not destroyed — the ONLY state that permits the PBS-DR auto-re-issue.
type HostDeletion struct {
HostID string
CustomerID string
DeletedAt time.Time
EscrowAcked bool
}
// LatestHostDeletion returns the customer's MOST RECENT host-deletion record (nil when the
// customer has none — every pre-v0.53.0 deletion, by design: no backfill invents provenance).
// The latest record is the one that orphaned a surviving ep0 tenancy, so the F-14 gate reads
// exactly this row — an older acked record must not whitelist a newer un-acked deletion.
func (s *Store) LatestHostDeletion(customerID string) (*HostDeletion, error) {
var d HostDeletion
var deletedAt string
var acked int
err := s.db.QueryRow(`
SELECT host_id, customer_id, deleted_at, escrow_acked
FROM host_deletions WHERE customer_id = ?
ORDER BY id DESC LIMIT 1`, customerID,
).Scan(&d.HostID, &d.CustomerID, &deletedAt, &acked)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
d.DeletedAt = parseSQLiteTime(deletedAt)
d.EscrowAcked = acked != 0
return &d, nil
}
// UpsertHost creates or updates a host identity (used by the admin mint). On
// conflict it updates only operator-settable identity fields + updated_at; it does
// NOT touch the reality columns (agent_version/last_report_at) or the inert intent
// columns (desired_*/dr_record_json) — those are owned elsewhere.
func (s *Store) UpsertHost(h *Host) error {
_, err := s.db.Exec(`
INSERT INTO hosts (host_id, customer_id, api_key, updated_at)
VALUES (?, ?, ?, datetime('now'))
ON CONFLICT(host_id) DO UPDATE SET
customer_id = excluded.customer_id,
api_key = excluded.api_key,
updated_at = datetime('now')`,
h.HostID, h.CustomerID, h.APIKey,
)
return err
}
// HostEscrow is the opaque R-wrapped escrow blob stored for a host (doc 03 §8a). Blob is
// ciphertext the hub cannot open.
type HostEscrow struct {
HostID string
Blob []byte
KeyFingerprint string
Posture string
CreatedAt string
UpdatedAt string
// ResticPwSHA256 (SLICE 3) — the non-reversible hash of the offsite repo password the identity blob
// covers ("" = legacy/password-less blob). Safe to store/serve; the password itself never reaches the hub.
ResticPwSHA256 string
// StaleAt (v0.57.0, 2.3) — non-empty when the offsite password was re-issued after this blob was
// sealed: the blob is stale (seals a password that no longer opens the repo). Cleared by a fresh ceremony.
StaleAt string
}
// SaveHostEscrow stores (last-write-wins) the OPAQUE escrow blob for a host. The hub keeps the
// bytes and NEVER decrypts them — there is no decrypt path. createdAt is the agent's timestamp.
// resticPwSHA256 is "" when the ceremony sealed no staged password (stored as-is; never auto-confirms).
// SaveHostEscrow returns superseded=true when it RETAINED a different-passphrase old blob into
// host_escrow_superseded before overwriting the current row (Part B, v0.60.0). A same-sha re-upload
// (idempotent re-ceremony of the same password) refreshes the current row and does NOT create a
// superseded row.
// demoteCurrentEscrowTx copies the host's CURRENT host_escrow row (if any) into
// host_escrow_superseded as a retained blob, inside the given tx. This is THE ONE escrow row-copy
// routine (v0.60.0): SaveHostEscrow uses it to retain a superseded different-passphrase blob before
// overwriting, and DeleteHost (v0.60.1) uses it to DEMOTE the current blob to retained custody
// instead of destroying it. Returns the number of rows copied (0 when the host has no current row).
// The hub never decrypts; custody is unchanged.
func demoteCurrentEscrowTx(tx *sql.Tx, hostID string) (int64, error) {
res, err := tx.Exec(`
INSERT INTO host_escrow_superseded (host_id, blob, key_fingerprint, posture, created_at, restic_pw_sha256, superseded_at)
SELECT host_id, blob, key_fingerprint, posture, created_at, COALESCE(restic_pw_sha256, ''), datetime('now')
FROM host_escrow WHERE host_id = ?`, hostID)
if err != nil {
return 0, err
}
return res.RowsAffected()
}
func (s *Store) SaveHostEscrow(hostID string, blob []byte, keyFingerprint, posture, createdAt, resticPwSHA256 string) (superseded bool, err error) {
tx, err := s.db.Begin()
if err != nil {
return false, err
}
defer func() {
if err != nil {
tx.Rollback()
}
}()
// Retain the current row iff it exists AND seals a DIFFERENT restic password (the incident: a
// recreated volume mints a new passphrase; the old must stay recoverable with its recovery code).
var curSHA string
var exists bool
switch scanErr := tx.QueryRow(`SELECT COALESCE(restic_pw_sha256,'') FROM host_escrow WHERE host_id = ?`, hostID).Scan(&curSHA); scanErr {
case nil:
exists = true
case sql.ErrNoRows:
exists = false
default:
err = scanErr
return false, err
}
if exists && curSHA != resticPwSHA256 {
if _, err = demoteCurrentEscrowTx(tx, hostID); err != nil {
return false, err
}
superseded = true
}
if _, err = tx.Exec(`
INSERT INTO host_escrow (host_id, blob, key_fingerprint, posture, created_at, restic_pw_sha256, updated_at)
VALUES (?, ?, ?, ?, ?, ?, datetime('now'))
ON CONFLICT(host_id) DO UPDATE SET
blob = excluded.blob,
key_fingerprint = excluded.key_fingerprint,
posture = excluded.posture,
created_at = excluded.created_at,
restic_pw_sha256 = excluded.restic_pw_sha256,
stale_at = NULL,
updated_at = datetime('now')`,
hostID, blob, keyFingerprint, posture, createdAt, resticPwSHA256); err != nil {
return false, err
}
err = tx.Commit()
return superseded, err
}
// CountSupersededEscrow returns how many retained (superseded) escrow blobs the hub holds for a host
// (the operator "N superseded escrow blobs retained" surface). 0 on no rows.
func (s *Store) CountSupersededEscrow(hostID string) (int, error) {
var n int
err := s.db.QueryRow(`SELECT COUNT(*) FROM host_escrow_superseded WHERE host_id = ?`, hostID).Scan(&n)
return n, err
}
// ListSupersededEscrow returns the retained (superseded) escrow blobs for a host, newest-superseded
// first. Opaque bytes — the hub never decrypts. Seeds the future guided-recovery flow (R-26).
func (s *Store) ListSupersededEscrow(hostID string) ([]HostEscrow, error) {
rows, err := s.db.Query(`
SELECT host_id, blob, key_fingerprint, posture, created_at, restic_pw_sha256, superseded_at
FROM host_escrow_superseded WHERE host_id = ? ORDER BY id DESC`, hostID)
if err != nil {
return nil, err
}
defer rows.Close()
var out []HostEscrow
for rows.Next() {
var e HostEscrow
if err := rows.Scan(&e.HostID, &e.Blob, &e.KeyFingerprint, &e.Posture, &e.CreatedAt, &e.ResticPwSHA256, &e.UpdatedAt); err != nil {
return nil, err
}
out = append(out, e)
}
return out, rows.Err()
}
// MarkEscrowStale flags a host's escrow blob as stale (v0.57.0, 2.3) — called when the offsite repo
// password is re-issued, because the blob then seals a password that no longer opens the repo. No-op
// when no escrow row exists; idempotent (only stamps the first re-issue since the last ceremony; a
// fresh ceremony clears stale_at via SaveHostEscrow's ON CONFLICT).
func (s *Store) MarkEscrowStale(hostID string) error {
_, err := s.db.Exec(`UPDATE host_escrow SET stale_at = datetime('now') WHERE host_id = ? AND stale_at IS NULL`, hostID)
return err
}
// GetHostEscrow returns the stored opaque escrow for a host (nil if none). Used by tests and
// (future, slice 10) restore-mode serving. The hub returns bytes verbatim; it never decrypts.
func (s *Store) GetHostEscrow(hostID string) (*HostEscrow, error) {
var e HostEscrow
err := s.db.QueryRow(`
SELECT host_id, blob, key_fingerprint, posture, created_at, updated_at, COALESCE(restic_pw_sha256, ''), COALESCE(stale_at, '')
FROM host_escrow WHERE host_id = ?`, hostID).
Scan(&e.HostID, &e.Blob, &e.KeyFingerprint, &e.Posture, &e.CreatedAt, &e.UpdatedAt, &e.ResticPwSHA256, &e.StaleAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
return &e, nil
}
// EscrowStatus (SLICE 3) is the non-secret escrow summary served in the report ACK so the controller can
// VERIFY-and-auto-confirm: not "a blob exists" but "the blob covers the CURRENT repo password" (hash match).
type EscrowStatus struct {
IdentityBlobPresent bool `json:"identity_blob_present"`
ResticPwSHA256 string `json:"restic_pw_sha256,omitempty"`
CreatedAt string `json:"created_at,omitempty"`
// Stale (v0.57.0, 2.3) — true when the offsite password was re-issued after the blob was sealed.
// When stale the ResticPwSHA256 is WITHHELD (emptied) so the controller cannot auto-confirm against
// a hash that no longer matches the live repo password — the ceremony must run again.
Stale bool `json:"escrow_stale,omitempty"`
}
// GetEscrowStatusForCustomer returns the escrow status of the customer's host (nil if the customer has no
// escrow row). With multiple hosts (not the current model), the most recently updated escrow wins.
func (s *Store) GetEscrowStatusForCustomer(customerID string) (*EscrowStatus, error) {
var st EscrowStatus
var identityPresent int
var staleAt string
err := s.db.QueryRow(`
SELECT (e.identity_blob IS NOT NULL), COALESCE(e.restic_pw_sha256, ''), e.created_at, COALESCE(e.stale_at, '')
FROM host_escrow e JOIN hosts h ON h.host_id = e.host_id
WHERE h.customer_id = ?
ORDER BY e.updated_at DESC LIMIT 1`, customerID).
Scan(&identityPresent, &st.ResticPwSHA256, &st.CreatedAt, &staleAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
st.IdentityBlobPresent = identityPresent == 1
// v0.57.0 (2.3): a stale blob must NOT auto-confirm — withhold the hash and flag it so the
// controller stays pending and the escrow wizard is offered again.
if staleAt != "" {
st.Stale = true
st.ResticPwSHA256 = ""
}
return &st, nil
}
// SetHostDesired sets a host's desired-state JSON and ATOMICALLY bumps its desired_generation
// (slice 10A — the operator "admin-set" write). Returns the NEW generation. The generation is
// the cheap change-signal carried on every heartbeat envelope; the agent re-fetches the full
// desired-state only when it advances. Errors with sql.ErrNoRows if the host does not exist.
func (s *Store) SetHostDesired(hostID string, desiredJSON []byte) (int64, error) {
res, err := s.db.Exec(`
UPDATE hosts SET desired_json = ?, desired_generation = desired_generation + 1,
updated_at = datetime('now')
WHERE host_id = ?`, string(desiredJSON), hostID)
if err != nil {
return 0, err
}
if n, _ := res.RowsAffected(); n == 0 {
return 0, sql.ErrNoRows // unknown host
}
var gen int64
if err := s.db.QueryRow(`SELECT desired_generation FROM hosts WHERE host_id = ?`, hostID).Scan(&gen); err != nil {
return 0, err
}
return gen, nil
}
// BumpHostDesired advances a host's desired_generation WITHOUT touching desired_json (S2). Used
// when hub-OWNED served state changes (the merge-at-read wireguard block) — the stored operator
// blob is not the thing that moved, so SetHostDesired (which replaces it) must not be used.
// Returns the new generation; sql.ErrNoRows for an unknown host.
func (s *Store) BumpHostDesired(hostID string) (int64, error) {
res, err := s.db.Exec(`
UPDATE hosts SET desired_generation = desired_generation + 1, updated_at = datetime('now')
WHERE host_id = ?`, hostID)
if err != nil {
return 0, err
}
if n, _ := res.RowsAffected(); n == 0 {
return 0, sql.ErrNoRows // unknown host
}
var gen int64
if err := s.db.QueryRow(`SELECT desired_generation FROM hosts WHERE host_id = ?`, hostID).Scan(&gen); err != nil {
return 0, err
}
return gen, nil
}
// BumpAllHostGenerations advances EVERY host's desired_generation (TASK H1). Used when a FLEET-WIDE
// hub-owned served-state value changes — the operator OOB peer's /32 flows into every host's
// merge-at-read wireguard block (oob_peer_ip), so every agent must re-fetch + re-render. Returns the
// number of hosts bumped.
func (s *Store) BumpAllHostGenerations() (int64, error) {
res, err := s.db.Exec(`UPDATE hosts SET desired_generation = desired_generation + 1, updated_at = datetime('now')`)
if err != nil {
return 0, err
}
n, _ := res.RowsAffected()
return n, nil
}
// SignedJob is one OPAQUE operator-signed destructive-op blob queued for a host (slice 10A). The
// hub stores + serves the bytes; it never forges, opens, or executes them (10B owns verify+run).
type SignedJob struct {
JobID string
Blob []byte
CreatedAt string
}
// EnqueueSignedJob appends an opaque signed-op blob to a host's queue (slice 10A). Operator-side;
// the hub holds no signing key — the blob arrives pre-signed.
func (s *Store) EnqueueSignedJob(hostID, jobID string, blob []byte) error {
_, err := s.db.Exec(`INSERT INTO signed_jobs (host_id, job_id, blob) VALUES (?, ?, ?)`,
hostID, jobID, blob)
return err
}
// GetSignedJobs returns a host's pending signed-op blobs, oldest first (slice 10A serving).
func (s *Store) GetSignedJobs(hostID string) ([]SignedJob, error) {
rows, err := s.db.Query(`SELECT job_id, blob, created_at FROM signed_jobs WHERE host_id = ? ORDER BY id ASC`, hostID)
if err != nil {
return nil, err
}
defer rows.Close()
var jobs []SignedJob
for rows.Next() {
var j SignedJob
if err := rows.Scan(&j.JobID, &j.Blob, &j.CreatedAt); err != nil {
return nil, err
}
jobs = append(jobs, j)
}
return jobs, rows.Err()
}
// CountSignedJobs returns the number of pending signed-op blobs for a host (drives the envelope's
// has_signed_ops flag — the cheap "fetch your jobs" notification).
func (s *Store) CountSignedJobs(hostID string) (int, error) {
var n int
err := s.db.QueryRow(`SELECT COUNT(*) FROM signed_jobs WHERE host_id = ?`, hostID).Scan(&n)
return n, err
}
// DeleteSignedJob removes a processed job from a host's queue (slice 10B completion). The agent
// calls it after executing OR terminally rejecting a job. Idempotent (deleting an absent job is a
// no-op, returns nil) — a retried completion must not error.
func (s *Store) DeleteSignedJob(hostID, jobID string) error {
_, err := s.db.Exec(`DELETE FROM signed_jobs WHERE host_id = ? AND job_id = ?`, hostID, jobID)
return err
}
// ---- slice 10D: DR capstone (recovery mode, DR bundle, re-enroll) ----------------------------
// SetRecoveryMode arms recovery mode for a host until `until` (the operator toggle; bounded
// auto-expiry). While active, the hub serves the restore directive + allows re-enroll. Errors
// ErrNoRows for an unknown host.
func (s *Store) SetRecoveryMode(hostID string, until time.Time) error {
res, err := s.db.Exec(`UPDATE hosts SET recovery_mode_until = ?, updated_at = datetime('now') WHERE host_id = ?`,
until.UTC().Format("2006-01-02 15:04:05"), hostID)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return sql.ErrNoRows
}
return nil
}
// ClearRecoveryMode disables recovery mode (operator confirm, or after re-enroll completes).
func (s *Store) ClearRecoveryMode(hostID string) error {
_, err := s.db.Exec(`UPDATE hosts SET recovery_mode_until = NULL, updated_at = datetime('now') WHERE host_id = ?`, hostID)
return err
}
// RotateHostAPIKey replaces a host's API key (the re-enroll credential rotation — the old box's hub
// access is revoked the instant this commits; purely hub-internal, no Cloudflare/PBS write needed).
func (s *Store) RotateHostAPIKey(hostID, newAPIKey string) error {
res, err := s.db.Exec(`UPDATE hosts SET api_key = ?, updated_at = datetime('now') WHERE host_id = ?`, newAPIKey, hostID)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return sql.ErrNoRows
}
return nil
}
// SaveHostDRBundle stores the IDENTITY escrow blob + the NON-secret DR directive alongside the
// existing K-escrow blob (slice 10D.1). The K-escrow row must already exist (slice-7 escrow upload);
// this updates the additive 10D columns. The hub holds only ciphertext + non-secret directive.
func (s *Store) SaveHostDRBundle(hostID string, identityBlob []byte, directiveJSON string) error {
if directiveJSON == "" {
directiveJSON = "{}"
}
res, err := s.db.Exec(`UPDATE host_escrow SET identity_blob = ?, directive_json = ?, updated_at = datetime('now') WHERE host_id = ?`,
identityBlob, directiveJSON, hostID)
if err != nil {
return err
}
if n, _ := res.RowsAffected(); n == 0 {
return sql.ErrNoRows // no K-escrow row yet — upload the escrow first
}
return nil
}
// HostDRBundle is the full DR directive served to a re-enrolling box (slice 10D): the two OPAQUE
// escrow blobs (K + identity — useless without R) + the non-secret directive fields.
type HostDRBundle struct {
KEscrowBlob []byte
IdentityBlob []byte
DirectiveJSON string
}
// GetHostDRBundle returns a host's DR bundle (nil if no escrow row). The blobs are opaque — the hub
// cannot open them (it has no R).
func (s *Store) GetHostDRBundle(hostID string) (*HostDRBundle, error) {
var b HostDRBundle
var directive sql.NullString
err := s.db.QueryRow(`SELECT blob, identity_blob, directive_json FROM host_escrow WHERE host_id = ?`, hostID).
Scan(&b.KEscrowBlob, &b.IdentityBlob, &directive)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
if directive.Valid {
b.DirectiveJSON = directive.String
}
return &b, nil
}
// SaveHostReport inserts a host_reports row and bumps the host's reality columns
// (agent_version/last_report_at/updated_at) — never the inert intent columns.
func (s *Store) SaveHostReport(hostID, customerID string, reportJSON []byte, d HostReportDenorm) error {
_, err := s.db.Exec(`
INSERT INTO host_reports (host_id, customer_id, report_json, agent_version,
cpu_percent, memory_percent, disk_percent, guest_total, guest_running, cloudflared_status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
hostID, customerID, string(reportJSON), d.AgentVersion,
d.CPUPercent, d.MemoryPercent, d.DiskPercent, d.GuestTotal, d.GuestRunning, d.CloudflaredStatus,
)
if err != nil {
return err
}
_, err = s.db.Exec(`
UPDATE hosts SET agent_version = ?, last_report_at = datetime('now'), updated_at = datetime('now')
WHERE host_id = ?`, d.AgentVersion, hostID)
return err
}
// GetLatestHostReportJSON returns the most recent host-report payload (report_json)
// for a customer, by received_at, or ("", nil) if the customer has no host-report.
// The backup deadline check uses it to read the agent's own backup reality
// (pbs_snapshots + vzdump backups) — the authoritative offsite-backup signal
// post-slice-8C, replacing the no-longer-emitted backup_completed event.
func (s *Store) GetLatestHostReportJSON(customerID string) (string, error) {
var j string
err := s.db.QueryRow(
`SELECT report_json FROM host_reports WHERE customer_id = ? ORDER BY received_at DESC LIMIT 1`,
customerID,
).Scan(&j)
if err == sql.ErrNoRows {
return "", nil
}
if err != nil {
return "", err
}
return j, nil
}
// UpsertGuestFromReport upserts the REALITY columns of a guest. On conflict it
// must NOT clobber the inert columns (api_key / desired_spec_json).
func (s *Store) UpsertGuestFromReport(g *Guest) error {
_, err := s.db.Exec(`
INSERT INTO guests (guest_id, customer_id, host_id, vmid, display_name, status,
controller_version, last_seen_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))
ON CONFLICT(guest_id) DO UPDATE SET
vmid = excluded.vmid,
display_name = excluded.display_name,
status = excluded.status,
controller_version = excluded.controller_version,
last_seen_at = datetime('now'),
updated_at = datetime('now')`,
g.GuestID, g.CustomerID, g.HostID, g.VMID, g.DisplayName, g.Status,
g.ControllerVersion,
)
return err
}
// guestRealitySelectCols are the report-driven reality columns (plus identity/timestamps)
// of a guest. It deliberately OMITS the secret/inert columns (api_key, desired_spec_json):
// the read-only Hosts view never renders them, so they are not selected.
const guestRealitySelectCols = `guest_id, customer_id, host_id, vmid, display_name, status,
controller_version, last_seen_at, created_at, updated_at`
func scanGuest(scan func(dest ...any) error) (*Guest, error) {
var g Guest
var lastSeen sql.NullString
var createdAt, updatedAt string
err := scan(&g.GuestID, &g.CustomerID, &g.HostID, &g.VMID, &g.DisplayName, &g.Status,
&g.ControllerVersion, &lastSeen, &createdAt, &updatedAt)
if err != nil {
return nil, err
}
if lastSeen.Valid && lastSeen.String != "" {
t := parseSQLiteTime(lastSeen.String)
g.LastSeenAt = &t
}
g.CreatedAt = parseSQLiteTime(createdAt)
g.UpdatedAt = parseSQLiteTime(updatedAt)
return &g, nil
}
// ListGuestsForHost returns the guests (controller LXCs) enrolled on a host, ordered by
// vmid. Reads only reality columns (no api_key / desired_spec_json). Returns an empty
// slice (never nil-error) when the host has no guests — the read-only Hosts detail view.
func (s *Store) ListGuestsForHost(hostID string) ([]Guest, error) {
rows, err := s.db.Query(`SELECT `+guestRealitySelectCols+` FROM guests WHERE host_id = ? ORDER BY vmid`, hostID)
if err != nil {
return nil, err
}
defer rows.Close()
guests := []Guest{}
for rows.Next() {
g, err := scanGuest(rows.Scan)
if err != nil {
return nil, err
}
guests = append(guests, *g)
}
return guests, rows.Err()
}
// GetHostStaleness returns per-host recency for the dead-man's-switch. Hosts that
// have never reported (NULL last_report_at) are skipped — a freshly-minted host is
// not "down" until it has checked in at least once.
func (s *Store) GetHostStaleness() ([]HostStaleRow, error) {
rows, err := s.db.Query(`SELECT host_id, customer_id, last_report_at FROM hosts WHERE last_report_at IS NOT NULL`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []HostStaleRow
for rows.Next() {
var r HostStaleRow
var last string
if err := rows.Scan(&r.HostID, &r.CustomerID, &last); err != nil {
return nil, err
}
r.LastReportAt = parseSQLiteTime(last)
out = append(out, r)
}
return out, rows.Err()
}
// CapabilityStatus mirrors the agent's capability.Status wire shape (felhom-agent v0.44.0): one
// privileged `sudo -n` grant the non-root agent depends on, and whether it is currently usable.
type CapabilityStatus struct {
Name string `json:"name"`
Feature string `json:"feature"`
Critical bool `json:"critical"`
Status string `json:"status"` // "ok" | "degraded"
Reason string `json:"reason,omitempty"`
}
// HostCapabilityRow is the per-host capability snapshot the HostCapabilityChecker reads — extracted
// from the latest host-report's report_json (no dedicated column; the array rides the report body).
type HostCapabilityRow struct {
HostID string
CustomerID string
Capabilities []CapabilityStatus
}
// GetHostCapabilities returns the latest capability snapshot per host (from the most recent
// host_reports row). Hosts whose latest report carries no capabilities array (a pre-v0.44.0 agent)
// yield an empty slice — the checker treats that as "ok/unknown" and never alerts, so an old agent
// can't trip a false degraded.
func (s *Store) GetHostCapabilities() ([]HostCapabilityRow, error) {
rows, err := s.db.Query(`
SELECT hr.host_id, hr.customer_id, hr.report_json
FROM host_reports hr
JOIN (SELECT host_id, MAX(id) AS mx FROM host_reports GROUP BY host_id) latest
ON hr.id = latest.mx`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []HostCapabilityRow
for rows.Next() {
var r HostCapabilityRow
var reportJSON string
if err := rows.Scan(&r.HostID, &r.CustomerID, &reportJSON); err != nil {
return nil, err
}
var body struct {
Capabilities []CapabilityStatus `json:"capabilities"`
}
_ = json.Unmarshal([]byte(reportJSON), &body) // a malformed/old body → nil caps → no alert
r.Capabilities = body.Capabilities
out = append(out, r)
}
return out, rows.Err()
}
// HostLeafRow is the per-host served-leaf-fingerprint the HostLeafChecker reads — extracted from the
// latest host-report's report_json (no dedicated column; the fp rides the report body, same as the
// capabilities snapshot). LeafFP is "" for a pre-v0.48.0 agent / local-API-disabled host.
type HostLeafRow struct {
HostID string
CustomerID string
LeafFP string
}
// HostDiskRow is the per-host root-filesystem usage the HostDiskChecker reads: the denormalized
// disk_percent column (the threshold signal) plus total/used bytes parsed from report_json (event detail
// only). DiskPercent is the HOST root fs (agent HostMetrics) — distinct from the controller's GUEST cgroup
// disk. A NULL/absent disk_percent (a pre-disk-reporting agent) yields 0 → the checker treats it as ok.
type HostDiskRow struct {
HostID string
CustomerID string
DiskPercent float64
DiskTotalBytes int64
DiskUsedBytes int64
}
// GetHostDiskUsage returns the latest root-fs usage per host (MAX(id) per host, mirroring
// GetHostCapabilities / GetHostLeafFingerprints). disk_percent comes from the denorm column; total/used
// bytes are parsed from the report body's host block for the event detail (no schema migration needed).
func (s *Store) GetHostDiskUsage() ([]HostDiskRow, error) {
rows, err := s.db.Query(`
SELECT hr.host_id, hr.customer_id, hr.disk_percent, hr.report_json
FROM host_reports hr
JOIN (SELECT host_id, MAX(id) AS mx FROM host_reports GROUP BY host_id) latest
ON hr.id = latest.mx`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []HostDiskRow
for rows.Next() {
var r HostDiskRow
var dp sql.NullFloat64
var reportJSON string
if err := rows.Scan(&r.HostID, &r.CustomerID, &dp, &reportJSON); err != nil {
return nil, err
}
r.DiskPercent = dp.Float64
var body struct {
Host struct {
DiskTotalBytes int64 `json:"disk_total_bytes"`
DiskUsedBytes int64 `json:"disk_used_bytes"`
} `json:"host"`
}
_ = json.Unmarshal([]byte(reportJSON), &body) // malformed/old body → zero bytes (detail only)
r.DiskTotalBytes = body.Host.DiskTotalBytes
r.DiskUsedBytes = body.Host.DiskUsedBytes
out = append(out, r)
}
return out, rows.Err()
}
// HostStorageTargetRow is one (host, storage target) fill observation, parsed from the latest report's
// storage_targets[] (no denorm column — modeled on GetHostDiskUsage's report_json parse). Percent is
// used_fraction×100 (the agent reports a 0..1 fraction). A target with no usable fraction yields 0 (ok).
type HostStorageTargetRow struct {
HostID string
CustomerID string
Name string // Proxmox storage id
Type string // local | local-dir | lvmthin | usb | nfs | cifs | pbs
MountPath string // host mountpoint ("" for network/lvm/root-backed dir)
Percent float64 // 0..100 (used_fraction × 100)
TotalBytes int64
UsedBytes int64
}
// GetHostStorageTargets returns the per-storage fill of every host's LATEST report (MAX(id) per host),
// parsed from report_json.storage_targets[] (mirrors GetHostDiskUsage). The StorageFillChecker keys on
// (host, name) and excludes the root-backed builtin — see monitor/storage_fill.go.
func (s *Store) GetHostStorageTargets() ([]HostStorageTargetRow, error) {
rows, err := s.db.Query(`
SELECT hr.host_id, hr.customer_id, hr.report_json
FROM host_reports hr
JOIN (SELECT host_id, MAX(id) AS mx FROM host_reports GROUP BY host_id) latest
ON hr.id = latest.mx`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []HostStorageTargetRow
for rows.Next() {
var hostID, customerID, reportJSON string
if err := rows.Scan(&hostID, &customerID, &reportJSON); err != nil {
return nil, err
}
var body struct {
StorageTargets []struct {
Name string `json:"name"`
Type string `json:"type"`
MountPath string `json:"mount_path"`
UsedFraction float64 `json:"used_fraction"`
TotalBytes int64 `json:"total_bytes"`
UsedBytes int64 `json:"used_bytes"`
} `json:"storage_targets"`
}
_ = json.Unmarshal([]byte(reportJSON), &body) // malformed/old body → no targets (never a false alert)
for _, t := range body.StorageTargets {
out = append(out, HostStorageTargetRow{
HostID: hostID, CustomerID: customerID,
Name: t.Name, Type: t.Type, MountPath: t.MountPath,
Percent: t.UsedFraction * 100, TotalBytes: t.TotalBytes, UsedBytes: t.UsedBytes,
})
}
}
return out, rows.Err()
}
// GetHostLeafFingerprints returns the latest reported local-API leaf fp per host (mirrors
// GetHostCapabilities — MAX(id) per host, parsed from report_json so there is no schema migration).
func (s *Store) GetHostLeafFingerprints() ([]HostLeafRow, error) {
rows, err := s.db.Query(`
SELECT hr.host_id, hr.customer_id, hr.report_json
FROM host_reports hr
JOIN (SELECT host_id, MAX(id) AS mx FROM host_reports GROUP BY host_id) latest
ON hr.id = latest.mx`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []HostLeafRow
for rows.Next() {
var r HostLeafRow
var reportJSON string
if err := rows.Scan(&r.HostID, &r.CustomerID, &reportJSON); err != nil {
return nil, err
}
var body struct {
LeafFingerprint string `json:"leaf_fingerprint"`
}
_ = json.Unmarshal([]byte(reportJSON), &body) // malformed/old body → "" → no alert
r.LeafFP = body.LeafFingerprint
out = append(out, r)
}
return out, rows.Err()
}
Reference in New Issue View Git Blame Copy Permalink