feat(audit): Live-Stream im Dashboard via WebSocket

Recent-Activity-Karte zeigt neue audit_log-Events jetzt sofort statt
in 15s-Polls.

internal/services/audit/audit.go:
  - Repo bekommt Subscribe()-Methode mit fan-out-channel (Buffer 32,
    non-blocking-send — langsame Clients droppen Events statt die
    Pipeline zu blockieren).
  - Log() macht jetzt INSERT … RETURNING id, created_at und broadcastet
    den fertigen Entry an alle Subscribers. Broadcast nur nach
    erfolgreichem INSERT — failed inserts erscheinen nicht.

internal/handlers/audit.go:
  - Neuer GET /api/v1/audit/live (WebSocket): sendet beim Connect die
    letzten 50 Einträge (oldest→newest), danach Live-Stream aus
    Subscribe-Channel. 30s-Ping gegen HAProxy-Tunnel-Timeout.
  - Recent (Poll-Endpoint) bleibt für Fallbacks erhalten.

UI Dashboard:
  - useAuditLive(keep=15)-Hook ersetzt das 15s-useQuery-Poll.
  - WebSocket auf wss://<host>/api/v1/audit/live; Auto-Reconnect alle
    2s nach Drop.
  - dedupe per id (Snapshot + erste live-Events können sich kurz
    überschneiden während des Subscribe-Race).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Debian
2026-05-13 15:39:04 +02:00
parent 5bdea1bced
commit 4a34629023
8 changed files with 203 additions and 19 deletions

View File

@@ -1,9 +1,13 @@
package handlers
import (
"context"
"net/http"
"strconv"
"time"
"github.com/gin-gonic/gin"
"github.com/gorilla/websocket"
"git.netcell-it.de/projekte/edgeguard-native/internal/handlers/response"
"git.netcell-it.de/projekte/edgeguard-native/internal/services/audit"
@@ -18,10 +22,12 @@ func NewAuditHandler(repo *audit.Repo) *AuditHandler { return &AuditHandler{Repo
func (h *AuditHandler) Register(rg *gin.RouterGroup) {
g := rg.Group("/audit")
g.GET("/recent", h.Recent)
g.GET("/live", h.Live)
}
// Recent returns the most recent audit_log entries — used by the
// dashboard's recent-activity card. ?limit=N (1100, default 10).
// dashboard fallback path (z.B. wenn WebSocket nicht verbinden kann).
// ?limit=N (1100, default 10).
func (h *AuditHandler) Recent(c *gin.Context) {
limit := 10
if v := c.Query("limit"); v != "" {
@@ -36,3 +42,74 @@ func (h *AuditHandler) Recent(c *gin.Context) {
}
response.OK(c, gin.H{"entries": rows})
}
// auditUpgrader: same-origin durch HAProxy, kein CheckOrigin.
var auditUpgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 4 * 1024,
CheckOrigin: func(r *http.Request) bool { return true },
}
// Live upgraded auf WebSocket: sendet einen Snapshot der letzten 50
// audit_log-Rows, danach jeden neuen Eintrag direkt aus dem
// Repo.broadcast()-Channel.
func (h *AuditHandler) Live(c *gin.Context) {
conn, err := auditUpgrader.Upgrade(c.Writer, c.Request, nil)
if err != nil {
return
}
defer conn.Close()
// Snapshot
if rows, err := h.Repo.ListRecent(c.Request.Context(), 50); err == nil {
// In aufsteigender Reihenfolge schicken (newest last) damit der
// Client nach unten scrollt + neue Events natürlich anhängt.
for i := len(rows) - 1; i >= 0; i-- {
if err := conn.WriteJSON(rows[i]); err != nil {
return
}
}
}
// Live-Subscribe
ch, unsub := h.Repo.Subscribe()
defer unsub()
ctx, cancel := context.WithCancel(c.Request.Context())
defer cancel()
// Read-Loop für close-frame + ping-pong
go func() {
defer cancel()
for {
if _, _, err := conn.NextReader(); err != nil {
return
}
}
}()
ping := time.NewTicker(30 * time.Second)
defer ping.Stop()
_ = conn.SetWriteDeadline(time.Now().Add(60 * time.Second))
for {
select {
case <-ctx.Done():
return
case e, ok := <-ch:
if !ok {
return
}
_ = conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if err := conn.WriteJSON(e); err != nil {
return
}
case <-ping.C:
_ = conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
if err := conn.WriteControl(websocket.PingMessage, nil,
time.Now().Add(5*time.Second)); err != nil {
return
}
}
}
}

View File

@@ -6,6 +6,7 @@ package audit
import (
"context"
"encoding/json"
"sync"
"time"
"github.com/jackc/pgx/v5/pgxpool"
@@ -13,9 +14,48 @@ import (
type Repo struct {
Pool *pgxpool.Pool
subsMu sync.RWMutex
subs map[chan Entry]struct{}
}
func New(pool *pgxpool.Pool) *Repo { return &Repo{Pool: pool} }
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.
@@ -82,9 +122,34 @@ func (r *Repo) Log(ctx context.Context, actor, action, subject string, detail an
if nodeID == "" {
nodeArg = nil
}
_, err := r.Pool.Exec(ctx,
`INSERT INTO audit_log (actor, action, subject, detail, node_id)
VALUES ($1, $2, $3, $4, $5)`,
actor, action, subjectArg, detailJSON, nodeArg)
return err
// 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
}