steward/internal/store/store.go

// Package store keeps Steward's data in SQLite: the configuration, leases,
// the history of changes, users, tokens, sessions and the audit log.
//
// The configuration is loaded into a model.State as a whole and saved as
// changes (model.Change); leases are read once at start and written one by
// one as the DHCP server hands them out.
package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	_ "modernc.org/sqlite"
)

// ErrNotFound is returned when a row does not exist.
var ErrNotFound = errors.New("not found")

// ErrConflict is returned for unique constraint violations.
var ErrConflict = errors.New("already exists")

// Store is a Steward database.
type Store struct {
	db *sql.DB
}

// DB exposes the database (backups, tests).
func (s *Store) DB() *sql.DB { return s.db }

// Path is the database file ("" in memory).
func (s *Store) Path() string {
	var file string
	if err := s.db.QueryRow(`SELECT file FROM pragma_database_list WHERE name = 'main'`).Scan(&file); err != nil {
		return ""
	}
	return file
}

// Open opens (creating if needed) the database at path and applies
// migrations. ":memory:" makes a throwaway database.
func Open(path string) (*Store, error) {
	dsn := "file:" + path + "?_pragma=busy_timeout(10000)&_pragma=foreign_keys(1)&_pragma=synchronous(NORMAL)&_txlock=immediate&_pragma=journal_mode(WAL)"
	if path == ":memory:" {
		dsn = "file::memory:?_pragma=foreign_keys(1)&_txlock=immediate"
	}
	db, err := sql.Open("sqlite", dsn)
	if err != nil {
		return nil, err
	}
	if path == ":memory:" {
		db.SetMaxOpenConns(1)
	}
	s := &Store{db: db}
	if err := s.migrate(context.Background()); err != nil {
		db.Close()
		return nil, fmt.Errorf("migrating database: %w", err)
	}
	return s, nil
}

// OpenExisting opens a database without creating or migrating it, for
// commands that run next to a running stewardd ('stewardd backup').
func OpenExisting(path string) (*Store, error) {
	db, err := sql.Open("sqlite", "file:"+path+"?mode=rw&_pragma=busy_timeout(10000)&_pragma=foreign_keys(1)&_txlock=immediate")
	if err != nil {
		return nil, err
	}
	if err := db.Ping(); err != nil {
		db.Close()
		return nil, err
	}
	return &Store{db: db}, nil
}

// Close closes the database.
func (s *Store) Close() error { return s.db.Close() }

// SchemaVersion is the schema version this Steward uses.
func SchemaVersion() int { return len(migrations) }

// Append new migrations; never edit one that has shipped.
var migrations = []string{
	// 1: initial schema
	`
CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE seq (name TEXT PRIMARY KEY, value INTEGER NOT NULL);
INSERT INTO seq (name, value) VALUES ('object', 0);

CREATE TABLE subnets (
	id INTEGER PRIMARY KEY,
	name TEXT NOT NULL DEFAULT '',
	interface TEXT NOT NULL,
	prefix TEXT NOT NULL,
	pools TEXT NOT NULL DEFAULT '[]',
	lease_time INTEGER NOT NULL DEFAULT 0,
	options TEXT NOT NULL DEFAULT '[]',
	disabled INTEGER NOT NULL DEFAULT 0,
	comment TEXT NOT NULL DEFAULT ''
);
CREATE TABLE reservations (
	id INTEGER PRIMARY KEY,
	mac TEXT NOT NULL DEFAULT '',
	client_id TEXT NOT NULL DEFAULT '',
	ip TEXT NOT NULL,
	hostname TEXT NOT NULL DEFAULT '',
	options TEXT NOT NULL DEFAULT '[]',
	comment TEXT NOT NULL DEFAULT ''
);

CREATE TABLE leases (
	ip TEXT PRIMARY KEY,
	subnet_id INTEGER NOT NULL,
	mac TEXT NOT NULL DEFAULT '',
	client_id TEXT NOT NULL DEFAULT '',
	hostname TEXT NOT NULL DEFAULT '',
	state TEXT NOT NULL,
	start TEXT NOT NULL,
	end TEXT NOT NULL,
	interface TEXT NOT NULL DEFAULT '',
	updated TEXT NOT NULL
);
CREATE INDEX leases_mac ON leases (mac);

CREATE TABLE history (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	time TEXT NOT NULL,
	user_id INTEGER,
	username TEXT NOT NULL,
	via TEXT NOT NULL,
	description TEXT NOT NULL,
	changes TEXT NOT NULL,
	revert_of INTEGER,
	reverted_by INTEGER
);

CREATE TABLE users (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	username TEXT NOT NULL UNIQUE COLLATE NOCASE,
	password_hash TEXT NOT NULL DEFAULT '',
	admin INTEGER NOT NULL DEFAULT 0,
	grants TEXT NOT NULL DEFAULT '[]',
	disabled INTEGER NOT NULL DEFAULT 0,
	comment TEXT NOT NULL DEFAULT '',
	created_at TEXT NOT NULL,
	updated_at TEXT NOT NULL,
	last_login_at TEXT
);
CREATE TABLE tokens (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
	name TEXT NOT NULL,
	public_id TEXT NOT NULL UNIQUE,
	secret_hash TEXT NOT NULL,
	grants TEXT,
	expires_at TEXT,
	created_at TEXT NOT NULL,
	last_used_at TEXT,
	last_used_ip TEXT NOT NULL DEFAULT '',
	revoked_at TEXT
);
CREATE TABLE sessions (
	id_hash TEXT PRIMARY KEY,
	user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
	csrf_token TEXT NOT NULL,
	created_at TEXT NOT NULL,
	expires_at TEXT NOT NULL,
	last_seen_at TEXT NOT NULL,
	ip TEXT NOT NULL DEFAULT '',
	user_agent TEXT NOT NULL DEFAULT ''
);
CREATE TABLE audit (
	id INTEGER PRIMARY KEY AUTOINCREMENT,
	time TEXT NOT NULL,
	user_id INTEGER,
	username TEXT NOT NULL DEFAULT '',
	auth TEXT NOT NULL DEFAULT '',
	token_id INTEGER,
	ip TEXT NOT NULL DEFAULT '',
	action TEXT NOT NULL,
	object TEXT NOT NULL DEFAULT '',
	result TEXT NOT NULL,
	detail TEXT NOT NULL DEFAULT ''
);
CREATE INDEX audit_time ON audit (time);
`,
	// 2: token rotation: a new secret, the old one working for a grace period.
	`
ALTER TABLE tokens ADD COLUMN rotated_at TEXT;
ALTER TABLE tokens ADD COLUMN prev_secret_hash TEXT NOT NULL DEFAULT '';
ALTER TABLE tokens ADD COLUMN prev_expires_at TEXT;
`,
	// 3: disabled reservations.
	`
ALTER TABLE reservations ADD COLUMN disabled INTEGER NOT NULL DEFAULT 0;
`,
	// 4: when a lease's client last answered a ping.
	`
ALTER TABLE leases ADD COLUMN last_seen TEXT;
`,
	// 5: the order addresses are handed out in (a pool's own is in pools).
	`
ALTER TABLE subnets ADD COLUMN pool_order TEXT NOT NULL DEFAULT '';
ALTER TABLE subnets ADD COLUMN address_order TEXT NOT NULL DEFAULT '';
`,
}

