edgeguard-native/internal/waf/alerts.go

package waf

import (
	"context"
	"log/slog"
	"sync"
	"sync/atomic"
	"time"

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

// Alert represents a single WAF rule match that was logged.
type Alert struct {
	ID        int64     `json:"id"`
	DomainID  *int64    `json:"domain_id,omitempty"`
	Hostname  string    `json:"hostname"`
	ClientIP  string    `json:"client_ip"`
	Method    string    `json:"method"`
	URI       string    `json:"uri"`
	RuleID    int       `json:"rule_id"`
	RuleMsg   string    `json:"rule_msg"`
	Severity  string    `json:"severity"`
	Action    string    `json:"action"` // "detected" | "blocked"
	CreatedAt time.Time `json:"created_at"`
}

// AlertWriter accepts Alert values via a buffered channel and writes
// them to PostgreSQL asynchronously so SPOE handling stays low-latency.
type AlertWriter struct {
	pool      *pgxpool.Pool
	ch        chan Alert
	stop      chan struct{}
	done      chan struct{}
	closeOnce sync.Once
	closed    atomic.Bool
}

// NewAlertWriter creates an AlertWriter and starts its background goroutine.
// bufSize is the number of unwritten alerts that can queue before drops.
func NewAlertWriter(pool *pgxpool.Pool, bufSize int) *AlertWriter {
	aw := &AlertWriter{
		pool: pool,
		ch:   make(chan Alert, bufSize),
		stop: make(chan struct{}),
		done: make(chan struct{}),
	}
	go aw.run()
	return aw
}

// Send enqueues an alert. Drops silently if the channel is full (or the
// writer is closing) to avoid slowing down / panicking SPOE handling.
func (aw *AlertWriter) Send(a Alert) {
	if aw.closed.Load() {
		return
	}
	select {
	case aw.ch <- a:
	default:
		slog.Warn("waf: alert channel full — dropping alert", "host", a.Hostname, "rule", a.RuleID)
	}
}

// Close stops the writer and flushes buffered alerts (best-effort).
// Safe to call multiple times. The channel is never closed → Send never
// panics even if it races with Close.
func (aw *AlertWriter) Close() {
	aw.closeOnce.Do(func() {
		aw.closed.Store(true)
		close(aw.stop)
	})
	<-aw.done
}

func (aw *AlertWriter) run() {
	defer close(aw.done)
	for {
		select {
		case a := <-aw.ch:
			aw.write(a)
		case <-aw.stop:
			// Restliche gepufferte Alerts noch wegschreiben, dann Ende.
			for {
				select {
				case a := <-aw.ch:
					aw.write(a)
				default:
					return
				}
			}
		}
	}
}

func (aw *AlertWriter) write(a Alert) {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Resolve domain_id from hostname (best-effort).
	var domainID *int64
	var id int64
	if err := aw.pool.QueryRow(ctx,
		`SELECT id FROM domains WHERE name = $1 AND active = true LIMIT 1`,
		a.Hostname,
	).Scan(&id); err == nil {
		domainID = &id
	}

	if _, err := aw.pool.Exec(ctx, `
		INSERT INTO waf_alerts
		    (domain_id, hostname, client_ip, method, uri,
		     rule_id, rule_msg, severity, action)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
	`, domainID, a.Hostname, a.ClientIP, a.Method, a.URI,
		a.RuleID, a.RuleMsg, a.Severity, a.Action,
	); err != nil {
		slog.Warn("waf: write alert to db failed", "error", err)
	}
}