Hammerhead/db.go

package hammerhead

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

	"git.veillax.com/Veillax/reef/model"
	"git.veillax.com/Veillax/reef/secretbox"
	_ "modernc.org/sqlite" // pure Go, so no C compiler is needed
)

// The configuration is held in memory (it is small and read constantly) and
// persisted to SQLite as one row per object, written incrementally in a
// transaction: only objects that changed since the last save are touched.
// Secrets are encrypted before they reach the file. Logs, traffic history and
// alerts are real tables with indexes, so they can be searched and pruned.

const schema = `
CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS entities (
	kind TEXT NOT NULL, id TEXT NOT NULL, data TEXT NOT NULL,
	PRIMARY KEY (kind, id)
);
CREATE TABLE IF NOT EXISTS access_log (
	id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, org_id TEXT NOT NULL DEFAULT '', mapping TEXT NOT NULL DEFAULT '',
	remote TEXT, method TEXT, host TEXT, path TEXT, status INTEGER, usr TEXT, result TEXT,
	bytes_in INTEGER, bytes_out INTEGER, ms INTEGER
);
CREATE INDEX IF NOT EXISTS access_org ON access_log (org_id, id);
CREATE INDEX IF NOT EXISTS access_map ON access_log (org_id, mapping, id);
CREATE INDEX IF NOT EXISTS access_ts ON access_log (ts);
CREATE TABLE IF NOT EXISTS audit_log (
	id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, org_id TEXT NOT NULL DEFAULT '',
	actor TEXT, action TEXT, detail TEXT
);
CREATE INDEX IF NOT EXISTS audit_org ON audit_log (org_id, id);
CREATE TABLE IF NOT EXISTS traffic_hourly (
	mapping TEXT NOT NULL, hour INTEGER NOT NULL, conns INTEGER NOT NULL DEFAULT 0,
	bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0,
	PRIMARY KEY (mapping, hour)
);
CREATE TABLE IF NOT EXISTS traffic_minutely (
	mapping TEXT NOT NULL, minute INTEGER NOT NULL, conns INTEGER NOT NULL DEFAULT 0,
	bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0,
	PRIMARY KEY (mapping, minute)
);
CREATE TABLE IF NOT EXISTS traffic_sources (
	mapping TEXT NOT NULL, hour INTEGER NOT NULL, ip TEXT NOT NULL, conns INTEGER NOT NULL DEFAULT 0,
	bytes_in INTEGER NOT NULL DEFAULT 0, bytes_out INTEGER NOT NULL DEFAULT 0,
	PRIMARY KEY (mapping, hour, ip)
);
CREATE TABLE IF NOT EXISTS alerts (
	key TEXT PRIMARY KEY, org_id TEXT NOT NULL DEFAULT '', kind TEXT, name TEXT, message TEXT,
	since INTEGER NOT NULL, fired INTEGER NOT NULL DEFAULT 0, notified INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS alert_events (
	id INTEGER PRIMARY KEY AUTOINCREMENT, ts INTEGER NOT NULL, org_id TEXT NOT NULL DEFAULT '',
	key TEXT, status TEXT, name TEXT, message TEXT
);
CREATE INDEX IF NOT EXISTS alert_events_org ON alert_events (org_id, id);
`

type DB struct {
	sql *sql.DB
	box *secretbox.Box

	mu    sync.Mutex // serialises persist
	cache map[string]map[string]string
	ver   string
}

func openDB(path string, box *secretbox.Box) (*DB, error) {
	conn, err := sql.Open("sqlite", path+"?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)")
	if err != nil {
		return nil, err
	}
	conn.SetMaxOpenConns(8)
	if _, err := conn.Exec(schema); err != nil {
		conn.Close()
		return nil, fmt.Errorf("prepare database: %w", err)
	}
	return &DB{sql: conn, box: box, cache: map[string]map[string]string{}}, nil
}

func (d *DB) Close() error { return d.sql.Close() }

func ms(t time.Time) int64 { return t.UnixMilli() }

// ---------- sealing of secret fields ----------

