276 lines
8.8 KiB
Go
276 lines
8.8 KiB
Go
// Package store hält den persistenten Zustand des Sultan-Servers (Mandanten,
|
||
// Knoten, Enrollment-Tokens) in SQLite, Schema gemäß Spec Abschnitt 4.3.
|
||
//
|
||
// Verwendet wird der reine Go-Treiber modernc.org/sqlite (kein cgo), damit
|
||
// sultan ein statisch gelinktes Single Binary bleibt.
|
||
package store
|
||
|
||
import (
|
||
"context"
|
||
"database/sql"
|
||
"errors"
|
||
"fmt"
|
||
"strings"
|
||
"time"
|
||
|
||
"modernc.org/sqlite"
|
||
sqlite3 "modernc.org/sqlite/lib"
|
||
)
|
||
|
||
var (
|
||
// ErrNotFound: Der Datensatz (oder ein referenzierter Datensatz) existiert nicht.
|
||
ErrNotFound = errors.New("store: nicht gefunden")
|
||
// ErrExists: Ein Datensatz mit derselben ID bzw. demselben eindeutigen Wert existiert bereits.
|
||
ErrExists = errors.New("store: existiert bereits")
|
||
// ErrInvalid: Die Eingabe verletzt eine Validierungsregel.
|
||
ErrInvalid = errors.New("store: ungültige Eingabe")
|
||
// ErrTokenUsed: Der Enrollment-Token wurde bereits eingelöst.
|
||
ErrTokenUsed = errors.New("store: Token bereits verwendet")
|
||
// ErrTokenExpired: Der Enrollment-Token ist abgelaufen.
|
||
ErrTokenExpired = errors.New("store: Token abgelaufen")
|
||
// ErrSchemaTooNew: Die Datenbank stammt von einer neueren sultan-Version.
|
||
ErrSchemaTooNew = errors.New("store: Datenbankschema neuer als diese Programmversion")
|
||
)
|
||
|
||
// migrations[i] hebt das Schema von Version i auf i+1. Bestehende Einträge
|
||
// dürfen nie verändert, nur neue angehängt werden.
|
||
var migrations = []string{
|
||
`CREATE TABLE tenants (
|
||
id TEXT PRIMARY KEY,
|
||
name TEXT NOT NULL,
|
||
email TEXT NOT NULL,
|
||
tier TEXT NOT NULL DEFAULT 'community',
|
||
max_routes INTEGER NOT NULL DEFAULT 5,
|
||
allowed_domains TEXT NOT NULL,
|
||
valid_until DATETIME NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'active',
|
||
created_at DATETIME NOT NULL,
|
||
updated_at DATETIME NOT NULL
|
||
);
|
||
CREATE TABLE nodes (
|
||
id TEXT PRIMARY KEY,
|
||
tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
|
||
hostname TEXT NOT NULL,
|
||
public_key TEXT NOT NULL UNIQUE,
|
||
last_seen DATETIME NOT NULL,
|
||
ip_address TEXT NOT NULL,
|
||
status TEXT NOT NULL DEFAULT 'online'
|
||
);
|
||
CREATE INDEX nodes_tenant_id ON nodes(tenant_id);
|
||
CREATE TABLE enrollment_tokens (
|
||
token_hash TEXT PRIMARY KEY,
|
||
tenant_id TEXT NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
|
||
expires_at DATETIME NOT NULL,
|
||
used INTEGER NOT NULL DEFAULT 0
|
||
);
|
||
CREATE INDEX enrollment_tokens_tenant_id ON enrollment_tokens(tenant_id);`,
|
||
// v2: Server-Einstellungen, die auch `sultan token` (eigener Prozess) braucht.
|
||
`CREATE TABLE settings (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL
|
||
);`,
|
||
// v3: Wartung pro Knoten (eigene Dimension neben online/offline/drained).
|
||
`ALTER TABLE nodes ADD COLUMN maintenance_since DATETIME;
|
||
ALTER TABLE nodes ADD COLUMN maintenance_until DATETIME;
|
||
ALTER TABLE nodes ADD COLUMN maintenance_reason TEXT NOT NULL DEFAULT '';
|
||
ALTER TABLE nodes ADD COLUMN maintenance_public_reason TEXT NOT NULL DEFAULT '';`,
|
||
// v4: Leases, Befehlswarteschlange, Features und Zahlungsstatus, Einlösezeitpunkt.
|
||
`CREATE TABLE leases (
|
||
node_id TEXT PRIMARY KEY REFERENCES nodes(id) ON DELETE CASCADE,
|
||
issued_at DATETIME NOT NULL,
|
||
renew_after DATETIME NOT NULL,
|
||
valid_until DATETIME NOT NULL,
|
||
document TEXT NOT NULL
|
||
);
|
||
CREATE TABLE node_commands (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
node_id TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
|
||
command TEXT NOT NULL,
|
||
reason TEXT NOT NULL DEFAULT '',
|
||
created_at DATETIME NOT NULL,
|
||
sent_at DATETIME
|
||
);
|
||
CREATE INDEX node_commands_pending ON node_commands(node_id, sent_at);
|
||
ALTER TABLE tenants ADD COLUMN features TEXT NOT NULL DEFAULT '[]';
|
||
ALTER TABLE tenants ADD COLUMN payment_overdue_since DATETIME;
|
||
ALTER TABLE enrollment_tokens ADD COLUMN used_at DATETIME;`,
|
||
// v5: versandte Benachrichtigungen (jedes Ereignis genau einmal).
|
||
`CREATE TABLE notifications (
|
||
key TEXT PRIMARY KEY,
|
||
sent_at DATETIME NOT NULL
|
||
);`,
|
||
}
|
||
|
||
// Store ist der Zugriff auf die SQLite-Datenbank. Er ist für parallele
|
||
// Nutzung aus mehreren Goroutinen geeignet.
|
||
type Store struct {
|
||
db *sql.DB
|
||
now func() time.Time
|
||
}
|
||
|
||
// Open öffnet (bzw. erstellt) die Datenbank unter path und bringt das Schema
|
||
// auf den aktuellen Stand.
|
||
func Open(ctx context.Context, path string) (*Store, error) {
|
||
db, err := OpenDB(ctx, path, migrations)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
return &Store{db: db, now: time.Now}, nil
|
||
}
|
||
|
||
// OpenDB öffnet (bzw. erstellt) eine SQLite-Datenbank mit den Einstellungen
|
||
// des Servers und wendet migrations an: migrations[i] hebt das Schema von
|
||
// Version i auf i+1 (Version in PRAGMA user_version). Die Liste darf nur
|
||
// wachsen; eine neuere Datenbank liefert ErrSchemaTooNew. Für weitere
|
||
// Datenbanken wie accountant.db.
|
||
func OpenDB(ctx context.Context, path string, migrations []string) (*sql.DB, error) {
|
||
// foreign_keys: sonst greift ON DELETE CASCADE nicht (gilt pro Verbindung).
|
||
// _txlock=immediate: Schreib-Transaktionen holen die Sperre sofort und
|
||
// warten per busy_timeout, statt beim Upgrade mit SQLITE_BUSY abzubrechen.
|
||
dsn := "file:" + path +
|
||
"?_pragma=foreign_keys(1)&_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_txlock=immediate"
|
||
db, err := sql.Open("sqlite", dsn)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if err := migrate(ctx, db, migrations); err != nil {
|
||
db.Close()
|
||
return nil, err
|
||
}
|
||
return db, nil
|
||
}
|
||
|
||
// Close schließt die Datenbank.
|
||
func (s *Store) Close() error { return s.db.Close() }
|
||
|
||
// SchemaVersion liefert die aktuelle Schema-Version.
|
||
func (s *Store) SchemaVersion(ctx context.Context) (int, error) {
|
||
return schemaVersion(ctx, s.db)
|
||
}
|
||
|
||
func schemaVersion(ctx context.Context, db *sql.DB) (int, error) {
|
||
var v int
|
||
err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&v)
|
||
return v, err
|
||
}
|
||
|
||
func migrate(ctx context.Context, db *sql.DB, migrations []string) error {
|
||
v, err := schemaVersion(ctx, db)
|
||
if err != nil {
|
||
return fmt.Errorf("store: Schema-Version lesen: %w", err)
|
||
}
|
||
if v > len(migrations) {
|
||
return fmt.Errorf("%w (Datenbank v%d, bekannt bis v%d)", ErrSchemaTooNew, v, len(migrations))
|
||
}
|
||
for i := v; i < len(migrations); i++ {
|
||
tx, err := db.BeginTx(ctx, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if _, err := tx.ExecContext(ctx, migrations[i]); err != nil {
|
||
tx.Rollback()
|
||
return fmt.Errorf("store: Migration auf v%d: %w", i+1, err)
|
||
}
|
||
if _, err := tx.ExecContext(ctx, fmt.Sprintf("PRAGMA user_version = %d", i+1)); err != nil {
|
||
tx.Rollback()
|
||
return err
|
||
}
|
||
if err := tx.Commit(); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// timeLayout hat feste Breite, damit Zeitstempel als Text korrekt sortieren
|
||
// und sich in SQL vergleichen lassen.
|
||
const timeLayout = "2006-01-02T15:04:05.000000000Z"
|
||
|
||
func formatTime(t time.Time) string { return t.UTC().Format(timeLayout) }
|
||
|
||
// FormatTime formatiert t im Speicherformat (UTC, feste Breite) – für
|
||
// Datenbanken, die OpenDB nutzen.
|
||
func FormatTime(t time.Time) string { return formatTime(t) }
|
||
|
||
// ScanTime liest einen so gespeicherten Zeitstempel nach t (NULL = Nullzeit).
|
||
func ScanTime(t *time.Time) sql.Scanner { return dbTime{t} }
|
||
|
||
// normTime entfernt Monotonic-Clock und Zeitzone, damit gespeicherte und
|
||
// gelesene Werte identisch sind.
|
||
func normTime(t time.Time) time.Time { return t.Round(0).UTC() }
|
||
|
||
// dbTime liest einen Zeitstempel unabhängig davon, ob der Treiber ihn als
|
||
// Text oder bereits als time.Time liefert.
|
||
type dbTime struct{ t *time.Time }
|
||
|
||
func (d dbTime) Scan(src any) error {
|
||
switch v := src.(type) {
|
||
case nil: // NULL = nicht gesetzt
|
||
*d.t = time.Time{}
|
||
return nil
|
||
case time.Time:
|
||
*d.t = normTime(v)
|
||
return nil
|
||
case string:
|
||
return d.parse(v)
|
||
case []byte:
|
||
return d.parse(string(v))
|
||
default:
|
||
return fmt.Errorf("store: unerwarteter Zeittyp %T", src)
|
||
}
|
||
}
|
||
|
||
func (d dbTime) parse(s string) error {
|
||
t, err := time.Parse(timeLayout, s)
|
||
if err != nil {
|
||
return fmt.Errorf("store: Zeitstempel %q: %w", s, err)
|
||
}
|
||
*d.t = t
|
||
return nil
|
||
}
|
||
|
||
// execer ist die gemeinsame Teilmenge von *sql.DB und *sql.Tx.
|
||
type execer interface {
|
||
ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
|
||
}
|
||
|
||
func sqliteCode(err error) int {
|
||
var se *sqlite.Error
|
||
if errors.As(err, &se) {
|
||
return se.Code()
|
||
}
|
||
return 0
|
||
}
|
||
|
||
// mapConstraint übersetzt SQLite-Constraint-Fehler in die Fehler des Packages.
|
||
func mapConstraint(err error, what string) error {
|
||
switch sqliteCode(err) {
|
||
case sqlite3.SQLITE_CONSTRAINT_PRIMARYKEY, sqlite3.SQLITE_CONSTRAINT_UNIQUE:
|
||
return fmt.Errorf("%s: %w", what, ErrExists)
|
||
case sqlite3.SQLITE_CONSTRAINT_FOREIGNKEY:
|
||
return fmt.Errorf("%s: Mandant: %w", what, ErrNotFound)
|
||
}
|
||
return err
|
||
}
|
||
|
||
func blank(s string) bool { return strings.TrimSpace(s) == "" }
|
||
|
||
func invalid(format string, args ...any) error {
|
||
return fmt.Errorf("%w: %s", ErrInvalid, fmt.Sprintf(format, args...))
|
||
}
|
||
|
||
// requireAffected liefert ErrNotFound, wenn res keine Zeile betroffen hat.
|
||
func requireAffected(res sql.Result, err error, what string) error {
|
||
if err != nil {
|
||
return err
|
||
}
|
||
n, err := res.RowsAffected()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if n == 0 {
|
||
return fmt.Errorf("%s: %w", what, ErrNotFound)
|
||
}
|
||
return nil
|
||
}
|