admin
a3ac6c9488
config_version counter (bumped on every config save) advertised in the report ACK; controller re-pulls + self-restarts on a change. Retire Trigger Update / Push Config / Pull Config / Show Diff handlers+routes+buttons and the inbound geo-notify (keep hub->Cloudflare geo removal). Setup command -> host-install; delete dead customer.html + config_detail.html. Closes AUDIT-hub-gui F-S1/F-S4. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HxLA1mZurFq9kt8hneFeCs
1834 lines
65 KiB
Go
1834 lines
65 KiB
Go
package store
|
||
|
||
import (
|
||
"database/sql"
|
||
"encoding/json"
|
||
"fmt"
|
||
"log"
|
||
"strconv"
|
||
"time"
|
||
|
||
_ "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.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'))
|
||
);
|
||
|
||
-- 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 '{}'`)
|
||
|
||
// 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
|
||
}
|
||
|
||
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()
|
||
}
|
||
|
||
// 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
|
||
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, 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,
|
||
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,
|
||
)
|
||
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, 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.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
|
||
}
|
||
|
||
// 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, 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.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.
|
||
func (s *Store) DeleteCustomerConfig(customerID string) error {
|
||
_, err := s.db.Exec("DELETE FROM customer_configs WHERE customer_id = ?", customerID)
|
||
return err
|
||
}
|
||
|
||
// 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, 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.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
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// 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"`
|
||
}
|
||
|
||
// 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"
|
||
)
|
||
|
||
// 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),
|
||
}
|
||
}
|
||
|
||
// 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
|
||
}
|
||
return s.setSetting(settingArtifactGoldenSHA256, m.GoldenSHA256)
|
||
}
|
||
|
||
// 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()
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// 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()
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// 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.
|
||
func (s *Store) SaveHostEscrow(hostID string, blob []byte, keyFingerprint, posture, createdAt string) error {
|
||
_, err := s.db.Exec(`
|
||
INSERT INTO host_escrow (host_id, blob, key_fingerprint, posture, created_at, 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,
|
||
updated_at = datetime('now')`,
|
||
hostID, blob, keyFingerprint, posture, createdAt,
|
||
)
|
||
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
|
||
FROM host_escrow WHERE host_id = ?`, hostID).
|
||
Scan(&e.HostID, &e.Blob, &e.KeyFingerprint, &e.Posture, &e.CreatedAt, &e.UpdatedAt)
|
||
if err == sql.ErrNoRows {
|
||
return nil, nil
|
||
}
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &e, 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
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// 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()
|
||
}
|