// sealed returns a copy of v with its secret fields encrypted.
func (d *DB) sealed(v any) any {
	b := d.box
	switch x := v.(type) {
	case *model.Edge:
		c := *x
		c.Token, c.AuthKey = b.Seal(c.Token), b.Seal(c.AuthKey)
		return &c
	case *model.Site:
		c := *x
		c.Token = b.Seal(c.Token)
		return &c
	case *User:
		c := *x
		c.TOTPSecret = b.Seal(c.TOTPSecret)
		return &c
	case *IdP:
		c := *x
		c.ClientSecret = b.Seal(c.ClientSecret)
		return &c
	case *model.MailEgress:
		c := *x
		c.Password = b.Seal(c.Password)
		return &c
	case *model.DKIMKey:
		c := *x
		c.PrivateKey = b.Seal(c.PrivateKey)
		return &c
	case *AlertChannel:
		c := *x
		c.Target, c.Token = b.Seal(c.Target), b.Seal(c.Token)
		return &c
	case Settings:
		x.SMTP.Pass = b.Seal(x.SMTP.Pass)
		return x
	}
	return v
}

func (d *DB) open1(s *string) error {
	v, err := d.box.Open(*s)
	*s = v
	return err
}

// unseal decrypts the secret fields of a loaded object in place.
func (d *DB) unseal(v any) error {
	switch x := v.(type) {
	case *model.Edge:
		return errors.Join(d.open1(&x.Token), d.open1(&x.AuthKey))
	case *model.Site:
		return d.open1(&x.Token)
	case *User:
		return d.open1(&x.TOTPSecret)
	case *IdP:
		return d.open1(&x.ClientSecret)
	case *model.MailEgress:
		return d.open1(&x.Password)
	case *model.DKIMKey:
		return d.open1(&x.PrivateKey)
	case *AlertChannel:
		return errors.Join(d.open1(&x.Target), d.open1(&x.Token))
	case *Settings:
		return d.open1(&x.SMTP.Pass)
	}
	return nil
}

// ---------- configuration ----------

type ent struct {
	kind, id string
	v        any
}

func collect(st *state) []ent {
	var out []ent
	add := func(kind, id string, v any) { out = append(out, ent{kind, id, v}) }
	add("settings", "main", st.Settings)
	for _, x := range st.Users {
		add("user", x.ID, x)
	}
	for _, x := range st.Orgs {
		add("org", x.ID, x)
	}
	for _, x := range st.Roles {
		add("role", x.ID, x)
	}
	for _, x := range st.Members {
		add("member", x.UserID+"|"+x.OrgID, x)
	}
	for _, x := range st.Invites {
		add("invite", x.ID, x)
	}
	for _, x := range st.APIKeys {
		add("apikey", x.ID, x)
	}
	for _, x := range st.IdPs {
		add("idp", x.ID, x)
	}
	for _, x := range st.Edges {
		add("edge", x.ID, x)
	}
	for _, x := range st.Sites {
		add("site", x.ID, x)
	}
	for _, x := range st.Domains {
		add("domain", x.ID, x)
	}
	for _, x := range st.Mappings {
		add("mapping", x.ID, x)
	}
	for _, x := range st.MailEgress {
		add("mailegress", x.ID, x)
	}
	for _, x := range st.DKIM {
		add("dkim", x.ID, x)
	}
	for _, x := range st.Channels {
		add("channel", x.ID, x)
	}
	for _, x := range st.Shares {
		add("share", x.ID, x)
	}
	for k, x := range st.Secrets {
		add("secret", k, x)
	}
	for k, x := range st.Totals {
		add("total", k, x)
	}
	for k, x := range st.Sessions {
		add("session", k, x)
	}
	return out
}

