edgeguard-native/internal/services/audit/audit.go

// Package audit appends rows to the audit_log table. Every mutation
// in the API funnels through this so the operator can answer
// "who did what when?" from a single SELECT.
package audit

import (
	"context"
	"encoding/json"
	"sync"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

type Repo struct {
	Pool *pgxpool.Pool

	subsMu sync.RWMutex
	subs   map[chan Entry]struct{}
}

func New(pool *pgxpool.Pool) *Repo {
	return &Repo{Pool: pool, subs: map[chan Entry]struct{}{}}
}

// Subscribe gibt einen Channel für Live-Audit-Events zurück + ein
// Unsubscribe-Cleanup. Channel-Buffer 32 — bei stehenden Clients
// werden Events gedropt (non-blocking-send).
func (r *Repo) Subscribe() (<-chan Entry, func()) {
	c := make(chan Entry, 32)
	r.subsMu.Lock()
	if r.subs == nil {
		r.subs = map[chan Entry]struct{}{}
	}
	r.subs[c] = struct{}{}
	r.subsMu.Unlock()
	return c, func() {
		r.subsMu.Lock()
		delete(r.subs, c)
		r.subsMu.Unlock()
		close(c)
	}
}

func (r *Repo) broadcast(e Entry) {
	r.subsMu.RLock()
	subs := make([]chan Entry, 0, len(r.subs))
	for c := range r.subs {
		subs = append(subs, c)
	}
	r.subsMu.RUnlock()
	for _, c := range subs {
		select {
		case c <- e:
		default:
		}
	}
}

// Entry mirrors one audit_log row — ListRecent returns these for
// the dashboard's recent-activity card.
type Entry struct {
	ID        int64           `json:"id"`
	Actor     string          `json:"actor"`
	Action    string          `json:"action"`
	Subject   *string         `json:"subject,omitempty"`
	Detail    json.RawMessage `json:"detail,omitempty"`
	NodeID    *string         `json:"node_id,omitempty"`
	CreatedAt time.Time       `json:"created_at"`
}

// ListRecent returns the most recent audit entries, newest first.
// Pass 0 for a sensible default (10).
func (r *Repo) ListRecent(ctx context.Context, limit int) ([]Entry, error) {
	if r == nil || r.Pool == nil {
		return []Entry{}, nil
	}
	if limit <= 0 || limit > 100 {
		limit = 10
	}
	rows, err := r.Pool.Query(ctx, `
SELECT id, actor, action, subject, detail, node_id, created_at
FROM audit_log
ORDER BY created_at DESC, id DESC
LIMIT $1`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]Entry, 0, limit)
	for rows.Next() {
		var e Entry
		if err := rows.Scan(&e.ID, &e.Actor, &e.Action, &e.Subject, &e.Detail, &e.NodeID, &e.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, rows.Err()
}

// SearchFilter beschreibt einen filter-gestützten Audit-Log-Abruf.
// Alle Felder optional — leere Werte werden vom Query ignoriert. Such-
// Strings sind case-insensitive ILIKE-Substring-Matches. Limit wird auf
// 500 gedeckelt (UI-Schutz vor versehentlichem Full-Scan), Offset für
// einfaches Paging.
type SearchFilter struct {
	Actor   string
	Action  string
	Subject string
	Since   *time.Time
	Until   *time.Time
	Limit   int
	Offset  int
}

// Search liefert audit_log-Einträge nach Filter, newest first. Filter-
// Felder werden via dynamisch zusammengesetzter WHERE-Klausel angewendet
// — Parametrisiert (kein String-Concat von User-Input).
func (r *Repo) Search(ctx context.Context, f SearchFilter) ([]Entry, error) {
	if r == nil || r.Pool == nil {
		return []Entry{}, nil
	}
	limit := f.Limit
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	offset := f.Offset
	if offset < 0 {
		offset = 0
	}
	args := []any{}
	where := ""
	add := func(cond string, val any) {
		args = append(args, val)
		if where == "" {
			where = " WHERE " + cond + "$" + itoa(len(args))
		} else {
			where += " AND " + cond + "$" + itoa(len(args))
		}
	}
	if f.Actor != "" {
		add("actor ILIKE ", "%"+f.Actor+"%")
	}
	if f.Action != "" {
		add("action ILIKE ", "%"+f.Action+"%")
	}
	if f.Subject != "" {
		add("subject ILIKE ", "%"+f.Subject+"%")
	}
	if f.Since != nil {
		add("created_at >= ", *f.Since)
	}
	if f.Until != nil {
		add("created_at <= ", *f.Until)
	}
	args = append(args, limit, offset)
	q := "SELECT id, actor, action, subject, detail, node_id, created_at FROM audit_log" +
		where +
		" ORDER BY created_at DESC, id DESC LIMIT $" + itoa(len(args)-1) +
		" OFFSET $" + itoa(len(args))
	rows, err := r.Pool.Query(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]Entry, 0, limit)
	for rows.Next() {
		var e Entry
		if err := rows.Scan(&e.ID, &e.Actor, &e.Action, &e.Subject, &e.Detail, &e.NodeID, &e.CreatedAt); err != nil {
			return nil, err
		}
		out = append(out, e)
	}
	return out, rows.Err()
}

func itoa(n int) string {
	// kleiner local-Helper damit wir nicht strconv für Single-Digit-
	// Parameter-Indizes importieren müssen.
	if n < 10 {
		return string(rune('0' + n))
	}
	// >9 Parameter ist hier in der Praxis nicht möglich (Filter <= 5 +
	// LIMIT/OFFSET = 7), aber Fallback für Robustness.
	s := ""
	for n > 0 {
		s = string(rune('0'+n%10)) + s
		n /= 10
	}
	return s
}

// Log writes one audit_log row. detail is JSON-encodable (typically a
// map[string]any) — empty map means "no payload". If pool is nil
// (e.g. dev env without DB), Log silently no-ops so handlers don't
// have to guard each call site.
func (r *Repo) Log(ctx context.Context, actor, action, subject string, detail any, nodeID string) error {
	if r == nil || r.Pool == nil {
		return nil
	}
	var detailJSON []byte
	if detail != nil {
		var err error
		detailJSON, err = json.Marshal(detail)
		if err != nil {
			return err
		}
	}
	var subjectArg any = subject
	if subject == "" {
		subjectArg = nil
	}
	var nodeArg any = nodeID
	if nodeID == "" {
		nodeArg = nil
	}
	// RETURNING id+created_at damit der Subscribe-Channel direkt einen
	// vollständigen Entry verteilen kann — Subscriber müssen nicht
	// erneut die DB hitten für die Anzeige.
	var e Entry
	e.Actor = actor
	e.Action = action
	if subject != "" {
		s := subject
		e.Subject = &s
	}
	if len(detailJSON) > 0 {
		e.Detail = json.RawMessage(detailJSON)
	}
	if nodeID != "" {
		n := nodeID
		e.NodeID = &n
	}
	err := r.Pool.QueryRow(ctx, `
INSERT INTO audit_log (actor, action, subject, detail, node_id)
VALUES ($1, $2, $3, $4, $5)
RETURNING id, created_at`,
		actor, action, subjectArg, detailJSON, nodeArg).
		Scan(&e.ID, &e.CreatedAt)
	if err != nil {
		return err
	}
	// Broadcast nach erfolgreichem INSERT — wenn DB ablehnt, sollen
	// Subscribers das Event auch nicht sehen.
	r.broadcast(e)
	return nil
}

// Cleanup löscht alle audit_log-Rows die älter als keepDays sind.
// Schutz vor unbounded growth bei langlebigen Boxen (audit_log kann
// sonst nach 1-2 Jahren mehrere GB Disk + entsprechende Query-Latenz
// haben). Liefert die Anzahl gelöschter Rows.
//
// keepDays <= 0 → no-op (Cleanup deaktiviert, alles bleibt erhalten).
// keepDays sollte deutlich über Audit-Anforderungen liegen — 90 Tage
// ist ein vernünftiger Default für Self-Service-Boxen.
func (r *Repo) Cleanup(ctx context.Context, keepDays int) (int64, error) {
	if r == nil || r.Pool == nil || keepDays <= 0 {
		return 0, nil
	}
	// make_interval(days => $1) nimmt $1 als int — sauber typisiert. Der frühere
	// ($1::text || ' days')::interval-Ansatz scheiterte, weil keepDays als int
	// übergeben wird und pgx int nicht als text (OID 25) encoden kann
	// ("cannot find encode plan") → Cleanup lief nie.
	tag, err := r.Pool.Exec(ctx, `
DELETE FROM audit_log
WHERE created_at < NOW() - make_interval(days => $1)`, keepDays)
	if err != nil {
		return 0, err
	}
	return tag.RowsAffected(), nil
}