uopi/internal/audit/sqlite.go

package audit

import (
	"database/sql"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	_ "modernc.org/sqlite" // pure-Go SQLite driver (no CGo, keeps the single static binary)
)

// defaultQueryLimit caps how many rows Query returns when the filter does not
// specify a smaller limit, protecting the API and UI from unbounded result sets.
const defaultQueryLimit = 1000

// writeBuffer is the depth of the async insert queue. When full, Record falls
// back to a synchronous insert so events are never silently dropped.
const writeBuffer = 4096

// sqliteRecorder appends events to a SQLite database. Inserts are normally
// handled by a background goroutine so Record does not block the calling write
// path; the queue has a synchronous fallback to guarantee durability under load.
type sqliteRecorder struct {
	db  *sql.DB
	log *slog.Logger

	ch     chan Event
	wg     sync.WaitGroup
	closed chan struct{}
	once   sync.Once
}

// NewSQLite opens (creating if needed) the audit database at path and starts the
// background writer. The returned Recorder must be Closed on shutdown.
func NewSQLite(path string, log *slog.Logger) (Recorder, error) {
	// WAL + a busy timeout let the async writer and synchronous query/fallback
	// paths share the file without "database is locked" errors.
	dsn := fmt.Sprintf("file:%s?_pragma=busy_timeout(5000)&_pragma=journal_mode(WAL)&_pragma=synchronous(NORMAL)", path)
	db, err := sql.Open("sqlite", dsn)
	if err != nil {
		return nil, fmt.Errorf("open audit db: %w", err)
	}
	if _, err := db.Exec(schema); err != nil {
		_ = db.Close()
		return nil, fmt.Errorf("init audit schema: %w", err)
	}
	r := &sqliteRecorder{
		db:     db,
		log:    log,
		ch:     make(chan Event, writeBuffer),
		closed: make(chan struct{}),
	}
	r.wg.Add(1)
	go r.writeLoop()
	return r, nil
}

const schema = `
CREATE TABLE IF NOT EXISTS audit_log (
	id         INTEGER PRIMARY KEY AUTOINCREMENT,
	ts_ns      INTEGER NOT NULL,
	actor      TEXT NOT NULL,
	actor_type TEXT NOT NULL,
	action     TEXT NOT NULL,
	ds         TEXT NOT NULL DEFAULT '',
	signal     TEXT NOT NULL DEFAULT '',
	value      TEXT NOT NULL DEFAULT '',
	detail     TEXT NOT NULL DEFAULT '',
	ip         TEXT NOT NULL DEFAULT '',
	outcome    TEXT NOT NULL DEFAULT 'ok',
	error      TEXT NOT NULL DEFAULT ''
);
CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log(ts_ns);
CREATE INDEX IF NOT EXISTS idx_audit_actor ON audit_log(actor);
CREATE INDEX IF NOT EXISTS idx_audit_action ON audit_log(action);
`

func (r *sqliteRecorder) Record(e Event) {
	if e.Time.IsZero() {
		e.Time = time.Now()
	}
	if e.Outcome == "" {
		e.Outcome = OutcomeOK
	}
	select {
	case <-r.closed:
		// Recorder is shutting down; best-effort synchronous insert.
		r.insert(e)
	case r.ch <- e:
	default:
		// Queue full: write synchronously rather than drop an audit record.
		r.insert(e)
	}
}

func (r *sqliteRecorder) writeLoop() {
	defer r.wg.Done()
	for e := range r.ch {
		r.insert(e)
	}
}

func (r *sqliteRecorder) insert(e Event) {
	_, err := r.db.Exec(
		`INSERT INTO audit_log (ts_ns, actor, actor_type, action, ds, signal, value, detail, ip, outcome, error)
		 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
		e.Time.UnixNano(), e.Actor, e.ActorType, e.Action,
		e.DS, e.Signal, e.Value, e.Detail, e.IP, e.Outcome, e.Error,
	)
	if err != nil {
		r.log.Error("audit: insert failed", "action", e.Action, "err", err)
	}
}

func (r *sqliteRecorder) Query(f Filter) ([]Event, error) {
	var where []string
	var args []any
	if !f.Start.IsZero() {
		where = append(where, "ts_ns >= ?")
		args = append(args, f.Start.UnixNano())
	}
	if !f.End.IsZero() {
		where = append(where, "ts_ns <= ?")
		args = append(args, f.End.UnixNano())
	}
	if f.Actor != "" {
		where = append(where, "actor = ?")
		args = append(args, f.Actor)
	}
	if f.Action != "" {
		where = append(where, "action = ?")
		args = append(args, f.Action)
	}
	if f.DS != "" {
		where = append(where, "ds = ?")
		args = append(args, f.DS)
	}
	if f.Signal != "" {
		where = append(where, "signal LIKE ?")
		args = append(args, "%"+f.Signal+"%")
	}
	limit := f.Limit
	if limit <= 0 || limit > defaultQueryLimit {
		limit = defaultQueryLimit
	}

	q := "SELECT ts_ns, actor, actor_type, action, ds, signal, value, detail, ip, outcome, error FROM audit_log"
	if len(where) > 0 {
		q += " WHERE " + strings.Join(where, " AND ")
	}
	q += " ORDER BY ts_ns DESC LIMIT ?"
	args = append(args, limit)

	rows, err := r.db.Query(q, args...)
	if err != nil {
		return nil, fmt.Errorf("query audit log: %w", err)
	}
	defer rows.Close()

	out := []Event{}
	for rows.Next() {
		var e Event
		var tsNs int64
		if err := rows.Scan(&tsNs, &e.Actor, &e.ActorType, &e.Action,
			&e.DS, &e.Signal, &e.Value, &e.Detail, &e.IP, &e.Outcome, &e.Error); err != nil {
			return nil, fmt.Errorf("scan audit row: %w", err)
		}
		e.Time = time.Unix(0, tsNs)
		out = append(out, e)
	}
	return out, rows.Err()
}

func (r *sqliteRecorder) Close() error {
	r.once.Do(func() {
		close(r.closed)
		close(r.ch)
	})
	r.wg.Wait()
	return r.db.Close()
}