// persist writes whatever changed since the last call, in one transaction.
func (d *DB) persist(st *state) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	ents := collect(st)
	type change struct {
		e     ent
		plain string
	}
	var changes []change
	seen := map[string]map[string]bool{}
	for _, e := range ents {
		b, err := json.Marshal(e.v)
		if err != nil {
			return err
		}
		if seen[e.kind] == nil {
			seen[e.kind] = map[string]bool{}
		}
		seen[e.kind][e.id] = true
		if d.cache[e.kind][e.id] != string(b) {
			changes = append(changes, change{e, string(b)})
		}
	}
	type gone struct{ kind, id string }
	var removed []gone
	for kind, ids := range d.cache {
		for id := range ids {
			if !seen[kind][id] {
				removed = append(removed, gone{kind, id})
			}
		}
	}
	ver := fmt.Sprint(st.Version)
	if len(changes) == 0 && len(removed) == 0 && ver == d.ver {
		return nil
	}
	tx, err := d.sql.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	for _, c := range changes {
		sealed, err := json.Marshal(d.sealed(c.e.v))
		if err != nil {
			return err
		}
		if _, err := tx.Exec(`INSERT INTO entities (kind, id, data) VALUES (?, ?, ?) ON CONFLICT (kind, id) DO UPDATE SET data = excluded.data`,
			c.e.kind, c.e.id, string(sealed)); err != nil {
			return err
		}
	}
	for _, g := range removed {
		if _, err := tx.Exec(`DELETE FROM entities WHERE kind = ? AND id = ?`, g.kind, g.id); err != nil {
			return err
		}
	}
	if ver != d.ver {
		if _, err := tx.Exec(`INSERT INTO meta (k, v) VALUES ('version', ?) ON CONFLICT (k) DO UPDATE SET v = excluded.v`, ver); err != nil {
			return err
		}
	}
	if err := tx.Commit(); err != nil {
		return err
	}
	for _, c := range changes {
		if d.cache[c.e.kind] == nil {
			d.cache[c.e.kind] = map[string]string{}
		}
		d.cache[c.e.kind][c.e.id] = c.plain
	}
	for _, g := range removed {
		delete(d.cache[g.kind], g.id)
	}
	d.ver = ver
	return nil
}

// load reads every stored object into st. It reports whether the database had any.
func (d *DB) load(st *state) (bool, error) {
	d.mu.Lock()
	defer d.mu.Unlock()
	rows, err := d.sql.Query(`SELECT kind, id, data FROM entities`)
	if err != nil {
		return false, err
	}
	defer rows.Close()
	any1 := false
	for rows.Next() {
		var kind, id, data string
		if err := rows.Scan(&kind, &id, &data); err != nil {
			return false, err
		}
		any1 = true
		var target any
		switch kind {
		case "settings":
			target = &st.Settings
		case "user":
			target = new(User)
		case "org":
			target = new(Org)
		case "role":
			target = new(Role)
		case "member":
			target = new(Member)
		case "invite":
			target = new(Invite)
		case "apikey":
			target = new(APIKey)
		case "idp":
			target = new(IdP)
		case "edge":
			target = new(model.Edge)
		case "site":
			target = new(model.Site)
		case "domain":
			target = new(model.Domain)
		case "mapping":
			target = new(model.Mapping)
		case "mailegress":
			target = new(model.MailEgress)
		case "dkim":
			target = new(model.DKIMKey)
		case "channel":
			target = new(AlertChannel)
		case "share":
			target = new(ShareLink)
		case "secret":
			target = new(Secret)
		case "total":
			target = new(Total)
		case "session":
			target = new(Session)
		default:
			continue
		}
		if err := json.Unmarshal([]byte(data), target); err != nil {
			return false, fmt.Errorf("read %s %s: %w", kind, id, err)
		}
		if err := d.unseal(target); err != nil {
			return false, fmt.Errorf("%s %s: %w", kind, id, err)
		}
		switch x := target.(type) {
		case *User:
			st.Users = append(st.Users, x)
		case *Org:
			st.Orgs = append(st.Orgs, x)
		case *Role:
			st.Roles = append(st.Roles, x)
		case *Member:
			st.Members = append(st.Members, x)
		case *Invite:
			st.Invites = append(st.Invites, x)
		case *APIKey:
			st.APIKeys = append(st.APIKeys, x)
		case *IdP:
			st.IdPs = append(st.IdPs, x)
		case *model.Edge:
			st.Edges = append(st.Edges, x)
		case *model.Site:
			st.Sites = append(st.Sites, x)
		case *model.Domain:
			st.Domains = append(st.Domains, x)
		case *model.Mapping:
			st.Mappings = append(st.Mappings, x)
		case *model.MailEgress:
			st.MailEgress = append(st.MailEgress, x)
		case *model.DKIMKey:
			st.DKIM = append(st.DKIM, x)
		case *AlertChannel:
			st.Channels = append(st.Channels, x)
		case *ShareLink:
			st.Shares = append(st.Shares, x)
		case *Secret:
			st.Secrets[id] = x
		case *Total:
			st.Totals[id] = x
		case *Session:
			st.Sessions[id] = x
		}
	}
	if err := rows.Err(); err != nil {
		return false, err
	}
	var v string
	if d.sql.QueryRow(`SELECT v FROM meta WHERE k = 'version'`).Scan(&v) == nil {
		fmt.Sscan(v, &st.Version)
	}
	// remember exactly what is stored, so the next persist only writes differences
	d.cache = map[string]map[string]string{}
	for _, e := range collect(st) {
		b, _ := json.Marshal(e.v)
		if d.cache[e.kind] == nil {
			d.cache[e.kind] = map[string]string{}
		}
		d.cache[e.kind][e.id] = string(b)
	}
	d.ver = fmt.Sprint(st.Version)
	return any1, nil
}