func (s *Store) migrate(ctx context.Context) error {
	if _, err := s.db.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)`); err != nil {
		return err
	}
	var v int
	err := s.db.QueryRowContext(ctx, `SELECT version FROM schema_version`).Scan(&v)
	if errors.Is(err, sql.ErrNoRows) {
		if _, err := s.db.ExecContext(ctx, `INSERT INTO schema_version (version) VALUES (0)`); err != nil {
			return err
		}
	} else if err != nil {
		return err
	}
	if v > len(migrations) {
		return fmt.Errorf("database schema version %d is newer than this Steward (%d); upgrade Steward", v, len(migrations))
	}
	for i := v; i < len(migrations); i++ {
		tx, err := s.db.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		if _, err := tx.ExecContext(ctx, migrations[i]); err != nil {
			tx.Rollback()
			return fmt.Errorf("migration %d: %w", i+1, err)
		}
		if _, err := tx.ExecContext(ctx, `UPDATE schema_version SET version = ?`, i+1); err != nil {
			tx.Rollback()
			return err
		}
		if err := tx.Commit(); err != nil {
			return err
		}
	}
	return nil
}

// Version returns the database's schema version.
func (s *Store) Version(ctx context.Context) (int, error) {
	var v int
	err := s.db.QueryRowContext(ctx, `SELECT version FROM schema_version`).Scan(&v)
	return v, err
}

// AllocateIDs reserves n object IDs (subnets and reservations share one
// sequence, so an ID names one object).
func (s *Store) AllocateIDs(ctx context.Context, n int) ([]int64, error) {
	if n <= 0 {
		return nil, nil
	}
	var last int64
	if err := s.db.QueryRowContext(ctx, `UPDATE seq SET value = value + ? WHERE name = 'object' RETURNING value`, n).Scan(&last); err != nil {
		return nil, err
	}
	ids := make([]int64, n)
	for i := range ids {
		ids[i] = last - int64(n) + 1 + int64(i)
	}
	return ids, nil
}

// GetMeta reads a meta value ("" when unset).
func (s *Store) GetMeta(ctx context.Context, key string) (string, error) {
	var v string
	err := s.db.QueryRowContext(ctx, `SELECT value FROM meta WHERE key = ?`, key).Scan(&v)
	if errors.Is(err, sql.ErrNoRows) {
		return "", nil
	}
	return v, err
}

// SetMeta writes a meta value.
func (s *Store) SetMeta(ctx context.Context, key, value string) error {
	_, err := s.db.ExecContext(ctx, `INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value`, key, value)
	return err
}

// Backup writes a consistent copy of the database to path.
func (s *Store) Backup(ctx context.Context, path string) error {
	_, err := s.db.ExecContext(ctx, `VACUUM INTO ?`, path)
	return err
}

// ---- helpers

const tsLayout = "2006-01-02T15:04:05.000000000Z07:00"

func ts(t time.Time) string { return t.UTC().Format(tsLayout) }

func tsPtr(t *time.Time) any {
	if t == nil {
		return nil
	}
	return ts(*t)
}

func parseTS(s string) time.Time {
	t, err := time.Parse(time.RFC3339Nano, s)
	if err != nil {
		return time.Time{}
	}
	return t
}

func parseTSPtr(s sql.NullString) *time.Time {
	if !s.Valid || s.String == "" {
		return nil
	}
	t := parseTS(s.String)
	return &t
}

func boolInt(b bool) int {
	if b {
		return 1
	}
	return 0
}

func jsonText(v any) string {
	b, err := json.Marshal(v)
	if err != nil {
		panic(err)
	}
	return string(b)
}

func isUnique(err error) bool {
	return err != nil && strings.Contains(err.Error(), "UNIQUE constraint failed")
}