// ---------- access log ----------

type accessFilter struct {
	Org, Mapping, Q, Result string
	StatusClass             int // 2, 3, 4 or 5; 0 = any
	Before                  int64
	Limit                   int
}

func likeEscape(s string) string {
	return strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(s)
}

func (d *DB) addAccess(es []model.LogEntry) error {
	if len(es) == 0 {
		return nil
	}
	tx, err := d.sql.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	st, err := tx.Prepare(`INSERT INTO access_log (ts, org_id, mapping, remote, method, host, path, status, usr, result, bytes_in, bytes_out, ms)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`)
	if err != nil {
		return err
	}
	defer st.Close()
	for _, e := range es {
		if e.Time.IsZero() {
			e.Time = time.Now()
		}
		if _, err := st.Exec(ms(e.Time), e.OrgID, e.Mapping, e.Remote, e.Method, e.Host, e.Path, e.Status, e.User, e.Result, e.BytesIn, e.BytesOut, e.Millis); err != nil {
			return err
		}
	}
	return tx.Commit()
}

func (d *DB) queryAccess(f accessFilter) []model.LogEntry {
	q := `SELECT id, ts, mapping, remote, method, host, path, status, usr, result, bytes_in, bytes_out, ms FROM access_log WHERE org_id = ?`
	args := []any{f.Org}
	if f.Mapping != "" {
		q += ` AND mapping = ?`
		args = append(args, f.Mapping)
	}
	if f.Q != "" {
		l := "%" + likeEscape(f.Q) + "%"
		q += ` AND (host LIKE ? ESCAPE '\' OR path LIKE ? ESCAPE '\' OR remote LIKE ? ESCAPE '\' OR usr LIKE ? ESCAPE '\')`
		args = append(args, l, l, l, l)
	}
	if f.StatusClass >= 2 && f.StatusClass <= 5 {
		q += ` AND status >= ? AND status < ?`
		args = append(args, f.StatusClass*100, f.StatusClass*100+100)
	}
	if f.Result != "" {
		q += ` AND result LIKE ? ESCAPE '\'`
		args = append(args, likeEscape(f.Result)+"%")
	}
	if f.Before > 0 {
		q += ` AND id < ?`
		args = append(args, f.Before)
	}
	if f.Limit <= 0 || f.Limit > 1000 {
		f.Limit = 200
	}
	q += ` ORDER BY id DESC LIMIT ?`
	args = append(args, f.Limit)
	rows, err := d.sql.Query(q, args...)
	out := []model.LogEntry{}
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		var e model.LogEntry
		var ts int64
		var remote, method, host, path, usr, result sql.NullString
		var status sql.NullInt64
		if rows.Scan(&e.ID, &ts, &e.Mapping, &remote, &method, &host, &path, &status, &usr, &result, &e.BytesIn, &e.BytesOut, &e.Millis) != nil {
			continue
		}
		e.Time, e.OrgID = time.UnixMilli(ts), f.Org
		e.Remote, e.Method, e.Host, e.Path, e.User, e.Result = remote.String, method.String, host.String, path.String, usr.String, result.String
		e.Status = int(status.Int64)
		out = append(out, e)
	}
	return out
}

// ---------- audit log ----------

func (d *DB) addAudit(e AuditEntry) {
	d.sql.Exec(`INSERT INTO audit_log (ts, org_id, actor, action, detail) VALUES (?, ?, ?, ?, ?)`, ms(e.Time), e.OrgID, e.Actor, e.Action, e.Detail)
}

func (d *DB) queryAudit(org string, limit int, before int64) []AuditEntry {
	if limit <= 0 || limit > 1000 {
		limit = 200
	}
	q, args := `SELECT id, ts, actor, action, detail FROM audit_log WHERE org_id = ?`, []any{org}
	if before > 0 {
		q += ` AND id < ?`
		args = append(args, before)
	}
	rows, err := d.sql.Query(q+` ORDER BY id DESC LIMIT ?`, append(args, limit)...)
	out := []AuditEntry{}
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		var e AuditEntry
		var ts int64
		var actor, action, detail sql.NullString
		if rows.Scan(&e.ID, &ts, &actor, &action, &detail) != nil {
			continue
		}
		e.Time, e.OrgID, e.Actor, e.Action, e.Detail = time.UnixMilli(ts), org, actor.String, action.String, detail.String
		out = append(out, e)
	}
	return out
}

// purge deletes logs older than the retention periods.
func (d *DB) purge(accessDays, auditDays int) {
	if accessDays > 0 {
		d.sql.Exec(`DELETE FROM access_log WHERE ts < ?`, ms(time.Now().AddDate(0, 0, -accessDays)))
	}
	if auditDays > 0 {
		d.sql.Exec(`DELETE FROM audit_log WHERE ts < ?`, ms(time.Now().AddDate(0, 0, -auditDays)))
	}
	d.sql.Exec(`DELETE FROM traffic_hourly WHERE hour < ?`, time.Now().Add(-trafficHourlyKeep).Unix()/3600)
	d.sql.Exec(`DELETE FROM traffic_minutely WHERE minute < ?`, time.Now().Add(-trafficMinutelyKeep).Unix()/60)
	d.sql.Exec(`DELETE FROM traffic_sources WHERE hour < ?`, time.Now().Add(-trafficSourcesKeep).Unix()/3600)
	d.sql.Exec(`DELETE FROM alert_events WHERE ts < ?`, ms(time.Now().AddDate(0, 0, -90)))
}

// ---------- traffic history ----------

// How long each resolution of traffic history is kept. Minutes are for the last hours, hours for the
// months, and the busiest client addresses sit in between.
const (
	trafficMinutelyKeep = 48 * time.Hour
	trafficSourcesKeep  = 7 * 24 * time.Hour
	trafficHourlyKeep   = 31 * 24 * time.Hour
)

func (d *DB) addTraffic(mapping string, st model.Stat) { d.addTrafficAt(mapping, st, time.Now()) }

func (d *DB) addTrafficAt(mapping string, st model.Stat, at time.Time) {
	d.sql.Exec(`INSERT INTO traffic_hourly (mapping, hour, conns, bytes_in, bytes_out) VALUES (?, ?, ?, ?, ?)
		ON CONFLICT (mapping, hour) DO UPDATE SET conns = conns + excluded.conns, bytes_in = bytes_in + excluded.bytes_in, bytes_out = bytes_out + excluded.bytes_out`,
		mapping, at.Unix()/3600, st.Conns, st.BytesIn, st.BytesOut)
	d.sql.Exec(`INSERT INTO traffic_minutely (mapping, minute, conns, bytes_in, bytes_out) VALUES (?, ?, ?, ?, ?)
		ON CONFLICT (mapping, minute) DO UPDATE SET conns = conns + excluded.conns, bytes_in = bytes_in + excluded.bytes_in, bytes_out = bytes_out + excluded.bytes_out`,
		mapping, at.Unix()/60, st.Conns, st.BytesIn, st.BytesOut)
	for _, src := range st.Sources {
		d.sql.Exec(`INSERT INTO traffic_sources (mapping, hour, ip, conns, bytes_in, bytes_out) VALUES (?, ?, ?, ?, ?, ?)
			ON CONFLICT (mapping, hour, ip) DO UPDATE SET conns = conns + excluded.conns, bytes_in = bytes_in + excluded.bytes_in, bytes_out = bytes_out + excluded.bytes_out`,
			mapping, at.Unix()/3600, src.IP, src.Conns, src.BytesIn, src.BytesOut)
	}
}

// TrafficPoint is the traffic of one time bucket.
type TrafficPoint struct {
	Time     int64 `json:"t"` // unix seconds at the start of the bucket
	Conns    int64 `json:"conns"`
	BytesIn  int64 `json:"bytesIn"`
	BytesOut int64 `json:"bytesOut"`
}

// trafficSeries sums the given ports' traffic into count buckets of the given size ("minute" or
// "hour") ending with the current one, oldest first. Empty buckets are included.
func (d *DB) trafficSeries(ids []string, unit string, count int, now time.Time) []TrafficPoint {
	table, col, secs := "traffic_hourly", "hour", int64(3600)
	if unit == "minute" {
		table, col, secs = "traffic_minutely", "minute", 60
	}
	cur := now.Unix() / secs
	first := cur - int64(count) + 1
	out := make([]TrafficPoint, count)
	for i := range out {
		out[i].Time = (first + int64(i)) * secs
	}
	if len(ids) == 0 {
		return out
	}
	q := `SELECT ` + col + `, SUM(conns), SUM(bytes_in), SUM(bytes_out) FROM ` + table +
		` WHERE ` + col + ` >= ? AND ` + col + ` <= ? AND mapping IN (` + placeholders(len(ids)) + `) GROUP BY ` + col
	args := []any{first, cur}
	for _, id := range ids {
		args = append(args, id)
	}
	rows, err := d.sql.Query(q, args...)
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		var at int64
		var p TrafficPoint
		if rows.Scan(&at, &p.Conns, &p.BytesIn, &p.BytesOut) != nil {
			continue
		}
		if i := int(at - first); i >= 0 && i < count {
			out[i].Conns, out[i].BytesIn, out[i].BytesOut = p.Conns, p.BytesIn, p.BytesOut
		}
	}
	return out
}

func placeholders(n int) string {
	if n <= 0 {
		return ""
	}
	return strings.TrimSuffix(strings.Repeat("?,", n), ",")
}

// TrafficSource is one client address's traffic over a period.
type TrafficSource struct {
	IP       string `json:"ip"`
	Conns    int64  `json:"conns"`
	BytesIn  int64  `json:"bytesIn"`
	BytesOut int64  `json:"bytesOut"`
}

// topSources lists the busiest client addresses of the given ports over the last hours hours.
func (d *DB) topSources(ids []string, hours, limit int, now time.Time) []TrafficSource {
	out := []TrafficSource{}
	if len(ids) == 0 {
		return out
	}
	args := []any{now.Unix()/3600 - int64(hours) + 1}
	for _, id := range ids {
		args = append(args, id)
	}
	args = append(args, limit)
	rows, err := d.sql.Query(`SELECT ip, SUM(conns), SUM(bytes_in), SUM(bytes_out) FROM traffic_sources WHERE hour >= ? AND mapping IN (`+placeholders(len(ids))+`)
		GROUP BY ip ORDER BY SUM(bytes_in) + SUM(bytes_out) DESC, SUM(conns) DESC, ip LIMIT ?`, args...)
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		var s TrafficSource
		if rows.Scan(&s.IP, &s.Conns, &s.BytesIn, &s.BytesOut) == nil {
			out = append(out, s)
		}
	}
	return out
}

// bytesSince is the total bytes (in + out) a port has carried from the start of the given hour (unix hours) until now.
func (d *DB) bytesSince(mapping string, fromHour int64) int64 {
	var n sql.NullInt64
	d.sql.QueryRow(`SELECT SUM(bytes_in + bytes_out) FROM traffic_hourly WHERE mapping = ? AND hour >= ?`, mapping, fromHour).Scan(&n)
	return n.Int64
}

// spikeWindow and spikeBaseline are the minutes a spike is judged over, and the minutes before it
// that say what normal looks like.
const (
	spikeWindow   = 5
	spikeBaseline = 6 * 60
)

// recentTraffic returns the bytes a port carried in the last spikeWindow minutes, the average for
// the same span over the baseline before it, and how many minutes that baseline covers (up to spikeBaseline;
// fewer for a port with a shorter history, which is averaged over what it has).
func (d *DB) recentTraffic(mapping string, now time.Time) (recent int64, baselinePerWindow float64, historyMinutes int64) {
	cur := now.Unix() / 60
	var r, b, first sql.NullInt64
	d.sql.QueryRow(`SELECT SUM(CASE WHEN minute > ? THEN bytes_in + bytes_out END), SUM(CASE WHEN minute <= ? THEN bytes_in + bytes_out END), MIN(minute)
		FROM traffic_minutely WHERE mapping = ? AND minute > ?`, cur-spikeWindow, cur-spikeWindow, mapping, cur-spikeWindow-spikeBaseline).Scan(&r, &b, &first)
	if !first.Valid {
		return 0, 0, 0
	}
	span := cur - spikeWindow - first.Int64 + 1
	if span > spikeBaseline {
		span = spikeBaseline
	}
	if span <= 0 {
		return r.Int64, 0, 0
	}
	return r.Int64, float64(b.Int64) / float64(span) * spikeWindow, span
}

// trafficHistory returns bytes (in + out) per hour for each mapping, oldest first.
func (d *DB) trafficHistory(ids []string, hours int) map[string][]int64 {
	out := map[string][]int64{}
	if hours <= 0 || hours > 24*30 {
		hours = 24
	}
	now := time.Now().Unix() / 3600
	for _, id := range ids {
		out[id] = make([]int64, hours)
	}
	rows, err := d.sql.Query(`SELECT mapping, hour, bytes_in + bytes_out FROM traffic_hourly WHERE hour > ?`, now-int64(hours))
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		var id string
		var h, n int64
		if rows.Scan(&id, &h, &n) != nil {
			continue
		}
		if series, ok := out[id]; ok {
			if i := int(h - (now - int64(hours) + 1)); i >= 0 && i < hours {
				series[i] += n
			}
		}
	}
	return out
}

// ---------- alerts ----------

func (d *DB) saveAlert(a *alertState) {
	fired := 0
	if a.Fired {
		fired = 1
	}
	d.sql.Exec(`INSERT INTO alerts (key, org_id, kind, name, message, since, fired, notified) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT (key) DO UPDATE SET name = excluded.name, message = excluded.message, fired = excluded.fired, notified = excluded.notified`,
		a.Key, a.OrgID, a.Kind, a.Name, a.Message, ms(a.Since), fired, ms(a.Notified))
}

func (d *DB) deleteAlert(key string) { d.sql.Exec(`DELETE FROM alerts WHERE key = ?`, key) }

func (d *DB) loadAlertRows() []alertState {
	rows, err := d.sql.Query(`SELECT key, org_id, kind, name, message, since, fired, notified FROM alerts`)
	if err != nil {
		return nil
	}
	defer rows.Close()
	var out []alertState
	for rows.Next() {
		var a alertState
		var since, notified, fired int64
		var kind, name, msg sql.NullString
		if rows.Scan(&a.Key, &a.OrgID, &kind, &name, &msg, &since, &fired, &notified) != nil {
			continue
		}
		a.Kind, a.Name, a.Message = kind.String, name.String, msg.String
		a.Since, a.Fired, a.Notified = time.UnixMilli(since), fired == 1, time.UnixMilli(notified)
		out = append(out, a)
	}
	return out
}

type alertEvent struct {
	ID      int64     `json:"id"`
	Time    time.Time `json:"time"`
	Status  string    `json:"status"`
	Name    string    `json:"name"`
	Message string    `json:"message"`
}

func (d *DB) addAlertEvent(t time.Time, org, key, status, name, msg string) {
	d.sql.Exec(`INSERT INTO alert_events (ts, org_id, key, status, name, message) VALUES (?, ?, ?, ?, ?, ?)`, ms(t), org, key, status, name, msg)
}

func (d *DB) queryAlertEvents(org string, limit int) []alertEvent {
	out := []alertEvent{}
	rows, err := d.sql.Query(`SELECT id, ts, status, name, message FROM alert_events WHERE org_id = ? ORDER BY id DESC LIMIT ?`, org, limit)
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		var e alertEvent
		var ts int64
		var status, name, msg sql.NullString
		if rows.Scan(&e.ID, &ts, &status, &name, &msg) != nil {
			continue
		}
		e.Time, e.Status, e.Name, e.Message = time.UnixMilli(ts), status.String, name.String, msg.String
		out = append(out, e)
	}
	return out
}