Update Datenbankproblem

2026-07-23 11:13:05 +02:00
parent f9662780f7
commit 42db7d18f6
18 changed files with 1802 additions and 220 deletions
--- a/ANALYSE-COMPACT-STORAGE.md
+++ b/ANALYSE-COMPACT-STORAGE.md
@@ -0,0 +1,59 @@
+# Analyse: Warum 19,2 Mio. Zeilen nach zwei Tagen entstanden
+
+## Befund
+
+Die vorige Version war noch nicht wirklich „Metadata-first“:
+
+1. `insertBatch()` führte für **jedes** eingehende Event weiterhin ein `INSERT INTO event_logs` aus.
+2. Danach wurden zusätzlich `event_count_buckets`, `event_catalog` und `event_occurrences` geschrieben.
+3. `event_occurrences` verwendete zu viele Dimensionen für die Eindeutigkeit. Provider, Benutzer, IP, Workstation, LogonType, Status und FailureReason konnten die Aggregation stark fragmentieren.
+4. Mehrere UEBA-/Detection-Regeln lasen weiterhin aus `event_logs`.
+5. `/ui` las die letzten Events und den 24h-Eventzähler aus `event_occurrences`. Bei hoher Kardinalität wurde damit bereits die Startseite teuer.
+
+19,2 Mio. Events in 48 Stunden entsprechen im Mittel ungefähr 111 Events/s. Diese Rate ist für kleine aggregierte Upserts nicht das Kernproblem. Eine breite Einzelzeile mit mehreren Sekundärindizes pro Event dagegen erzeugt sehr viel Daten- und Index-I/O.
+
+## Korrektur
+
+### Universal: `event_count_buckets`
+
+Für jedes Event wird nur noch ein Counter auf
+
+`5-Minuten-Bucket × Host × Channel × Event-ID`
+
+hochgezählt. Ein Host mit 200 verschiedenen Event-IDs erzeugt damit theoretisch höchstens etwa 57.600 Counter-Zeilen pro Tag, selbst wenn dahinter Millionen reale Events liegen (200 IDs × 288 Buckets), praktisch meist deutlich weniger.
+
+### Kontext nur bei Bedarf
+
+`event_occurrences` wird nur für eine konfigurierbare Allowlist von Security-IDs geschrieben. Der Dimensionsschlüssel enthält nur stabile, detection-relevante Werte. Beispiel 4740: Zieluser und CallerComputer/Workstation; nicht FailureReason, Status, IP oder Subject-Machine-Account.
+
+### Keine Einzelzeilen im Normalbetrieb
+
+`STORE_EVENT_ROWS=false` ist der neue Standard. `event_logs` wird dann überhaupt nicht beschrieben. `STORE_RAW_XML=true` aktiviert implizit wieder Full-Event-Zeilen und ist deshalb nur für kurze, gezielte Forensik gedacht.
+
+### UI
+
+`/ui` verwendet nur `event_count_buckets`. Die Eventliste hat ohne explizites `from` automatisch ein 24h-Fenster. User-/IP-Suchen wechseln gezielt auf die kleine Kontexttabelle.
+
+## Sofortmaßnahme für die bestehende Datenbank
+
+Nach Deployment des neuen Backends:
+
+```sql
+SOURCE deploy/mariadb/migrations/004-compact-storage.sql;
+SOURCE deploy/mariadb/emergency-compact-cleanup.sql;
+SOURCE deploy/mariadb/cardinality-diagnostics.sql;
+```
+
+Der Emergency-Cleanup löscht absichtlich die redundante alte Full-Event-/Kontext-Historie, nicht jedoch `event_count_buckets`, `event_catalog`, Detections oder Baselines.
+
+## Retention
+
+Die detaillierten Zeit-Buckets werden standardmäßig 30 Tage gehalten. `event_catalog` bleibt dauerhaft und behält First/Last-Seen plus Gesamtzähler. Damit wächst die Zeitreihenhistorie nicht unbegrenzt.
+
+## Erwartetes Monitoring
+
+Der wichtigste Wert ist künftig nicht `COUNT(*)` allein, sondern das Verhältnis:
+
+`SUM(cnt) / COUNT(*)`
+
+in den Aggregattabellen. Je höher dieser Wert, desto mehr reale Events werden durch eine Datenbankzeile repräsentiert. `cardinality-diagnostics.sql` zeigt diesen Wert für die letzten 24 Stunden.
--- a/ANALYSE-METADATA-FIRST.md
+++ b/ANALYSE-METADATA-FIRST.md
@@ -0,0 +1,104 @@
+> **Überholt:** Dieses Dokument beschreibt die erste Metadata-first-Iteration.
+> Für den produktiven Stand nach dem 19,2-Mio.-Zeilen-Befund gilt
+> `ANALYSE-COMPACT-STORAGE.md`. Insbesondere werden `event_logs` jetzt
+> standardmäßig gar nicht mehr geschrieben und `event_occurrences` ist nur noch
+> eine selektive Security-Kontexttabelle.
+
+# Untersuchung und Anpassung: Metadata-first
+
+## Befund
+
+Die Performanceprobleme kommen nicht nur von „zu viel XML“, sondern aus mehreren sich verstärkenden Ursachen:
+
+1. Jedes Event wurde breit in `event_logs` und zusätzlich vollständig in `event_log_raw` gespeichert.
+2. Die Partitionswartung verwendete `event_logs_raw`, die Tabelle heißt jedoch `event_log_raw`. Der Lauf brach damit ab; alte Partitionen wurden nicht gelöscht.
+3. `event_count_buckets` und `ueba_context_buckets` waren nicht partitioniert und wuchsen unbegrenzt.
+4. `target_user_norm` und `subject_user_norm` waren im Schema und in Indizes vorgesehen, wurden beim Insert aber nicht befüllt.
+5. Dashboard und Event-UI lasen direkt aus der größten Einzeltabelle. Die Event-Filter enthielten zusätzlich eine korrelierte `EXISTS`-Abfrage auf Detections.
+6. Die „neue Event-ID“-Regel verwendete einen teuren `NOT EXISTS`-Vergleich gegen die gesamte Eventhistorie.
+7. Zählerregeln wie Failed-Logon-Spike, Reboot-Spike und Password-Spray zählten Einzelevents, obwohl dafür Aggregate ausreichen.
+
+Ohne Zugriff auf die produktive Datenbank konnte ich keine realen `EXPLAIN ANALYZE`-Pläne oder Größenverteilungen messen. Die oben genannten Punkte ergeben sich direkt aus Schema und Abfragepfaden im Projekt.
+
+## Umgesetztes Zielmodell
+
+### Hot Data
+
+`event_logs` enthält weiterhin kompakte Einzelevent-Metadaten, weil einige Regeln die genaue Reihenfolge benötigen, etwa „erfolgreicher Login nach Fehlversuchen“. Die Standardaufbewahrung beträgt 72 Stunden.
+
+### Langfristige Metadaten
+
+`event_occurrences` speichert pro Zeit-Bucket und Dimensionskombination:
+
+- Host, Channel, Event-ID und Provider
+- Target-/Subject-User
+- Source-IP und Workstation/Gerät
+- Logon-Type, Status und Fehlergrund
+- Anzahl sowie erstes und letztes Auftreten
+
+Die UI liest Bucket-Zeilen direkt mit `ORDER BY bucket_start DESC LIMIT ...`; dadurch bleibt die Abfrage index- und partitionsfreundlich und benötigt keinen großen `GROUP BY`.
+
+### Event-Katalog
+
+`event_catalog` enthält pro Host/Channel/Event-ID nur erstes Auftreten, letztes Auftreten und Gesamtzahl. Die „neue Event-ID“-Regel benötigt dadurch keinen Anti-Join mehr.
+
+### Raw XML
+
+Raw-XML ist mit `STORE_RAW_XML=false` standardmäßig deaktiviert. Bei temporärem forensischem Bedarf kann es aktiviert und mit eigener kurzer `RAW_RETENTION` betrieben werden.
+
+### Direkter Metadaten-Ingest
+
+Collector können weiterhin XML in `msg` senden. Neu ist ein `meta`-Objekt, das ohne XML auskommt. Für Event 4740 wird `CallerComputerName` als Workstation/Gerät verarbeitet.
+
+## Erwartete Wirkung
+
+- Wegfall der dauerhaften XML-Doppelhaltung.
+- Garantierte Aufbewahrungsgrenzen durch korrigierte und erweiterte Partitionswartung.
+- Deutlich weniger Zeilen in langfristigen Abfragen durch Zeit-Buckets.
+- Schnellere Dashboard-/Event-Seiten, da sie nicht mehr `event_logs` durchsuchen.
+- Schnellere Zählerregeln durch `SUM(cnt)` auf `event_occurrences`.
+- Kleine, konstante Abfrage für neue Event-IDs über `event_catalog`.
+- Nutzbare User-Indizes, da normalisierte User-Spalten nun tatsächlich befüllt und in wichtigen Regeln verwendet werden.
+
+Die tatsächliche Kompressionswirkung hängt von der Kardinalität ab. Viele identische Events innerhalb einer Minute werden stark verdichtet. Einzigartige Kombinationen aus User, IP und Workstation reduzieren sich weniger, bleiben aber deutlich kleiner als Voll-XML plus breite Einzeleventzeile.
+
+## Lockout-Abfrage
+
+```sql
+SELECT
+    first_event_ts,
+    last_event_ts,
+    cnt,
+    hostname AS reporting_host,
+    target_user,
+    workstation AS caller_device,
+    status_text,
+    failure_reason
+FROM event_occurrences
+WHERE channel_name = 'Security'
+  AND event_id = 4740
+  AND bucket_start >= UTC_TIMESTAMP() - INTERVAL 7 DAY
+ORDER BY bucket_start DESC;
+```
+
+Bei einem Domain-Lockout ist `hostname` typischerweise der meldende Domain Controller; `workstation` enthält – sofern im Event vorhanden – `CallerComputerName` und damit das verursachende Gerät.
+
+## Empfohlener Rollout
+
+1. Datenbank sichern und Ingest stoppen.
+2. Tabellen- und Partitionsgrößen mit `deploy/mariadb/diagnostics.sql` erfassen.
+3. `deploy/mariadb/migrations/002-metadata-first.sql` in einem Wartungsfenster ausführen.
+4. `.env.example` nach `.env` kopieren beziehungsweise bestehende Secrets übernehmen und die neuen Variablen ergänzen.
+5. Backend deployen, zunächst `STORE_RAW_XML=false`, `EVENT_RETENTION=72h`, `METADATA_BUCKET=1m`.
+6. Partition-Maintenance-Logs und Datenbankgrößen kontrollieren.
+7. Nach 3–7 Tagen Schwellenwerte und Bucketgröße anhand der realen Eventrate nachjustieren.
+
+## Wann ein Wechsel der Datenbank sinnvoll wäre
+
+MariaDB ist für dieses Metadatenmodell weiterhin geeignet. Ein Wechsel zu ClickHouse wäre interessant, wenn dauerhaft sehr hohe Raten und große analytische Zeitreihen über Jahre benötigt werden. OpenSearch wäre sinnvoll, wenn Volltextsuche im Rohinhalt eine Kernanforderung bleibt. Für „welches Event, wann, auf welchem Gerät, für welchen User und wie oft“ ist ein Engine-Wechsel derzeit nicht erforderlich.
+
+## Ergänzung: `new_event_id` ist Inventar, nicht automatisch Incident
+
+Die bisherige Regel hat jedes erste Auftreten pauschal als offenen Medium-Alarm behandelt. Das überschätzt insbesondere diagnostische und operative Channels. Die neue Standardklassifikation ist `inventory`: nach Lernphase und Mindestanzahl entsteht höchstens eine plausible Info-Meldung. `Microsoft-Windows-WMI-Activity/Operational` wird standardmäßig ignoriert. Nur explizit konfigurierte High-Risk-IDs oder der Betriebsmodus `alert` können eine offene Detection erzeugen.
+
+Für reproduzierbare Lasttests liegt unter `cmd/siem-stress-agent` ein begrenzter Metadaten-Agent. Er markiert sämtliche Daten eindeutig, benötigt `--confirm-load-test` und besitzt harte Maximalwerte für Rate, Laufzeit, Worker und Eventzahl.
--- a/Dockerfile.stress
+++ b/Dockerfile.stress
@@ -0,0 +1,10 @@
+FROM golang:1.26 AS builder
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY cmd/siem-stress-agent ./cmd/siem-stress-agent
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -trimpath -ldflags="-s -w" -o /out/siem-stress-agent ./cmd/siem-stress-agent
+
+FROM gcr.io/distroless/static-debian12:nonroot
+COPY --from=builder /out/siem-stress-agent /siem-stress-agent
+ENTRYPOINT ["/siem-stress-agent"]
--- a/README.md
+++ b/README.md
@@ -1,22 +1,57 @@
-# SIEM Backend – Metadata-first Variante
+# SIEM Backend – Compact Metadata Variante

-## Warum die Datenbank bisher wächst
+## Ursache des erneuten Datenbankwachstums

-Die ursprüngliche Implementierung speichert jedes Windows-Event zweimal: als breite normalisierte Zeile in `event_logs` und zusätzlich als vollständiges XML in `event_log_raw`. Außerdem enthielt die Partitionswartung einen Tabellennamenfehler (`event_logs_raw` statt `event_log_raw`). Der Wartungslauf brach deshalb ab, bevor alte Partitionen gelöscht wurden. Auch `event_count_buckets` und `ueba_context_buckets` waren nicht partitioniert und konnten unbegrenzt wachsen.
+Die vorherige „Metadata-first“-Variante hatte noch einen entscheidenden Fehler: `insertBatch()` schrieb weiterhin **jedes einzelne Event** in `event_logs`. Zusätzlich war der Schlüssel von `event_occurrences` zu fein (u. a. Benutzer, IP, Workstation, Status und FailureReason), sodass diese Tabelle je nach Eventquelle fast wieder Event-Kardinalität erreichen konnte.
+
+Bei 19,2 Mio. Events in zwei Tagen entspricht der Eingang im Mittel rund 9,6 Mio. Events/Tag bzw. 111 Events/Sekunde. Für eine MariaDB ist diese Eventrate als reine Aggregatlast unkritisch; problematisch war die Speicherung und spätere Abfrage nahezu jeder Einzelzeile.

 ## Neues Speichermodell

- `event_logs`: kurzlebige Hot-Daten für Regeln, die eine genaue Reihenfolge einzelner Events benötigen. Standard: 72 Stunden.
- `event_occurrences`: langfristige Metadaten-Aggregate pro Zeit-Bucket, Host, Channel, Event-ID, Benutzer, IP und Workstation. Enthält `cnt`, `first_event_ts` und `last_event_ts`. Standard: 180 Tage.
- `event_catalog`: kleine, dauerhafte Landkarte des ersten und letzten Auftretens jeder Event-ID pro Host und Channel.
- `event_log_raw`: optionales Raw-XML. Standardmäßig deaktiviert; bei Aktivierung nur 24 Stunden Aufbewahrung.
- `event_count_buckets` und `ueba_context_buckets`: jetzt partitioniert und automatisch bereinigt.
+- `event_count_buckets`: **universeller Primärspeicher für alle Events**. Eine Zeile pro 5-Minuten-Bucket, Host, Channel und Event-ID mit `cnt`, erstem und letztem Auftreten.
+- `event_catalog`: dauerhafte kleine Landkarte pro Host/Channel/Event-ID.
+- `event_occurrences`: nur noch **ausgewählte Security-Kontexte** (z. B. 4624, 4625, 4672, 4740). Benutzer/IP/Workstation werden nur dort gespeichert, wo sie für Detection/Forensik benötigt werden.
+- `event_logs`: Full-Event-Hotstore ist jetzt standardmäßig **aus** (`STORE_EVENT_ROWS=false`).
+- `event_log_raw`: Raw-XML ist standardmäßig **aus** (`STORE_RAW_XML=false`). Raw-XML impliziert technisch Full-Event-Zeilen und sollte nur für kurze gezielte Forensik aktiviert werden.

-Für einen User-Lockout (typischerweise Security Event 4740) kann die Anwendung damit direkt beantworten: wann er zuerst/zuletzt im Bucket auftrat, auf welchem Host, für welchen Benutzer und wie oft. Das vollständige XML ist dafür nicht erforderlich. `CallerComputerName` wird dabei als Gerät/Workstation übernommen.
+Für einen Lockout (Security 4740) bleiben damit beispielsweise Event-ID, Host, Zielbenutzer, CallerComputer/Workstation, Anzahl sowie erstes/letztes Auftreten erhalten. Status-/Failure-Text, Source-IP und Subject-Machine-Account werden bei 4740 bewusst nicht zur Dimensionsbildung verwendet.
+
+## Produktionskonfiguration
+
+```env
+STORE_EVENT_ROWS=false
+STORE_RAW_XML=false
+METADATA_CONTEXT_EVENT_IDS=4624,4625,4648,4672,4720,4726,4728,4732,4740,4756,4768,4769,4771,4776
+BASELINE_WINDOW=5m
+METADATA_BUCKET=5m
+EVENT_RETENTION=6h
+RAW_RETENTION=6h
+METADATA_RETENTION=720h
+PARTITION_MAINTENANCE_ENABLED=true
+```
+
+`EVENT_RETENTION` und `RAW_RETENTION` spielen im empfohlenen Modus keine Rolle, weil diese Tabellen nicht mehr beschrieben werden. Sie begrenzen lediglich die Datenmenge, falls Full-Event-/Raw-Speicherung bewusst aktiviert wird.
+
+## UI und Detection
+
+`/ui` liest die 24h-Eventzahl und die letzten Events jetzt ausschließlich aus `event_count_buckets`. Ungefilterte Eventseiten erhalten außerdem automatisch ein 24h-Zeitfenster. Damit berührt die Startseite weder `event_logs` noch die detaillierte Kontexttabelle.
+
+Failed-Logon-Spikes und Reboot-Spikes verwenden die universellen Counter. Password-Spray, Success-after-Failures, neue Source-IP und UEBA verwenden die kompakte Security-Kontexttabelle. Dynamische Regeln auf `hostname`, `channel`, `event_id`, `target_user`, `subject_user`, `src_ip` und `workstation` funktionieren ohne Full-Event-Store. Regeln auf `msg` oder `process_name` benötigen weiterhin `STORE_EVENT_ROWS=true` und werden ansonsten bewusst übersprungen und geloggt.
+
+## Bestehende Installation retten
+
+1. Datenbank-Snapshot/Backup erstellen und Ingest kurz stoppen.
+2. Neues Backend deployen und sicherstellen, dass `STORE_EVENT_ROWS=false` sowie `STORE_RAW_XML=false` gesetzt sind.
+3. `deploy/mariadb/migrations/004-compact-storage.sql` ausführen.
+4. Backend starten und im Log die Zeile `storage mode: store_event_rows=false store_raw_xml=false ...` prüfen.
+5. Erst danach `deploy/mariadb/emergency-compact-cleanup.sql` ausführen. Das Script leert `event_logs`, `event_log_raw` und die alte hochkardinale `event_occurrences`-Historie. `event_count_buckets`, `event_catalog`, Detections und Baselines bleiben erhalten.
+6. Mit `deploy/mariadb/cardinality-diagnostics.sql` prüfen, wie viele reale Events eine Aggregatzeile repräsentiert.
+
+Der Cleanup ist bewusst nicht Bestandteil der normalen Migration, da er historische Full-Event-/Kontextdaten löscht.

 ## Bevorzugtes Metadaten-Ingest

-Bestehende Collector dürfen weiterhin `msg` mit XML senden. Neue oder angepasste Collector können stattdessen direkt ein `meta`-Objekt senden; dann muss das Backend das XML weder speichern noch parsen:
+Collector können weiterhin XML senden, besser ist aber das strukturierte `meta`-Objekt. Das Backend extrahiert daraus nur die für die jeweilige Event-ID benötigten Dimensionen.

 ```json
 [
@@ -28,7 +63,6 @@ Bestehende Collector dürfen weiterhin `msg` mit XML senden. Neue oder angepasst
    "ts": "2026-07-18T12:34:56Z",
    "meta": {
      "target_user": "alice",
-      "subject_user": "DC01$",
      "device": "CLIENT-42",
      "provider": "Microsoft-Windows-Security-Auditing"
    }
@@ -36,38 +70,19 @@ Bestehende Collector dürfen weiterhin `msg` mit XML senden. Neue oder angepasst
 ]
 ```

-Mindestens `msg` oder `meta` ist erforderlich. Für Regeln, die auf `msg`/Volltext prüfen, muss Raw-XML weiterhin vom Collector geliefert werden; im reinen Metadatenmodus sollten solche Regeln auf strukturierte Felder umgestellt werden.
-
-## Wichtige Einstellungen
-
-```env
-STORE_RAW_XML=false
-METADATA_BUCKET=1m
-EVENT_RETENTION=72h
-RAW_RETENTION=24h
-METADATA_RETENTION=4320h
-PARTITION_MAINTENANCE_ENABLED=true
-```
-
-`METADATA_BUCKET=1m` ist ein guter Ausgangspunkt. Bei sehr hohen Raten kann auf `5m` erhöht werden; dadurch sinkt die Zeilenzahl weiter, die zeitliche Auflösung wird aber gröber.
-
-## Bestehende Installation migrieren
-
-1. Datenbank sichern und Ingest vorübergehend stoppen.
-2. `deploy/mariadb/migrations/002-metadata-first.sql` in einem Wartungsfenster ausführen. Das Script wärmt `event_catalog` aus den kleineren Baseline-Buckets vor, damit bekannte Event-IDs nicht als neu alarmiert werden. Die beiden `ALTER TABLE ... PARTITION BY`-Operationen können große Tabellen neu aufbauen und sperren.
-3. Neue Umgebungsvariablen aus `dot_env` übernehmen.
-4. Backend aktualisieren und starten.
-5. Im Log kontrollieren, dass `partition maintenance completed` erscheint. Die frühere Meldung zur nicht vorhandenen Tabelle `event_logs_raw` darf nicht mehr auftreten.
-6. Nach Ablauf von `EVENT_RETENTION` prüfen, ob alte `event_logs`-Partitionen verschwinden. Raw-XML kann nach erfolgreicher Validierung separat gelöscht oder durch die kurze `RAW_RETENTION` automatisch entfernt werden. `event_occurrences` wird nicht rückwirkend aus der alten Volltabelle aufgebaut; die langfristige Metadatenhistorie beginnt mit dem neuen Backend.
-
 ## Datenbankwahl

-MariaDB bleibt für dieses Modell sinnvoll: Die Abfragen sind überwiegend zeitbasierte Filter, gruppierte Zähler und kleine relationale Dimensionen. Ein Wechsel zu ClickHouse oder OpenSearch lohnt sich erst, wenn trotz Aggregation sehr hohe Eventraten, ad-hoc Volltextsuchen oder jahrelange Rohdatenhaltung erforderlich sind. Für die hier beschriebene Metadatenanforderung wäre ein sofortiger Engine-Wechsel zusätzlicher Betriebsaufwand ohne zwingenden Nutzen.
+MariaDB bleibt für dieses Modell sinnvoll. Der entscheidende Punkt ist, dass die Datenbank nicht mehr als Roh-Event-Store verwendet wird. Die langfristige Last besteht nun überwiegend aus Upserts auf Zeit-Buckets und kleinen relationalen Dimensionen. Ein Wechsel zu ClickHouse wäre erst dann sinnvoll, wenn sehr große Rohdatenmengen oder breite Ad-hoc-Analysen ausdrücklich wieder Teil des Ziels werden.

 ---

 # SIEM-lite Admin-Handbuch: Anlernphase, manuelle Bewertung und Lerneffekt

+> **Architekturhinweis:** Das folgende ältere Handbuch enthält an einigen Stellen
+> noch SQL-/Tabellenbeispiele aus der Full-Event-Architektur. Für Speichermodell,
+> Performance und aktuelle Detection-Datenquellen ist ausschließlich der Abschnitt
+> „Compact Metadata Variante“ oben maßgeblich.
+
 Stand: 2026-04-27  
 Zielgruppe: Administratoren ohne SIEM-/SOC-Vorerfahrung  
 Projekt: `siem-backend` / `SIEM-lite Security Operations`
--- a/cmd/siem-stress-agent/main.go
+++ b/cmd/siem-stress-agent/main.go
@@ -0,0 +1,476 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"flag"
+	"fmt"
+	"io"
+	"math"
+	"net/http"
+	"net/url"
+	"os"
+	"os/signal"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+)
+
+const (
+	maxRate     = 100_000
+	maxWorkers  = 64
+	maxDuration = time.Hour
+	maxEvents   = 5_000_000
+)
+
+type config struct {
+	Endpoint      string
+	Hostname      string
+	APIKey        string
+	EnrollmentKey string
+	Scenario      string
+	Rate          int
+	BatchSize     int
+	Workers       int
+	Duration      time.Duration
+	EventLimit    int64
+	Timeout       time.Duration
+	Confirm       bool
+}
+
+type event struct {
+	Hostname string    `json:"host"`
+	Channel  string    `json:"channel"`
+	EventID  uint32    `json:"id"`
+	Source   string    `json:"source"`
+	Time     time.Time `json:"ts"`
+	Metadata metadata  `json:"meta"`
+}
+
+type metadata struct {
+	Computer     string `json:"computer,omitempty"`
+	ProviderName string `json:"provider,omitempty"`
+	TargetUser   string `json:"target_user,omitempty"`
+	TargetDomain string `json:"target_domain,omitempty"`
+	SubjectUser  string `json:"subject_user,omitempty"`
+	Workstation  string `json:"workstation,omitempty"`
+	Device       string `json:"device,omitempty"`
+	SrcIP        string `json:"src_ip,omitempty"`
+	SrcPort      string `json:"src_port,omitempty"`
+	LogonType    string `json:"logon_type,omitempty"`
+	ProcessName  string `json:"process_name,omitempty"`
+	StatusText   string `json:"status,omitempty"`
+	SubStatus    string `json:"sub_status,omitempty"`
+	Failure      string `json:"failure_reason,omitempty"`
+}
+
+type result struct {
+	Events   int
+	Duration time.Duration
+	Status   int
+	Err      error
+}
+
+type stats struct {
+	requests  atomic.Int64
+	accepted  atomic.Int64
+	failed    atomic.Int64
+	bytesSent atomic.Int64
+	mu        sync.Mutex
+	latencies []time.Duration
+	statuses  map[int]int64
+	errors    map[string]int64
+}
+
+func main() {
+	cfg := parseFlags()
+	if err := validateConfig(cfg); err != nil {
+		fmt.Fprintln(os.Stderr, "Konfigurationsfehler:", err)
+		os.Exit(2)
+	}
+
+	runCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	produceCtx, cancelProduction := context.WithTimeout(runCtx, cfg.Duration)
+	defer cancelProduction()
+
+	transport := &http.Transport{
+		MaxIdleConns:        cfg.Workers * 2,
+		MaxIdleConnsPerHost: cfg.Workers,
+		IdleConnTimeout:     30 * time.Second,
+	}
+	client := &http.Client{Transport: transport, Timeout: cfg.Timeout}
+	defer transport.CloseIdleConnections()
+
+	jobs := make(chan []event, cfg.Workers*2)
+	results := make(chan result, cfg.Workers*2)
+	st := &stats{statuses: make(map[int]int64), errors: make(map[string]int64)}
+
+	var workers sync.WaitGroup
+	for i := 0; i < cfg.Workers; i++ {
+		workers.Add(1)
+		go func() {
+			defer workers.Done()
+			for batch := range jobs {
+				results <- sendBatch(runCtx, client, cfg, batch)
+			}
+		}()
+	}
+
+	collectorDone := make(chan struct{})
+	go func() {
+		defer close(collectorDone)
+		for r := range results {
+			st.requests.Add(1)
+			st.mu.Lock()
+			st.latencies = append(st.latencies, r.Duration)
+			if r.Status != 0 {
+				st.statuses[r.Status]++
+			}
+			if r.Err != nil {
+				st.failed.Add(int64(r.Events))
+				st.errors[shortError(r.Err)]++
+			} else if r.Status >= 200 && r.Status < 300 {
+				st.accepted.Add(int64(r.Events))
+			} else {
+				st.failed.Add(int64(r.Events))
+			}
+			st.mu.Unlock()
+		}
+	}()
+
+	fmt.Printf("SIEM-Stresstest startet: endpoint=%s host=%s scenario=%s rate=%d/s batch=%d workers=%d duration=%s limit=%d\n",
+		cfg.Endpoint, cfg.Hostname, cfg.Scenario, cfg.Rate, cfg.BatchSize, cfg.Workers, cfg.Duration, cfg.EventLimit)
+	started := time.Now()
+	produced := produce(produceCtx, cfg, jobs, st)
+	close(jobs)
+	workers.Wait()
+	close(results)
+	<-collectorDone
+
+	printReport(st, produced, time.Since(started))
+	if st.failed.Load() > 0 {
+		os.Exit(1)
+	}
+}
+
+func parseFlags() config {
+	var cfg config
+	flag.StringVar(&cfg.Endpoint, "url", getenv("SIEM_STRESS_URL", "http://127.0.0.1:8090/ingest"), "vollständige Ingest-URL")
+	flag.StringVar(&cfg.Hostname, "host", getenv("SIEM_STRESS_HOST", "SIEM-STRESS-01"), "Test-Hostname; alle Events eines Batches verwenden diesen Host")
+	flag.StringVar(&cfg.APIKey, "api-key", os.Getenv("SIEM_STRESS_API_KEY"), "API-Key des Test-Agenten")
+	flag.StringVar(&cfg.EnrollmentKey, "enrollment-key", os.Getenv("SIEM_STRESS_ENROLLMENT_KEY"), "Enrollment-Key für die erste Registrierung")
+	flag.StringVar(&cfg.Scenario, "scenario", getenv("SIEM_STRESS_SCENARIO", "mixed"), "mixed, failed-logon, lockout, normal oder catalog")
+	flag.IntVar(&cfg.Rate, "rate", getenvInt("SIEM_STRESS_RATE", 200), "Zielrate in Events pro Sekunde")
+	flag.IntVar(&cfg.BatchSize, "batch", getenvInt("SIEM_STRESS_BATCH", 100), "Events pro HTTP-Request, maximal 1000")
+	flag.IntVar(&cfg.Workers, "workers", getenvInt("SIEM_STRESS_WORKERS", 4), "parallele HTTP-Worker")
+	flag.DurationVar(&cfg.Duration, "duration", getenvDuration("SIEM_STRESS_DURATION", 30*time.Second), "maximale Laufzeit")
+	flag.Int64Var(&cfg.EventLimit, "max-events", getenvInt64("SIEM_STRESS_MAX_EVENTS", 100_000), "zusätzliche harte Obergrenze der erzeugten Events")
+	flag.DurationVar(&cfg.Timeout, "timeout", getenvDuration("SIEM_STRESS_TIMEOUT", 20*time.Second), "HTTP-Timeout pro Request")
+	flag.BoolVar(&cfg.Confirm, "confirm-load-test", getenvBool("SIEM_STRESS_CONFIRM", false), "bestätigt, dass das Zielsystem für diesen Lasttest autorisiert ist")
+	flag.Parse()
+	return cfg
+}
+
+func validateConfig(cfg config) error {
+	if !cfg.Confirm {
+		return errors.New("--confirm-load-test fehlt")
+	}
+	u, err := url.Parse(cfg.Endpoint)
+	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
+		return errors.New("--url muss eine gültige HTTP(S)-Ingest-URL sein")
+	}
+	if strings.TrimSpace(cfg.Hostname) == "" || len(cfg.Hostname) > 191 {
+		return errors.New("--host ist leer oder länger als 191 Zeichen")
+	}
+	if strings.TrimSpace(cfg.APIKey) == "" {
+		return errors.New("--api-key ist erforderlich")
+	}
+	if cfg.Rate < 1 || cfg.Rate > maxRate {
+		return fmt.Errorf("--rate muss zwischen 1 und %d liegen", maxRate)
+	}
+	if cfg.BatchSize < 1 || cfg.BatchSize > 1000 {
+		return errors.New("--batch muss zwischen 1 und 1000 liegen")
+	}
+	if cfg.Workers < 1 || cfg.Workers > maxWorkers {
+		return fmt.Errorf("--workers muss zwischen 1 und %d liegen", maxWorkers)
+	}
+	if cfg.Duration <= 0 || cfg.Duration > maxDuration {
+		return fmt.Errorf("--duration muss größer 0 und höchstens %s sein", maxDuration)
+	}
+	if cfg.EventLimit < 1 || cfg.EventLimit > maxEvents {
+		return fmt.Errorf("--max-events muss zwischen 1 und %d liegen", maxEvents)
+	}
+	if cfg.Timeout <= 0 || cfg.Timeout > 2*time.Minute {
+		return errors.New("--timeout muss größer 0 und höchstens 2m sein")
+	}
+	switch strings.ToLower(cfg.Scenario) {
+	case "mixed", "failed-logon", "lockout", "normal", "catalog":
+	default:
+		return errors.New("unbekanntes --scenario")
+	}
+	return nil
+}
+
+func produce(ctx context.Context, cfg config, jobs chan<- []event, st *stats) int64 {
+	batchInterval := time.Duration(float64(time.Second) * float64(cfg.BatchSize) / float64(cfg.Rate))
+	if batchInterval < time.Millisecond {
+		batchInterval = time.Millisecond
+	}
+	ticker := time.NewTicker(batchInterval)
+	defer ticker.Stop()
+
+	var produced int64
+	var sequence int64
+	for produced < cfg.EventLimit {
+		remaining := cfg.EventLimit - produced
+		n := cfg.BatchSize
+		if int64(n) > remaining {
+			n = int(remaining)
+		}
+		batch := make([]event, n)
+		now := time.Now().UTC()
+		for i := range batch {
+			sequence++
+			batch[i] = generateEvent(cfg, sequence, now.Add(time.Duration(i)*time.Microsecond))
+		}
+		encoded, _ := json.Marshal(batch)
+		st.bytesSent.Add(int64(len(encoded)))
+		select {
+		case jobs <- batch:
+			produced += int64(n)
+		case <-ctx.Done():
+			return produced
+		}
+		select {
+		case <-ticker.C:
+		case <-ctx.Done():
+			return produced
+		}
+	}
+	return produced
+}
+
+func generateEvent(cfg config, seq int64, ts time.Time) event {
+	base := event{
+		Hostname: cfg.Hostname,
+		Source:   "SIEM-Stress-Agent",
+		Time:     ts,
+		Metadata: metadata{
+			Computer:     cfg.Hostname,
+			ProviderName: "SIEM-Stress-Agent",
+			TargetDomain: "STRESS",
+			SubjectUser:  "stress-agent$",
+			Workstation:  fmt.Sprintf("STRESS-CLIENT-%03d", seq%25),
+			Device:       fmt.Sprintf("STRESS-CLIENT-%03d", seq%25),
+			SrcIP:        fmt.Sprintf("198.18.%d.%d", (seq/250)%250, seq%250+1),
+			SrcPort:      strconv.FormatInt(40000+seq%20000, 10),
+			ProcessName:  `C:\\Program Files\\SIEM-Stress-Agent\\stress.exe`,
+		},
+	}
+
+	scenario := strings.ToLower(cfg.Scenario)
+	if scenario == "mixed" {
+		switch seq % 10 {
+		case 0, 1, 2, 3:
+			scenario = "failed-logon"
+		case 4:
+			scenario = "lockout"
+		case 5, 6, 7:
+			scenario = "normal"
+		case 8:
+			base.Channel, base.EventID = "System", 1074
+			return base
+		default:
+			base.Channel, base.EventID = "Microsoft-Windows-WMI-Activity/Operational", 5857
+			return base
+		}
+	}
+
+	switch scenario {
+	case "failed-logon":
+		base.Channel, base.EventID = "Security", 4625
+		base.Metadata.TargetUser = fmt.Sprintf("stress-user-%03d", seq%100)
+		base.Metadata.LogonType = "3"
+		base.Metadata.StatusText = "0xC000006D"
+		base.Metadata.SubStatus = "0xC000006A"
+		base.Metadata.Failure = "Unknown user name or bad password"
+	case "lockout":
+		base.Channel, base.EventID = "Security", 4740
+		base.Metadata.TargetUser = fmt.Sprintf("stress-lockout-%03d", seq%20)
+	case "normal":
+		base.Channel, base.EventID = "Security", 4624
+		base.Metadata.TargetUser = fmt.Sprintf("stress-user-%03d", seq%100)
+		base.Metadata.LogonType = "3"
+	case "catalog":
+		base.Channel = "SIEM-Stress-Agent/Operational"
+		base.EventID = uint32(50_000 + seq%5_000)
+		base.Metadata.TargetUser = fmt.Sprintf("catalog-user-%03d", seq%50)
+	}
+	return base
+}
+
+func sendBatch(ctx context.Context, client *http.Client, cfg config, batch []event) result {
+	body, err := json.Marshal(batch)
+	if err != nil {
+		return result{Events: len(batch), Err: err}
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.Endpoint, bytes.NewReader(body))
+	if err != nil {
+		return result{Events: len(batch), Err: err}
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-API-Key", cfg.APIKey)
+	if cfg.EnrollmentKey != "" {
+		req.Header.Set("X-Enrollment-Key", cfg.EnrollmentKey)
+	}
+	req.Header.Set("User-Agent", "siem-stress-agent/1.0")
+
+	started := time.Now()
+	resp, err := client.Do(req)
+	duration := time.Since(started)
+	if err != nil {
+		return result{Events: len(batch), Duration: duration, Err: err}
+	}
+	defer resp.Body.Close()
+	limited, readErr := io.ReadAll(io.LimitReader(resp.Body, 4096))
+	if readErr != nil {
+		return result{Events: len(batch), Duration: duration, Status: resp.StatusCode, Err: readErr}
+	}
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		return result{Events: len(batch), Duration: duration, Status: resp.StatusCode, Err: fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(limited)))}
+	}
+	return result{Events: len(batch), Duration: duration, Status: resp.StatusCode}
+}
+
+func printReport(st *stats, produced int64, elapsed time.Duration) {
+	st.mu.Lock()
+	latencies := append([]time.Duration(nil), st.latencies...)
+	statuses := cloneMap(st.statuses)
+	errorsByText := cloneStringMap(st.errors)
+	st.mu.Unlock()
+	sort.Slice(latencies, func(i, j int) bool { return latencies[i] < latencies[j] })
+
+	accepted := st.accepted.Load()
+	failed := st.failed.Load()
+	throughput := float64(accepted) / math.Max(elapsed.Seconds(), 0.001)
+	fmt.Println("\n--- Ergebnis ---")
+	fmt.Printf("Laufzeit:          %s\n", elapsed.Round(time.Millisecond))
+	fmt.Printf("Erzeugte Events:   %d\n", produced)
+	fmt.Printf("Akzeptierte Events:%d\n", accepted)
+	fmt.Printf("Fehlgeschlagen:    %d\n", failed)
+	fmt.Printf("HTTP-Requests:     %d\n", st.requests.Load())
+	fmt.Printf("Durchsatz:         %.1f Events/s\n", throughput)
+	fmt.Printf("JSON gesendet:     %.2f MiB\n", float64(st.bytesSent.Load())/(1024*1024))
+	if len(latencies) > 0 {
+		fmt.Printf("Latenz p50/p95/p99:%s / %s / %s\n", percentile(latencies, 0.50), percentile(latencies, 0.95), percentile(latencies, 0.99))
+	}
+	fmt.Printf("HTTP-Status:       %v\n", statuses)
+	if len(errorsByText) > 0 {
+		fmt.Printf("Fehler:            %v\n", errorsByText)
+	}
+}
+
+func percentile(values []time.Duration, p float64) time.Duration {
+	if len(values) == 0 {
+		return 0
+	}
+	idx := int(math.Ceil(float64(len(values))*p)) - 1
+	if idx < 0 {
+		idx = 0
+	}
+	if idx >= len(values) {
+		idx = len(values) - 1
+	}
+	return values[idx].Round(time.Millisecond)
+}
+
+func shortError(err error) string {
+	text := strings.TrimSpace(err.Error())
+	if len(text) > 180 {
+		return text[:180] + "..."
+	}
+	return text
+}
+
+func getenv(key, fallback string) string {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	return value
+}
+
+func getenvInt(key string, fallback int) int {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	n, err := strconv.Atoi(value)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "ungültiger Integer in %s: %v\n", key, err)
+		os.Exit(2)
+	}
+	return n
+}
+
+func getenvInt64(key string, fallback int64) int64 {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	n, err := strconv.ParseInt(value, 10, 64)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "ungültiger Integer in %s: %v\n", key, err)
+		os.Exit(2)
+	}
+	return n
+}
+
+func getenvDuration(key string, fallback time.Duration) time.Duration {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	d, err := time.ParseDuration(value)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "ungültige Dauer in %s: %v\n", key, err)
+		os.Exit(2)
+	}
+	return d
+}
+
+func getenvBool(key string, fallback bool) bool {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	parsed, err := strconv.ParseBool(value)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "ungültiger Boolean in %s: %v\n", key, err)
+		os.Exit(2)
+	}
+	return parsed
+}
+
+func cloneMap(in map[int]int64) map[int]int64 {
+	out := make(map[int]int64, len(in))
+	for k, v := range in {
+		out[k] = v
+	}
+	return out
+}
+
+func cloneStringMap(in map[string]int64) map[string]int64 {
+	out := make(map[string]int64, len(in))
+	for k, v := range in {
+		out[k] = v
+	}
+	return out
+}
--- a/cmd/siem-stress-agent/main_test.go
+++ b/cmd/siem-stress-agent/main_test.go
@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestSendBatch(t *testing.T) {
+	var got []event
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("X-API-Key") != "test-key" {
+			t.Fatalf("unexpected api key")
+		}
+		if err := json.NewDecoder(r.Body).Decode(&got); err != nil {
+			t.Fatalf("decode: %v", err)
+		}
+		w.WriteHeader(http.StatusAccepted)
+	}))
+	defer server.Close()
+
+	cfg := config{Endpoint: server.URL, APIKey: "test-key", Timeout: time.Second}
+	batch := []event{generateEvent(config{Hostname: "SIEM-STRESS-01", Scenario: "lockout"}, 1, time.Now().UTC())}
+	res := sendBatch(context.Background(), server.Client(), cfg, batch)
+	if res.Err != nil || res.Status != http.StatusAccepted {
+		t.Fatalf("unexpected result: %+v", res)
+	}
+	if len(got) != 1 || got[0].EventID != 4740 || got[0].Source != "SIEM-Stress-Agent" {
+		t.Fatalf("unexpected payload: %+v", got)
+	}
+}
+
+func TestValidateConfigRequiresConfirmation(t *testing.T) {
+	cfg := config{
+		Endpoint:   "http://127.0.0.1:8080/ingest",
+		Hostname:   "SIEM-STRESS-01",
+		APIKey:     "x",
+		Scenario:   "mixed",
+		Rate:       1,
+		BatchSize:  1,
+		Workers:    1,
+		Duration:   time.Second,
+		EventLimit: 1,
+		Timeout:    time.Second,
+	}
+	if err := validateConfig(cfg); err == nil {
+		t.Fatal("expected missing confirmation to fail")
+	}
+}
--- a/compose.yml
+++ b/compose.yml
@@ -69,7 +69,9 @@ services:
      NEW_EVENT_ID_IGNORE_CHANNELS: ${NEW_EVENT_ID_IGNORE_CHANNELS}
      NEW_EVENT_ID_ALERT_CHANNELS: ${NEW_EVENT_ID_ALERT_CHANNELS}
      NEW_EVENT_ID_HIGH_RISK_IDS: ${NEW_EVENT_ID_HIGH_RISK_IDS}
+      STORE_EVENT_ROWS: ${STORE_EVENT_ROWS}
      STORE_RAW_XML: ${STORE_RAW_XML}
+      METADATA_CONTEXT_EVENT_IDS: ${METADATA_CONTEXT_EVENT_IDS}
      METADATA_BUCKET: ${METADATA_BUCKET}
      EVENT_RETENTION: ${EVENT_RETENTION}
      RAW_RETENTION: ${RAW_RETENTION}
--- a/deploy/mariadb/cardinality-diagnostics.sql
+++ b/deploy/mariadb/cardinality-diagnostics.sql
@@ -0,0 +1,35 @@
+-- Shows where rows are growing and how effective aggregation currently is.
+SET time_zone = '+00:00';
+
+SELECT table_name,
+       table_rows,
+       ROUND(data_length/1024/1024,1) AS data_mb,
+       ROUND(index_length/1024/1024,1) AS index_mb,
+       ROUND((data_length+index_length)/1024/1024,1) AS total_mb
+FROM information_schema.tables
+WHERE table_schema = DATABASE()
+ORDER BY data_length + index_length DESC;
+
+SELECT 'event_count_buckets' AS source,
+       COUNT(*) AS rows_24h,
+       COALESCE(SUM(cnt),0) AS represented_events_24h,
+       ROUND(COALESCE(SUM(cnt),0) / NULLIF(COUNT(*),0), 1) AS events_per_row
+FROM event_count_buckets
+WHERE bucket_start >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
+UNION ALL
+SELECT 'event_occurrences',
+       COUNT(*),
+       COALESCE(SUM(cnt),0),
+       ROUND(COALESCE(SUM(cnt),0) / NULLIF(COUNT(*),0), 1)
+FROM event_occurrences
+WHERE bucket_start >= UTC_TIMESTAMP() - INTERVAL 24 HOUR;
+
+SELECT channel_name, event_id,
+       COUNT(*) AS context_rows_24h,
+       SUM(cnt) AS represented_events_24h,
+       ROUND(SUM(cnt)/NULLIF(COUNT(*),0),1) AS events_per_row
+FROM event_occurrences
+WHERE bucket_start >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
+GROUP BY channel_name, event_id
+ORDER BY context_rows_24h DESC
+LIMIT 50;
--- a/deploy/mariadb/diagnostics.sql
+++ b/deploy/mariadb/diagnostics.sql
@@ -0,0 +1,52 @@
+-- Read-only diagnostics for sizing and partition verification.
+SET time_zone = '+00:00';
+
+SELECT
+    TABLE_NAME,
+    ENGINE,
+    TABLE_ROWS,
+    ROUND(DATA_LENGTH / 1024 / 1024, 1) AS data_mb,
+    ROUND(INDEX_LENGTH / 1024 / 1024, 1) AS index_mb,
+    ROUND((DATA_LENGTH + INDEX_LENGTH) / 1024 / 1024, 1) AS total_mb
+FROM information_schema.TABLES
+WHERE TABLE_SCHEMA = DATABASE()
+ORDER BY (DATA_LENGTH + INDEX_LENGTH) DESC;
+
+SELECT
+    TABLE_NAME,
+    COUNT(*) AS partitions,
+    MIN(PARTITION_NAME) AS oldest_partition,
+    MAX(PARTITION_NAME) AS newest_partition,
+    ROUND(SUM(DATA_LENGTH + INDEX_LENGTH) / 1024 / 1024, 1) AS total_mb
+FROM information_schema.PARTITIONS
+WHERE TABLE_SCHEMA = DATABASE()
+  AND PARTITION_NAME IS NOT NULL
+GROUP BY TABLE_NAME
+ORDER BY total_mb DESC;
+
+SELECT
+    channel_name,
+    event_id,
+    SUM(cnt) AS events_24h,
+    COUNT(*) AS metadata_rows_24h,
+    ROUND(SUM(cnt) / NULLIF(COUNT(*), 0), 2) AS compression_factor
+FROM event_occurrences
+WHERE bucket_start >= UTC_TIMESTAMP() - INTERVAL 24 HOUR
+GROUP BY channel_name, event_id
+ORDER BY events_24h DESC
+LIMIT 50;
+
+SELECT
+    first_event_ts,
+    last_event_ts,
+    cnt,
+    hostname,
+    target_user,
+    workstation,
+    src_ip
+FROM event_occurrences
+WHERE channel_name = 'Security'
+  AND event_id = 4740
+  AND bucket_start >= UTC_TIMESTAMP() - INTERVAL 7 DAY
+ORDER BY bucket_start DESC
+LIMIT 200;
--- a/deploy/mariadb/emergency-compact-cleanup.sql
+++ b/deploy/mariadb/emergency-compact-cleanup.sql
@@ -0,0 +1,43 @@
+-- EMERGENCY CLEANUP FOR THE PREVIOUS METADATA-FIRST BUILD
+--
+-- Run ONLY AFTER the new backend is deployed with:
+--   STORE_EVENT_ROWS=false
+--   STORE_RAW_XML=false
+--
+-- This intentionally removes redundant per-event/full-context history generated
+-- by older builds. Aggregate event history remains in event_count_buckets and
+-- event_catalog. Existing detections and UEBA baselines are not deleted.
+--
+-- Take a database backup/snapshot first if old raw/event detail may still matter.
+
+SET NAMES utf8mb4;
+SET time_zone = '+00:00';
+
+SELECT 'before' AS phase,
+       table_name,
+       table_rows,
+       ROUND((data_length + index_length) / 1024 / 1024, 1) AS size_mb
+FROM information_schema.tables
+WHERE table_schema = DATABASE()
+  AND table_name IN ('event_logs','event_log_raw','event_occurrences','event_count_buckets','event_catalog','detections')
+ORDER BY size_mb DESC;
+
+-- Raw XML must be cleared before its logical parent event rows.
+TRUNCATE TABLE event_log_raw;
+TRUNCATE TABLE event_logs;
+
+-- Old versions used too many dimensions and could make this table nearly as
+-- large as event_logs. The new backend will repopulate only selected Security
+-- contexts in 5-minute buckets.
+TRUNCATE TABLE event_occurrences;
+
+ANALYZE TABLE event_count_buckets, event_catalog, detections;
+
+SELECT 'after' AS phase,
+       table_name,
+       table_rows,
+       ROUND((data_length + index_length) / 1024 / 1024, 1) AS size_mb
+FROM information_schema.tables
+WHERE table_schema = DATABASE()
+  AND table_name IN ('event_logs','event_log_raw','event_occurrences','event_count_buckets','event_catalog','detections')
+ORDER BY size_mb DESC;
--- a/deploy/mariadb/init/001-schema.sql
+++ b/deploy/mariadb/init/001-schema.sql
@@ -62,7 +62,7 @@ CREATE TABLE agents (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

 -- ---------------------------------------------------------------------
-- Event Logs, normalisierte Haupttabelle
+-- Event Logs, optionaler kurzlebiger Full-Event-Hotstore (STORE_EVENT_ROWS=true)
 -- ---------------------------------------------------------------------

 CREATE TABLE event_logs (
@@ -202,8 +202,9 @@ CREATE TABLE event_catalog (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

 -- ---------------------------------------------------------------------
-- Langfristige, kompakte Event-Metadaten
-- Eine Zeile pro Zeit-Bucket und Dimensionskombination statt einer Zeile pro XML.
+-- Kompakte Security-Kontext-Metadaten für eine kleine Event-ID-Allowlist.
+-- Alle Events werden universell in event_count_buckets gezählt; diese Tabelle
+-- speichert nur Benutzer/IP/Workstation-Kontext für ausgewählte Security-Events.
 -- ---------------------------------------------------------------------

 CREATE TABLE event_occurrences (
@@ -234,7 +235,15 @@ CREATE TABLE event_occurrences (
    KEY idx_occurrences_host_event_time (hostname, channel_name, event_id, bucket_start),
    KEY idx_occurrences_target_user_time (target_user, bucket_start),
    KEY idx_occurrences_subject_user_time (subject_user, bucket_start),
-    KEY idx_occurrences_src_ip_time (src_ip, bucket_start)
+    KEY idx_occurrences_src_ip_time (src_ip, bucket_start),
+    KEY idx_occurrences_security_event_time (
+        channel_name,
+        event_id,
+        bucket_start,
+        hostname,
+        target_user,
+        src_ip
+    )
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
 PARTITION BY RANGE COLUMNS(bucket_start) (
    PARTITION pmax VALUES LESS THAN (MAXVALUE)
@@ -419,6 +428,13 @@ CREATE TABLE event_count_buckets (
    KEY idx_event_count_buckets_time (
        bucket_start,
        bucket_end
+    ),
+
+    KEY idx_event_count_buckets_event_time (
+        channel_name,
+        event_id,
+        bucket_start,
+        hostname
    )
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
 PARTITION BY RANGE COLUMNS(bucket_start) (
--- a/deploy/mariadb/migrations/002-metadata-first.sql
+++ b/deploy/mariadb/migrations/002-metadata-first.sql
@@ -0,0 +1,98 @@
+-- Metadata-first migration for an existing SIEM-lite database.
+-- IMPORTANT: ALTER TABLE ... PARTITION BY can rebuild and lock large tables.
+-- Run during a maintenance window and take a backup first.
+
+SET NAMES utf8mb4;
+SET time_zone = '+00:00';
+
+CREATE TABLE IF NOT EXISTS event_catalog (
+    hostname VARCHAR(191) NOT NULL,
+    channel_name VARCHAR(128) NOT NULL,
+    event_id INT UNSIGNED NOT NULL,
+    first_seen DATETIME(6) NOT NULL,
+    last_seen DATETIME(6) NOT NULL,
+    total_count BIGINT UNSIGNED NOT NULL DEFAULT 0,
+    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
+    PRIMARY KEY (hostname, channel_name, event_id),
+    KEY idx_event_catalog_first_seen (first_seen),
+    KEY idx_event_catalog_last_seen (last_seen)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+-- Warm the catalog from the much smaller baseline buckets so the
+-- new_event_id rule does not treat every known event type as new after migration.
+INSERT INTO event_catalog
+(hostname, channel_name, event_id, first_seen, last_seen, total_count, updated_at)
+SELECT hostname,
+       channel_name,
+       event_id,
+       MIN(COALESCE(first_event_ts, bucket_start)),
+       MAX(COALESCE(last_event_ts, bucket_end)),
+       SUM(cnt),
+       UTC_TIMESTAMP(6)
+FROM event_count_buckets
+GROUP BY hostname, channel_name, event_id
+ON DUPLICATE KEY UPDATE
+    first_seen = LEAST(first_seen, VALUES(first_seen)),
+    last_seen = GREATEST(last_seen, VALUES(last_seen)),
+    total_count = GREATEST(total_count, VALUES(total_count)),
+    updated_at = UTC_TIMESTAMP(6);
+
+CREATE TABLE IF NOT EXISTS event_occurrences (
+    bucket_start DATETIME(6) NOT NULL,
+    bucket_end DATETIME(6) NOT NULL,
+    dimension_key BINARY(16) NOT NULL,
+    hostname VARCHAR(191) NOT NULL,
+    channel_name VARCHAR(128) NOT NULL,
+    event_id INT UNSIGNED NOT NULL,
+    provider_name VARCHAR(191) NOT NULL DEFAULT '',
+    target_user VARCHAR(191) NOT NULL DEFAULT '',
+    subject_user VARCHAR(191) NOT NULL DEFAULT '',
+    src_ip VARCHAR(64) NOT NULL DEFAULT '',
+    workstation VARCHAR(191) NOT NULL DEFAULT '',
+    logon_type VARCHAR(32) NOT NULL DEFAULT '',
+    status_text VARCHAR(128) NOT NULL DEFAULT '',
+    failure_reason VARCHAR(255) NOT NULL DEFAULT '',
+    cnt BIGINT UNSIGNED NOT NULL DEFAULT 0,
+    first_event_ts DATETIME(6) NOT NULL,
+    last_event_ts DATETIME(6) NOT NULL,
+    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) ON UPDATE CURRENT_TIMESTAMP(6),
+    PRIMARY KEY (bucket_start, dimension_key),
+    KEY idx_occurrences_time_host_event (bucket_start, hostname, channel_name, event_id),
+    KEY idx_occurrences_host_event_time (hostname, channel_name, event_id, bucket_start),
+    KEY idx_occurrences_target_user_time (target_user, bucket_start),
+    KEY idx_occurrences_subject_user_time (subject_user, bucket_start),
+    KEY idx_occurrences_src_ip_time (src_ip, bucket_start)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+PARTITION BY RANGE COLUMNS(bucket_start) (
+    PARTITION pmax VALUES LESS THAN (MAXVALUE)
+);
+
+-- Partition bucket tables only when needed, so this migration can be rerun.
+DELIMITER $$
+CREATE PROCEDURE partition_bucket_table_if_needed(IN p_table VARCHAR(64))
+BEGIN
+    DECLARE v_partition_count INT DEFAULT 0;
+
+    SELECT COUNT(*) INTO v_partition_count
+    FROM information_schema.PARTITIONS
+    WHERE TABLE_SCHEMA = DATABASE()
+      AND TABLE_NAME = p_table
+      AND PARTITION_NAME IS NOT NULL;
+
+    IF v_partition_count = 0 THEN
+        SET @partition_sql = CONCAT(
+            'ALTER TABLE `', p_table, '` ',
+            'PARTITION BY RANGE COLUMNS(bucket_start) ',
+            '(PARTITION pmax VALUES LESS THAN (MAXVALUE))'
+        );
+        PREPARE partition_stmt FROM @partition_sql;
+        EXECUTE partition_stmt;
+        DEALLOCATE PREPARE partition_stmt;
+    END IF;
+END$$
+DELIMITER ;
+
+CALL partition_bucket_table_if_needed('event_occurrences');
+CALL partition_bucket_table_if_needed('event_count_buckets');
+CALL partition_bucket_table_if_needed('ueba_context_buckets');
+DROP PROCEDURE partition_bucket_table_if_needed;
--- a/deploy/mariadb/migrations/003-new-event-id-classification.sql
+++ b/deploy/mariadb/migrations/003-new-event-id-classification.sql
@@ -0,0 +1,27 @@
+-- Metadata-first SIEM: realistischere Einordnung der Regel new_event_id
+-- Diese Migration verändert kein Schema. Sie bereinigt nur bereits offene
+-- WMI-Operational-Inventarmeldungen, die von der alten Regel als Medium-Alarm
+-- angelegt wurden. Vor produktiver Ausführung wie üblich ein Backup erstellen.
+
+START TRANSACTION;
+
+UPDATE detections
+SET severity = 'info',
+    status = 'legitimate',
+    is_legitimate = 1,
+    reviewed_by = 'system:migration-003',
+    reviewed_at = UTC_TIMESTAMP(6),
+    analyst_note = CASE
+        WHEN analyst_note IS NULL OR TRIM(analyst_note) = '' THEN
+            'Automatisch neu klassifiziert: neue Event-IDs im WMI-Activity/Operational-Channel sind Inventar-/Diagnoseinformationen und allein kein Incident.'
+        ELSE CONCAT(
+            analyst_note,
+            '\nAutomatisch neu klassifiziert: neue Event-IDs im WMI-Activity/Operational-Channel sind Inventar-/Diagnoseinformationen und allein kein Incident.'
+        )
+    END
+WHERE rule_name = 'new_event_id'
+  AND channel_name = 'Microsoft-Windows-WMI-Activity/Operational'
+  AND status IN ('open', 'acknowledged', 'investigating', 'plausible')
+  AND is_legitimate = 0;
+
+COMMIT;
--- a/deploy/mariadb/migrations/004-compact-storage.sql
+++ b/deploy/mariadb/migrations/004-compact-storage.sql
@@ -0,0 +1,21 @@
+-- Compact-storage migration.
+-- Safe to apply without deleting event history. The destructive cleanup of old
+-- per-event rows is intentionally kept in ../emergency-compact-cleanup.sql.
+
+SET NAMES utf8mb4;
+SET time_zone = '+00:00';
+
+CREATE INDEX IF NOT EXISTS idx_event_count_buckets_event_time
+    ON event_count_buckets (channel_name, event_id, bucket_start, hostname);
+
+CREATE INDEX IF NOT EXISTS idx_occurrences_security_event_time
+    ON event_occurrences (
+        channel_name,
+        event_id,
+        bucket_start,
+        hostname,
+        target_user,
+        src_ip
+    );
+
+ANALYZE TABLE event_count_buckets, event_occurrences, event_catalog, detections;
--- a/10
+++ b/10
@@ -53,11 +53,13 @@ BASELINE_SUPPRESS_FOR=1h
 #BASELINE_MIN_COUNT=20
 # Metadata-first storage
 # Raw XML is disabled by default. Enable only for short forensic retention.
+STORE_EVENT_ROWS=false
 STORE_RAW_XML=false
-METADATA_BUCKET=1m
-EVENT_RETENTION=72h
-RAW_RETENTION=24h
-METADATA_RETENTION=4320h
+METADATA_CONTEXT_EVENT_IDS=4624,4625,4648,4672,4720,4726,4728,4732,4740,4756,4768,4769,4771,4776
+METADATA_BUCKET=5m
+EVENT_RETENTION=6h
+RAW_RETENTION=6h
+METADATA_RETENTION=720h

 # Partition maintenance
 PARTITION_MAINTENANCE_ENABLED=true
--- a/main.go
+++ b/main.go
@@ -1186,11 +1186,13 @@ type Config struct {

 	// Metadata-first storage: raw XML is optional, compact event rows are short-lived,
 	// and aggregated occurrences remain queryable for a much longer period.
-	StoreRawXML       bool
-	MetadataBucket    time.Duration
-	EventRetention    time.Duration
-	RawRetention      time.Duration
-	MetadataRetention time.Duration
+	StoreEventRows          bool
+	StoreRawXML             bool
+	MetadataBucket          time.Duration
+	EventRetention          time.Duration
+	RawRetention            time.Duration
+	MetadataRetention       time.Duration
+	MetadataContextEventIDs map[uint32]struct{}
 }

 type LogPayload struct {
@@ -1610,6 +1612,10 @@ var (
 func main() {
 	cfg := loadConfig()
 	logger := log.New(os.Stdout, "", log.LstdFlags|log.Lmicroseconds|log.LUTC)
+	logger.Printf("storage mode: store_event_rows=%t store_raw_xml=%t metadata_bucket=%s context_event_ids=%d", cfg.StoreEventRows, cfg.StoreRawXML, cfg.MetadataBucket, len(cfg.MetadataContextEventIDs))
+	if cfg.StoreRawXML && !cfg.StoreEventRows {
+		logger.Printf("storage warning: STORE_RAW_XML=true implicitly requires per-event rows; high-cardinality hot storage is active")
+	}
 	loc, err := time.LoadLocation(cfg.Timezone)
 	if err != nil {
 		logger.Printf("invalid TZ %q, falling back to Local: %v", cfg.Timezone, err)
@@ -1992,6 +1998,8 @@ ON DUPLICATE KEY UPDATE
    first_event_ts = LEAST(first_event_ts, VALUES(first_event_ts)),
    last_event_ts = GREATEST(last_event_ts, VALUES(last_event_ts)),
    bucket_end = GREATEST(bucket_end, VALUES(bucket_end)),
+    provider_name = VALUES(provider_name),
+    failure_reason = VALUES(failure_reason),
    updated_at = UTC_TIMESTAMP(6)
 `)

@@ -3767,7 +3775,7 @@ func (s *server) getDashboardStats(ctx context.Context) (DashboardStats, error)
 	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM agents WHERE is_enabled = 1 AND last_seen >= ?`, time.Now().UTC().Add(-s.cfg.OfflineAfter)).Scan(&stats.AgentsActive); err != nil {
 		return stats, err
 	}
-	if err := s.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(cnt), 0) FROM event_occurrences WHERE bucket_start >= ?`, time.Now().UTC().Add(-24*time.Hour)).Scan(&stats.Events24h); err != nil {
+	if err := s.db.QueryRowContext(ctx, `SELECT COALESCE(SUM(cnt), 0) FROM event_count_buckets WHERE bucket_start >= ?`, time.Now().UTC().Add(-24*time.Hour)).Scan(&stats.Events24h); err != nil {
 		return stats, err
 	}
 	if err := s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM detections WHERE created_at >= ?`, time.Now().UTC().Add(-24*time.Hour)).Scan(&stats.Detections24h); err != nil {
@@ -3818,9 +3826,15 @@ func (s *server) listEvents(ctx context.Context, f EventFilter) ([]EventRow, err
 		f.Limit = 100
 	}

-	// Deliberately return bucket rows directly. This keeps ORDER BY + LIMIT on the
-	// partition key and avoids a large GROUP BY across the selected time range.
-	query := `
+	// User/IP filters need the deliberately small security-context table. Every
+	// other view reads the universal low-cardinality counter table. In particular
+	// the /ui start page never touches event_occurrences anymore.
+	useContext := f.User != "" || f.SrcIP != ""
+	args := make([]any, 0, 16)
+	var query string
+
+	if useContext {
+		query = `
 SELECT hostname, channel_name, event_id, provider_name,
       target_user, subject_user, src_ip, workstation,
       logon_type, status_text, failure_reason,
@@ -3828,7 +3842,17 @@ SELECT hostname, channel_name, event_id, provider_name,
 FROM event_occurrences
 WHERE 1=1
 `
-	args := make([]any, 0, 16)
+	} else {
+		query = `
+SELECT hostname, channel_name, event_id, '',
+       '', '', '', '',
+       '', '', '',
+       cnt, first_event_ts, last_event_ts
+FROM event_count_buckets
+WHERE 1=1
+`
+	}
+
 	if f.Host != "" {
 		query += ` AND hostname = ?`
 		args = append(args, f.Host)
@@ -3841,20 +3865,34 @@ WHERE 1=1
 		query += ` AND event_id = ?`
 		args = append(args, f.EventID)
 	}
-	if f.User != "" {
+	if useContext && f.User != "" {
 		query += ` AND (target_user = ? OR subject_user = ?)`
 		u := normalizeUsername(f.User)
 		args = append(args, u, u)
 	}
-	if f.SrcIP != "" {
+	if useContext && f.SrcIP != "" {
 		query += ` AND src_ip = ?`
 		args = append(args, f.SrcIP)
 	}
+
+	// Never let an unfiltered UI request range over the whole retention period.
+	// A user can still explicitly request an older range with from/to.
 	if f.TimeFrom != "" {
 		if t, err := parseUIRFC3339(f.TimeFrom); err == nil {
 			query += ` AND bucket_start >= ?`
-			args = append(args, bucketStart(t, s.cfg.MetadataBucket))
+			bucketSize := s.cfg.BaselineWindow
+			if useContext {
+				bucketSize = s.cfg.MetadataBucket
+			}
+			args = append(args, bucketStart(t, bucketSize))
 		}
+	} else {
+		query += ` AND bucket_start >= ?`
+		bucketSize := s.cfg.BaselineWindow
+		if useContext {
+			bucketSize = s.cfg.MetadataBucket
+		}
+		args = append(args, bucketStart(time.Now().UTC().Add(-24*time.Hour), bucketSize))
 	}
 	if f.TimeTo != "" {
 		if t, err := parseUIRFC3339(f.TimeTo); err == nil {
@@ -3863,11 +3901,15 @@ WHERE 1=1
 		}
 	}
 	if f.Rule != "" || f.Severity != "" {
+		tableName := "event_count_buckets"
+		if useContext {
+			tableName = "event_occurrences"
+		}
 		query += ` AND EXISTS (
 		SELECT 1 FROM detections d
-		WHERE d.hostname = event_occurrences.hostname
-		  AND event_occurrences.last_event_ts >= d.window_start
-		  AND event_occurrences.first_event_ts <= d.window_end`
+		WHERE d.hostname = ` + tableName + `.hostname
+		  AND ` + tableName + `.last_event_ts >= d.window_start
+		  AND ` + tableName + `.first_event_ts <= d.window_end`
 		if f.Rule != "" {
 			query += ` AND d.rule_name = ?`
 			args = append(args, f.Rule)
@@ -4008,11 +4050,13 @@ func loadConfig() Config {
 		PartitionBehind:              getenvDuration("PARTITION_BEHIND", 6*time.Hour),
 		PartitionRetention:           getenvDuration("PARTITION_RETENTION", 30*24*time.Hour),

-		StoreRawXML:       getenvBool("STORE_RAW_XML", false),
-		MetadataBucket:    getenvDuration("METADATA_BUCKET", time.Minute),
-		EventRetention:    getenvDuration("EVENT_RETENTION", 72*time.Hour),
-		RawRetention:      getenvDuration("RAW_RETENTION", 24*time.Hour),
-		MetadataRetention: getenvDuration("METADATA_RETENTION", 180*24*time.Hour),
+		StoreEventRows:          getenvBool("STORE_EVENT_ROWS", false),
+		StoreRawXML:             getenvBool("STORE_RAW_XML", false),
+		MetadataBucket:          getenvDuration("METADATA_BUCKET", 5*time.Minute),
+		EventRetention:          getenvDuration("EVENT_RETENTION", 6*time.Hour),
+		RawRetention:            getenvDuration("RAW_RETENTION", 6*time.Hour),
+		MetadataRetention:       getenvDuration("METADATA_RETENTION", 30*24*time.Hour),
+		MetadataContextEventIDs: getenvUint32Set("METADATA_CONTEXT_EVENT_IDS", "4624,4625,4648,4672,4688,4697,4720,4726,4728,4732,4740,4756,4768,4769,4771,4776,7045"),
 	}
 }

@@ -4329,6 +4373,62 @@ VALUES (?, ?, UTC_TIMESTAMP(6), UTC_TIMESTAMP(6), ?, 1)

 var errUnauthorized = errors.New("unauthorized")

+func shouldStoreMetadataContext(cfg Config, channel string, eventID uint32) bool {
+	if len(cfg.MetadataContextEventIDs) == 0 {
+		return false
+	}
+	_, ok := cfg.MetadataContextEventIDs[eventID]
+	if !ok {
+		return false
+	}
+	// Context extraction is currently designed around Windows Security semantics.
+	// Event 7045 and other non-Security IDs can still be counted universally via
+	// event_count_buckets without creating a high-cardinality context row.
+	return strings.EqualFold(strings.TrimSpace(channel), "Security")
+}
+
+func compactMetadataContext(eventID uint32, norm NormalizedEvent) (targetUser, subjectUser, srcIP, workstation, logonType, statusText, failureReason string) {
+	targetUser = normalizeUsername(norm.TargetUser)
+	subjectUser = normalizeUsername(norm.SubjectUser)
+	srcIP = strings.TrimSpace(norm.SrcIP)
+	workstation = strings.TrimSpace(norm.Workstation)
+	logonType = strings.TrimSpace(norm.LogonType)
+	statusText = strings.TrimSpace(norm.StatusText)
+	failureReason = strings.TrimSpace(norm.FailureReason)
+
+	// Keep only dimensions that are actually useful for the event family. This is
+	// the key protection against event_occurrences degenerating into one row/event.
+	switch eventID {
+	case 4624, 4625: // successful / failed logon
+		subjectUser = ""
+	case 4672: // special privileges assigned
+		targetUser, srcIP, workstation, logonType, statusText, failureReason = "", "", "", "", "", ""
+	case 4740: // account locked out
+		subjectUser, srcIP, logonType, statusText, failureReason = "", "", "", "", ""
+	case 4768, 4769, 4771, 4776: // Kerberos / credential validation
+		subjectUser = ""
+	default:
+		// Account/group-management events benefit mostly from actor + target.
+		srcIP, logonType, statusText, failureReason = "", "", "", ""
+	}
+
+	if len(failureReason) > 255 {
+		failureReason = failureReason[:255]
+	}
+	return
+}
+
+func retainCompactContext(eventID uint32, targetUser, subjectUser string) bool {
+	switch eventID {
+	case 4624, 4625, 4740, 4768, 4769, 4771, 4776:
+		return !isNoiseAccount(targetUser)
+	case 4672:
+		return !isNoiseAccount(subjectUser)
+	default:
+		return !isNoiseAccount(targetUser) || !isNoiseAccount(subjectUser)
+	}
+}
+
 func (s *server) insertBatch(ctx context.Context, agentID uint64, batch []LogPayload) error {
 	start := time.Now()

@@ -4338,10 +4438,13 @@ func (s *server) insertBatch(ctx context.Context, agentID uint64, batch []LogPay
 	}
 	defer func() { _ = tx.Rollback() }()

+	// Full per-event rows are now strictly opt-in. STORE_RAW_XML implies an event
+	// row because event_log_raw references event_log_id.
+	storeEventRows := s.cfg.StoreEventRows || s.cfg.StoreRawXML
 	var sb strings.Builder
-	args := make([]any, 0, len(batch)*29)
-
-	sb.WriteString(`
+	args := make([]any, 0, len(batch)*30)
+	if storeEventRows {
+		sb.WriteString(`
 INSERT INTO event_logs (
 	agent_id, hostname, channel_name, event_id, source, computer, provider_name,
 	level_value, task_value, opcode_value, keywords,
@@ -4352,27 +4455,17 @@ INSERT INTO event_logs (
 	failure_reason, ts, received_at, msg, msg_sha256
 ) VALUES
 `)
+	}

 	bucketAgg := make(map[string]*EventCountBucketAgg)
 	occurrenceAgg := make(map[string]*MetadataOccurrenceAgg)
 	rawEvents := make([]RawEventInsert, 0, len(batch))
+	privilegedLogonMetric := make(map[string]int)
+	privilegedFailureMetric := make(map[string]int)
+	privilegedGrantMetric := make(map[string]int)

 	for i, item := range batch {
-		if i > 0 {
-			sb.WriteString(",")
-		}
-		sb.WriteString("(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,UTC_TIMESTAMP(6),'',?)")
-
 		msgHash := payloadFingerprint(item)
-		if s.cfg.StoreRawXML && strings.TrimSpace(item.Message) != "" {
-			rawEvents = append(rawEvents, RawEventInsert{
-				EventOffset: i,
-				Message:     item.Message,
-				SHA256:      msgHash,
-				Time:        item.Time.UTC(),
-			})
-		}
-
 		norm := NormalizeEventXML(item.Message)
 		if item.Metadata != nil {
 			norm = overlayMetadata(norm, *item.Metadata)
@@ -4383,6 +4476,33 @@ INSERT INTO event_logs (
 		providerName := firstNonEmpty(norm.ProviderName, item.Source)
 		eventTS := item.Time.UTC()

+		if storeEventRows {
+			if i > 0 {
+				sb.WriteString(",")
+			}
+			sb.WriteString("(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,UTC_TIMESTAMP(6),'',?)")
+			args = append(args,
+				agentID, realHost, item.Channel, item.EventID, item.Source, norm.Computer, providerName,
+				norm.LevelValue, norm.TaskValue, norm.OpcodeValue, norm.Keywords,
+				norm.TargetUser, targetUserNorm, norm.TargetDomain,
+				norm.SubjectUser, subjectUserNorm, norm.SubjectDomain,
+				norm.Workstation, norm.SrcIP, norm.SrcPort, norm.LogonType, norm.ProcessName,
+				norm.AuthenticationPackage, norm.LogonProcess, norm.StatusText, norm.SubStatusText,
+				norm.FailureReason, eventTS, msgHash,
+			)
+		}
+
+		if s.cfg.StoreRawXML && strings.TrimSpace(item.Message) != "" {
+			rawEvents = append(rawEvents, RawEventInsert{
+				EventOffset: i,
+				Message:     item.Message,
+				SHA256:      msgHash,
+				Time:        eventTS,
+			})
+		}
+
+		// Universal low-cardinality counter: one row per host/channel/event ID and
+		// baseline bucket, regardless of how many individual events arrive.
 		baselineBucketSize := s.cfg.BaselineWindow
 		if baselineBucketSize <= 0 {
 			baselineBucketSize = 5 * time.Minute
@@ -4406,93 +4526,77 @@ INSERT INTO event_logs (
 			agg.LastTS = eventTS
 		}

-		metadataBucketSize := s.cfg.MetadataBucket
-		if metadataBucketSize <= 0 {
-			metadataBucketSize = time.Minute
-		}
-		mbs := bucketStart(eventTS, metadataBucketSize)
-		mbe := mbs.Add(metadataBucketSize)
-		failureReason := norm.FailureReason
-		if len(failureReason) > 255 {
-			failureReason = failureReason[:255]
-		}
-		dimKey := metadataDimensionKey(
-			realHost, item.Channel, strconv.FormatUint(uint64(item.EventID), 10), providerName,
-			targetUserNorm, subjectUserNorm, norm.SrcIP, norm.Workstation,
-			norm.LogonType, norm.StatusText, failureReason,
-		)
-		occKey := mbs.Format(time.RFC3339Nano) + "|" + hex.EncodeToString(dimKey)
-		occ := occurrenceAgg[occKey]
-		if occ == nil {
-			occ = &MetadataOccurrenceAgg{
-				BucketStart: mbs, BucketEnd: mbe, DimensionKey: dimKey,
-				Hostname: realHost, Channel: item.Channel, EventID: item.EventID,
-				ProviderName: providerName, TargetUser: targetUserNorm, SubjectUser: subjectUserNorm,
-				SrcIP: norm.SrcIP, Workstation: norm.Workstation, LogonType: norm.LogonType,
-				StatusText: norm.StatusText, FailureReason: failureReason,
-				FirstTS: eventTS, LastTS: eventTS,
+		// Context rows are intentionally restricted to a small allow-list of
+		// security events. All other events exist only in event_count_buckets and
+		// event_catalog, preventing metadata storage from mirroring raw cardinality.
+		if shouldStoreMetadataContext(s.cfg, item.Channel, item.EventID) {
+			metadataBucketSize := s.cfg.MetadataBucket
+			if metadataBucketSize <= 0 {
+				metadataBucketSize = 5 * time.Minute
+			}
+			mbs := bucketStart(eventTS, metadataBucketSize)
+			mbe := mbs.Add(metadataBucketSize)
+			target, subject, srcIP, workstation, logonType, statusText, failureReason := compactMetadataContext(item.EventID, norm)
+			if retainCompactContext(item.EventID, target, subject) {
+				dimKey := metadataDimensionKey(
+					realHost, item.Channel, strconv.FormatUint(uint64(item.EventID), 10),
+					target, subject, srcIP, workstation, logonType, statusText,
+				)
+				occKey := mbs.Format(time.RFC3339Nano) + "|" + hex.EncodeToString(dimKey)
+				occ := occurrenceAgg[occKey]
+				if occ == nil {
+					occ = &MetadataOccurrenceAgg{
+						BucketStart: mbs, BucketEnd: mbe, DimensionKey: dimKey,
+						Hostname: realHost, Channel: item.Channel, EventID: item.EventID,
+						ProviderName: providerName, TargetUser: target, SubjectUser: subject,
+						SrcIP: srcIP, Workstation: workstation, LogonType: logonType,
+						StatusText: statusText, FailureReason: failureReason,
+						FirstTS: eventTS, LastTS: eventTS,
+					}
+					occurrenceAgg[occKey] = occ
+				}
+				occ.Count++
+				if eventTS.Before(occ.FirstTS) {
+					occ.FirstTS = eventTS
+				}
+				if eventTS.After(occ.LastTS) {
+					occ.LastTS = eventTS
+					// Preserve the most recent descriptive values without making them
+					// part of the uniqueness key.
+					occ.ProviderName = providerName
+					occ.FailureReason = failureReason
+				}
 			}
-			occurrenceAgg[occKey] = occ
-		}
-		occ.Count++
-		if eventTS.Before(occ.FirstTS) {
-			occ.FirstTS = eventTS
-		}
-		if eventTS.After(occ.LastTS) {
-			occ.LastTS = eventTS
 		}

-		// Privileged-user metrics are based only on compact normalized metadata.
 		if item.Channel == "Security" && item.EventID == 4624 && targetUserNorm != "" {
-			priv, err := s.detector.isPrivilegedUser(ctx, targetUserNorm)
-			if err != nil {
-				s.logger.Printf("privileged user check failed for %s: %v", targetUserNorm, err)
-			} else if priv {
-				s.detector.privilegedLogonsTotal.WithLabelValues(targetUserNorm, realHost).Inc()
-			}
+			privilegedLogonMetric[targetUserNorm+"\x00"+realHost]++
 		}
 		if item.Channel == "Security" && item.EventID == 4625 && targetUserNorm != "" {
-			priv, err := s.detector.isPrivilegedUser(ctx, targetUserNorm)
-			if err != nil {
-				s.logger.Printf("privileged user check failed for %s: %v", targetUserNorm, err)
-			} else if priv {
-				s.detector.privilegedLogonFailuresTotal.WithLabelValues(targetUserNorm, realHost).Inc()
-			}
+			privilegedFailureMetric[targetUserNorm+"\x00"+realHost]++
 		}
 		if item.Channel == "Security" && item.EventID == 4672 && subjectUserNorm != "" {
-			priv, err := s.detector.isPrivilegedUser(ctx, subjectUserNorm)
-			if err != nil {
-				s.logger.Printf("privileged user check failed for %s: %v", subjectUserNorm, err)
-			} else if priv {
-				s.detector.privilegedLogonsTotal.WithLabelValues(subjectUserNorm, realHost).Inc()
-			}
+			privilegedGrantMetric[subjectUserNorm+"\x00"+realHost]++
 		}
-
-		args = append(args,
-			agentID, realHost, item.Channel, item.EventID, item.Source, norm.Computer, providerName,
-			norm.LevelValue, norm.TaskValue, norm.OpcodeValue, norm.Keywords,
-			norm.TargetUser, targetUserNorm, norm.TargetDomain,
-			norm.SubjectUser, subjectUserNorm, norm.SubjectDomain,
-			norm.Workstation, norm.SrcIP, norm.SrcPort, norm.LogonType, norm.ProcessName,
-			norm.AuthenticationPackage, norm.LogonProcess, norm.StatusText, norm.SubStatusText,
-			norm.FailureReason, eventTS, msgHash,
-		)
 	}

-	res, err := tx.ExecContext(ctx, sb.String(), args...)
-	if err != nil {
-		return err
-	}
-	firstID, err := res.LastInsertId()
-	if err != nil {
-		return fmt.Errorf("event_logs LastInsertId: %w", err)
-	}
-	affected, err := res.RowsAffected()
-	if err != nil {
-		return fmt.Errorf("event_logs RowsAffected: %w", err)
-	}
-	if affected != int64(len(batch)) {
-		return fmt.Errorf("event_logs insert affected %d rows, expected %d", affected, len(batch))
+	var firstID int64
+	if storeEventRows {
+		res, err := tx.ExecContext(ctx, sb.String(), args...)
+		if err != nil {
+			return err
+		}
+		firstID, err = res.LastInsertId()
+		if err != nil {
+			return fmt.Errorf("event_logs LastInsertId: %w", err)
+		}
+		affected, err := res.RowsAffected()
+		if err != nil {
+			return fmt.Errorf("event_logs RowsAffected: %w", err)
+		}
+		if affected != int64(len(batch)) {
+			return fmt.Errorf("event_logs insert affected %d rows, expected %d", affected, len(batch))
+		}
 	}

 	if s.cfg.StoreRawXML {
@@ -4513,6 +4617,35 @@ INSERT INTO event_logs (
 		return err
 	}

+	// Privileged-user metrics are intentionally evaluated after the transaction and
+	// only once per unique user. Older builds performed a DB lookup per event.
+	privilegedCache := make(map[string]bool)
+	checkPrivileged := func(user string) bool {
+		if v, ok := privilegedCache[user]; ok {
+			return v
+		}
+		v, err := s.detector.isPrivilegedUser(ctx, user)
+		if err != nil {
+			s.logger.Printf("privileged user check failed for %s: %v", user, err)
+			privilegedCache[user] = false
+			return false
+		}
+		privilegedCache[user] = v
+		return v
+	}
+	emitMetric := func(values map[string]int, counter *prometheus.CounterVec) {
+		for key, count := range values {
+			parts := strings.SplitN(key, "\x00", 2)
+			if len(parts) != 2 || !checkPrivileged(parts[0]) {
+				continue
+			}
+			counter.WithLabelValues(parts[0], parts[1]).Add(float64(count))
+		}
+	}
+	emitMetric(privilegedLogonMetric, s.detector.privilegedLogonsTotal)
+	emitMetric(privilegedGrantMetric, s.detector.privilegedLogonsTotal)
+	emitMetric(privilegedFailureMetric, s.detector.privilegedLogonFailuresTotal)
+
 	dbBatchSizeHist.Observe(float64(len(batch)))
 	dbInsertEventsTotal.Add(float64(len(batch)))
 	dbTxDurationHist.Observe(time.Since(start).Seconds())
@@ -5123,7 +5256,212 @@ func (d *detector) evaluateDynamicRule(ctx context.Context, r DynamicRule) error
 	return d.evaluateThresholdRule(ctx, r, channels, eventIDs)
 }

+func dynamicRuleNeedsContext(r DynamicRule) bool {
+	switch strings.ToLower(strings.TrimSpace(r.MatchField)) {
+	case "target_user", "subject_user", "src_ip", "workstation":
+		return true
+	default:
+		return false
+	}
+}
+
+func dynamicRuleNeedsFullEvent(r DynamicRule) bool {
+	switch strings.ToLower(strings.TrimSpace(r.MatchField)) {
+	case "process_name", "msg":
+		return true
+	default:
+		return false
+	}
+}
+
+func metadataRuleMatchCondition(r DynamicRule, context bool) (string, []any) {
+	if r.MatchField == "" || r.MatchOperator == "" || r.MatchValue == "" {
+		return "", nil
+	}
+	fieldMap := map[string]string{
+		"hostname": "hostname",
+		"channel":  "channel_name",
+		"event_id": "event_id",
+	}
+	if context {
+		fieldMap["target_user"] = "target_user"
+		fieldMap["subject_user"] = "subject_user"
+		fieldMap["src_ip"] = "src_ip"
+		fieldMap["workstation"] = "workstation"
+	}
+	col, ok := fieldMap[strings.ToLower(strings.TrimSpace(r.MatchField))]
+	if !ok {
+		return "", nil
+	}
+	val := strings.TrimSpace(r.MatchValue)
+	switch strings.ToLower(strings.TrimSpace(r.MatchOperator)) {
+	case "eq":
+		return col + " = ?", []any{val}
+	case "contains":
+		return col + " LIKE ?", []any{"%" + val + "%"}
+	case "in":
+		values := parseCSVStrings(val)
+		if len(values) == 0 {
+			return "", nil
+		}
+		return buildInClause(strings.TrimSpace(col), len(values))[5:], stringSliceAny(values)
+	default:
+		return "", nil
+	}
+}
+
+func stringSliceAny(in []string) []any {
+	out := make([]any, 0, len(in))
+	for _, v := range in {
+		out = append(out, v)
+	}
+	return out
+}
+
 func (d *detector) evaluateSingleEventRule(ctx context.Context, r DynamicRule, channels []string, eventIDs []uint32) error {
+	if dynamicRuleNeedsFullEvent(r) {
+		if !d.cfg.StoreEventRows {
+			d.logger.Printf("dynamic rule %s skipped: match_field=%s requires STORE_EVENT_ROWS=true", r.Name, r.MatchField)
+			return nil
+		}
+		return d.evaluateSingleEventRuleLegacy(ctx, r, channels, eventIDs)
+	}
+
+	windowEnd := time.Now().UTC()
+	windowStart := windowEnd.Add(-d.cfg.DetectionInterval)
+	useContext := dynamicRuleNeedsContext(r)
+	table := "event_count_buckets"
+	selectFields := "hostname, channel_name, event_id, '', '', '', '', SUM(cnt), MIN(first_event_ts), MAX(last_event_ts)"
+	groupFields := "hostname, channel_name, event_id"
+	bucketSize := d.cfg.BaselineWindow
+	if useContext {
+		table = "event_occurrences"
+		selectFields = "hostname, channel_name, event_id, target_user, subject_user, src_ip, workstation, SUM(cnt), MIN(first_event_ts), MAX(last_event_ts)"
+		groupFields = "hostname, channel_name, event_id, target_user, subject_user, src_ip, workstation"
+		bucketSize = d.cfg.MetadataBucket
+	}
+
+	query := "SELECT " + selectFields + " FROM " + table + " WHERE bucket_start >= ? AND bucket_start < ?"
+	args := []any{bucketStart(windowStart, bucketSize), windowEnd}
+	query += buildInClause("channel_name", len(channels))
+	for _, ch := range channels {
+		args = append(args, ch)
+	}
+	query += buildInClause("event_id", len(eventIDs))
+	for _, id := range eventIDs {
+		args = append(args, id)
+	}
+	if cond, condArgs := metadataRuleMatchCondition(r, useContext); cond != "" {
+		query += " AND " + cond
+		args = append(args, condArgs...)
+	}
+	query += " GROUP BY " + groupFields + " ORDER BY MAX(last_event_ts) DESC LIMIT 500"
+
+	rows, err := d.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var host, channel, target, subject, srcIP, workstation string
+		var eventID uint32
+		var count int
+		var firstSeen, lastSeen time.Time
+		if err := rows.Scan(&host, &channel, &eventID, &target, &subject, &srcIP, &workstation, &count, &firstSeen, &lastSeen); err != nil {
+			return err
+		}
+		if suppressed, err := d.isDynamicRuleSuppressed(ctx, r, host, lastSeen); err != nil {
+			return err
+		} else if suppressed {
+			continue
+		}
+		created, err := d.insertDetection(ctx, Detection{
+			RuleName: "dynamic_" + r.Name, Severity: r.Severity, Hostname: host, Channel: channel, EventID: eventID,
+			Score: float64(maxInt(count, 1)), WindowStart: firstSeen, WindowEnd: lastSeen,
+			Summary: fmt.Sprintf("Dynamic Rule %s auf %s ausgelöst: EventID %d (%d Vorkommen)", r.Name, host, eventID, count),
+			Details: mustJSON(map[string]any{"rule_id": r.ID, "rule_name": r.Name, "description": r.Description, "count": count, "target_user": target, "subject_user": subject, "src_ip": srcIP, "workstation": workstation, "storage": table}),
+		})
+		if err != nil {
+			return err
+		}
+		if created {
+			d.detectionHitsTotal.WithLabelValues("dynamic_"+r.Name, r.Severity).Inc()
+		}
+	}
+	return rows.Err()
+}
+
+func (d *detector) evaluateThresholdRule(ctx context.Context, r DynamicRule, channels []string, eventIDs []uint32) error {
+	if dynamicRuleNeedsFullEvent(r) {
+		if !d.cfg.StoreEventRows {
+			d.logger.Printf("dynamic rule %s skipped: match_field=%s requires STORE_EVENT_ROWS=true", r.Name, r.MatchField)
+			return nil
+		}
+		return d.evaluateThresholdRuleLegacy(ctx, r, channels, eventIDs)
+	}
+
+	windowEnd := time.Now().UTC()
+	windowStart := windowEnd.Add(time.Duration(-r.ThresholdWindowSeconds) * time.Second)
+	useContext := dynamicRuleNeedsContext(r)
+	table := "event_count_buckets"
+	bucketSize := d.cfg.BaselineWindow
+	if useContext {
+		table = "event_occurrences"
+		bucketSize = d.cfg.MetadataBucket
+	}
+	query := "SELECT hostname, SUM(cnt) AS cnt, MIN(first_event_ts), MAX(last_event_ts) FROM " + table + " WHERE bucket_start >= ? AND bucket_start < ?"
+	args := []any{bucketStart(windowStart, bucketSize), windowEnd}
+	query += buildInClause("channel_name", len(channels))
+	for _, ch := range channels {
+		args = append(args, ch)
+	}
+	query += buildInClause("event_id", len(eventIDs))
+	for _, id := range eventIDs {
+		args = append(args, id)
+	}
+	if cond, condArgs := metadataRuleMatchCondition(r, useContext); cond != "" {
+		query += " AND " + cond
+		args = append(args, condArgs...)
+	}
+	query += " GROUP BY hostname HAVING SUM(cnt) >= ?"
+	args = append(args, r.ThresholdCount)
+
+	rows, err := d.db.QueryContext(ctx, query, args...)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var host string
+		var count int
+		var firstSeen, lastSeen time.Time
+		if err := rows.Scan(&host, &count, &firstSeen, &lastSeen); err != nil {
+			return err
+		}
+		if suppressed, err := d.isDynamicRuleSuppressed(ctx, r, host, windowEnd); err != nil {
+			return err
+		} else if suppressed {
+			continue
+		}
+		score := float64(count) / float64(r.ThresholdCount)
+		created, err := d.insertDetection(ctx, Detection{
+			RuleName: "dynamic_" + r.Name, Severity: r.Severity, Hostname: host, Score: score,
+			WindowStart: windowStart, WindowEnd: windowEnd,
+			Summary: fmt.Sprintf("Dynamic Rule %s auf %s: %d Events in %d Sekunden", r.Name, host, count, r.ThresholdWindowSeconds),
+			Details: mustJSON(map[string]any{"rule_id": r.ID, "rule_name": r.Name, "description": r.Description, "count": count, "threshold_count": r.ThresholdCount, "threshold_window_sec": r.ThresholdWindowSeconds, "first_seen": firstSeen.UTC().Format(time.RFC3339Nano), "last_seen": lastSeen.UTC().Format(time.RFC3339Nano), "event_ids": r.EventIDs, "channels": r.Channel, "match_field": r.MatchField, "match_operator": r.MatchOperator, "match_value": r.MatchValue, "storage": table}),
+		})
+		if err != nil {
+			return err
+		}
+		if created {
+			d.detectionHitsTotal.WithLabelValues("dynamic_"+r.Name, r.Severity).Inc()
+			d.anomalyScoreGauge.WithLabelValues(host, "dynamic_"+r.Name).Set(score)
+		}
+	}
+	return rows.Err()
+}
+
+func (d *detector) evaluateSingleEventRuleLegacy(ctx context.Context, r DynamicRule, channels []string, eventIDs []uint32) error {
 	windowEnd := time.Now().UTC()
 	windowStart := windowEnd.Add(-d.cfg.DetectionInterval)

@@ -5240,7 +5578,7 @@ WHERE ts >= ? AND ts < ?
 	return rows.Err()
 }

-func (d *detector) evaluateThresholdRule(ctx context.Context, r DynamicRule, channels []string, eventIDs []uint32) error {
+func (d *detector) evaluateThresholdRuleLegacy(ctx context.Context, r DynamicRule, channels []string, eventIDs []uint32) error {
 	windowEnd := time.Now().UTC()
 	windowStart := windowEnd.Add(time.Duration(-r.ThresholdWindowSeconds) * time.Second)

@@ -5570,15 +5908,15 @@ func (d *detector) runUEBABaselineUpdate(ctx context.Context) error {
 	windowStart := windowEnd.Add(-d.cfg.UEBANewContextWindow)

 	rows, err := d.db.QueryContext(ctx, `
-SELECT target_user_norm, hostname, src_ip, workstation, COUNT(*)
-FROM event_logs
+SELECT target_user, hostname, src_ip, workstation, SUM(cnt)
+FROM event_occurrences
 WHERE channel_name = 'Security'
  AND event_id = 4624
-  AND ts >= ? AND ts < ?
-  AND target_user_norm <> ''
-  AND target_user_norm NOT LIKE '%$'
-GROUP BY target_user_norm, hostname, src_ip, workstation
-`, windowStart, windowEnd)
+  AND bucket_start >= ? AND bucket_start < ?
+  AND target_user <> ''
+  AND target_user NOT LIKE '%$'
+GROUP BY target_user, hostname, src_ip, workstation
+`, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd)
 	if err != nil {
 		return err
 	}
@@ -5627,21 +5965,21 @@ func (d *detector) runUEBANewUserContextRule(ctx context.Context) error {
 	const q = `
 SELECT
    e.hostname,
-    e.target_user_norm,
+    e.target_user,
    e.src_ip,
    e.workstation,
-    MIN(e.ts) AS first_seen,
-    COUNT(*) AS cnt
-FROM event_logs e
+    MIN(e.first_event_ts) AS first_seen,
+    SUM(e.cnt) AS cnt
+FROM event_occurrences e
 WHERE e.channel_name = 'Security'
  AND e.event_id = 4624
-  AND e.ts >= ? AND e.ts < ?
-  AND e.target_user_norm <> ''
-  AND e.target_user_norm NOT LIKE '%$'
-GROUP BY e.hostname, e.target_user_norm, e.src_ip, e.workstation
+  AND e.bucket_start >= ? AND e.bucket_start < ?
+  AND e.target_user <> ''
+  AND e.target_user NOT LIKE '%$'
+GROUP BY e.hostname, e.target_user, e.src_ip, e.workstation
 `

-	rows, err := d.db.QueryContext(ctx, q, windowStart, windowEnd)
+	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd)
 	if err != nil {
 		return fmt.Errorf("query ueba_new_user_context after %s: %w", time.Since(start), err)
 	}
@@ -6046,14 +6384,14 @@ func (d *detector) runFailedLogonSpikeRule(ctx context.Context) error {

 	const q = `
 SELECT hostname, SUM(cnt) AS cnt
-FROM event_occurrences
+FROM event_count_buckets
 WHERE channel_name = 'Security'
  AND event_id = 4625
  AND bucket_start >= ? AND bucket_start < ?
 GROUP BY hostname
 HAVING SUM(cnt) >= ?
 `
-	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd, d.cfg.FailedLogonThreshold)
+	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.BaselineWindow), windowEnd, d.cfg.FailedLogonThreshold)
 	if err != nil {
 		return err
 	}
@@ -6103,14 +6441,14 @@ func (d *detector) runRebootSpikeRule(ctx context.Context) error {

 	const q = `
 SELECT hostname, SUM(cnt) AS cnt
-FROM event_occurrences
+FROM event_count_buckets
 WHERE channel_name = 'System'
  AND event_id IN (1074, 6005, 6006)
  AND bucket_start >= ? AND bucket_start < ?
 GROUP BY hostname
 HAVING SUM(cnt) >= ?
 `
-	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd, d.cfg.RebootThreshold)
+	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.BaselineWindow), windowEnd, d.cfg.RebootThreshold)
 	if err != nil {
 		return err
 	}
@@ -6355,29 +6693,29 @@ func (d *detector) runSuccessAfterFailuresRule(ctx context.Context) error {
 	windowStart := windowEnd.Add(-d.cfg.SuccessAfterFailureWindow)

 	const q = `
-SELECT s.hostname, s.target_user_norm, s.src_ip, COUNT(*) AS success_count
-FROM event_logs s
+SELECT s.hostname, s.target_user, s.src_ip, SUM(s.cnt) AS success_count
+FROM event_occurrences s
 WHERE s.channel_name = 'Security'
  AND s.event_id = 4624
-  AND s.ts >= ? AND s.ts < ?
-  AND s.target_user_norm <> ''
+  AND s.bucket_start >= ? AND s.bucket_start < ?
+  AND s.target_user <> ''
  AND EXISTS (
    SELECT 1
-    FROM event_logs f
+    FROM event_occurrences f
    WHERE f.hostname = s.hostname
      AND f.channel_name = 'Security'
      AND f.event_id = 4625
-      AND f.target_user_norm = s.target_user_norm
+      AND f.target_user = s.target_user
      AND (
           (f.src_ip = s.src_ip AND s.src_ip <> '' AND s.src_ip <> '-')
           OR (s.src_ip = '' OR s.src_ip = '-' OR f.src_ip = '' OR f.src_ip = '-')
      )
-      AND f.ts >= DATE_SUB(s.ts, INTERVAL ? SECOND)
-      AND f.ts < s.ts
+      AND f.last_event_ts >= DATE_SUB(s.first_event_ts, INTERVAL ? SECOND)
+      AND f.first_event_ts < s.last_event_ts
  )
-GROUP BY s.hostname, s.target_user_norm, s.src_ip
+GROUP BY s.hostname, s.target_user, s.src_ip
 `
-	rows, err := d.db.QueryContext(ctx, q, windowStart, windowEnd, int(d.cfg.SuccessAfterFailureWindow.Seconds()))
+	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd, int(d.cfg.SuccessAfterFailureWindow.Seconds()))
 	if err != nil {
 		return err
 	}
@@ -6428,28 +6766,28 @@ func (d *detector) runNewSourceIPForUserRule(ctx context.Context) error {
 	windowStart := windowEnd.Add(-d.cfg.NewSourceIPWindow)

 	const q = `
-SELECT e.hostname, e.target_user_norm, e.src_ip, MIN(e.ts) AS first_seen, COUNT(*) AS cnt
-FROM event_logs e
+SELECT e.hostname, e.target_user, e.src_ip, MIN(e.first_event_ts) AS first_seen, SUM(e.cnt) AS cnt
+FROM event_occurrences e
 WHERE e.channel_name = 'Security'
  AND e.event_id = 4624
-  AND e.ts >= ? AND e.ts < ?
-  AND e.target_user_norm <> ''
-  AND e.target_user_norm NOT LIKE '%$'
+  AND e.bucket_start >= ? AND e.bucket_start < ?
+  AND e.target_user <> ''
+  AND e.target_user NOT LIKE '%$'
  AND e.src_ip <> ''
  AND e.src_ip <> '-'
  AND e.src_ip <> '::1'
  AND e.src_ip <> '127.0.0.1'
-  AND e.target_user_norm NOT IN (
+  AND e.target_user NOT IN (
      'system',
      'localsystem',
      'local service',
      'network service',
      'anonymous logon'
  )
-GROUP BY e.hostname, e.target_user_norm, e.src_ip
+GROUP BY e.hostname, e.target_user, e.src_ip
 `

-	rows, err := d.db.QueryContext(ctx, q, windowStart, windowEnd)
+	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd)
 	if err != nil {
 		return err
 	}
@@ -7076,25 +7414,25 @@ func (d *detector) runOffHoursLoginRule(ctx context.Context) error {
 	windowStart := windowEnd.Add(-5 * time.Minute)

 	const q = `
-SELECT hostname, target_user_norm, COUNT(*) AS cnt
-FROM event_logs
+SELECT hostname, target_user, SUM(cnt) AS cnt
+FROM event_occurrences
 WHERE channel_name = 'Security'
  AND event_id = 4624
-  AND ts >= ? AND ts < ?
-  AND target_user_norm <> ''
-  AND target_user_norm NOT LIKE '%$'
+  AND bucket_start >= ? AND bucket_start < ?
+  AND target_user <> ''
+  AND target_user NOT LIKE '%$'
  AND logon_type IN ('2', '7', '10', '11')
-  AND target_user_norm NOT IN (
+  AND target_user NOT IN (
      'system',
      'localsystem',
      'local service',
      'network service',
      'anonymous logon'
  )
-GROUP BY hostname, target_user_norm
+GROUP BY hostname, target_user
 `

-	rows, err := d.db.QueryContext(ctx, q, windowStart, windowEnd)
+	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd)
 	if err != nil {
 		return err
 	}
@@ -7200,17 +7538,17 @@ func (d *detector) runFirstTimePrivilegedRule(ctx context.Context) error {
 	windowStart := windowEnd.Add(-10 * time.Minute)

 	const q = `
-SELECT hostname, subject_user_norm, COUNT(*) AS cnt
-FROM event_logs
+SELECT hostname, subject_user, SUM(cnt) AS cnt
+FROM event_occurrences
 WHERE channel_name = 'Security'
  AND event_id = 4672
-  AND ts >= ? AND ts < ?
-  AND subject_user_norm <> ''
-  AND subject_user_norm NOT LIKE '%$'
-GROUP BY hostname, subject_user_norm
+  AND bucket_start >= ? AND bucket_start < ?
+  AND subject_user <> ''
+  AND subject_user NOT LIKE '%$'
+GROUP BY hostname, subject_user
 `

-	rows, err := d.db.QueryContext(ctx, q, windowStart, windowEnd)
+	rows, err := d.db.QueryContext(ctx, q, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd)
 	if err != nil {
 		return err
 	}
@@ -7352,23 +7690,23 @@ func (d *detector) runAdminNewHostRule(ctx context.Context) error {
 	lookbackStart := windowEnd.Add(-d.cfg.UEBALookback)

 	rows, err := d.db.QueryContext(ctx, `
-SELECT e.hostname, e.target_user_norm, COUNT(*) AS cnt
-FROM event_logs e
+SELECT e.hostname, e.target_user, SUM(e.cnt) AS cnt
+FROM event_occurrences e
 WHERE e.channel_name = 'Security'
  AND e.event_id = 4624
-  AND e.ts >= ? AND e.ts < ?
-  AND e.target_user_norm <> ''
-  AND e.target_user_norm NOT LIKE '%$'
+  AND e.bucket_start >= ? AND e.bucket_start < ?
+  AND e.target_user <> ''
+  AND e.target_user NOT LIKE '%$'
  AND NOT EXISTS (
      SELECT 1
      FROM ueba_user_baseline b
-      WHERE b.username = e.target_user_norm
+      WHERE b.username = e.target_user
        AND b.hostname = e.hostname
        AND b.last_seen >= ?
        AND b.first_seen < ?
  )
-GROUP BY e.hostname, e.target_user_norm
-`, windowStart, windowEnd, lookbackStart, windowStart)
+GROUP BY e.hostname, e.target_user
+`, bucketStart(windowStart, d.cfg.MetadataBucket), windowEnd, lookbackStart, windowStart)
 	if err != nil {
 		return err
 	}
--- a/main_test.go
+++ b/main_test.go
@@ -0,0 +1,131 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+func TestNormalizeLockoutEvent4740(t *testing.T) {
+	xml := `<Event><System><Provider Name="Microsoft-Windows-Security-Auditing"/><Computer>DC01.example.local</Computer></System><EventData><Data Name="TargetUserName">alice</Data><Data Name="TargetDomainName">EXAMPLE</Data><Data Name="SubjectUserName">DC01$</Data><Data Name="CallerComputerName">CLIENT-42</Data></EventData></Event>`
+	got := NormalizeEventXML(xml)
+	if got.TargetUser != "alice" {
+		t.Fatalf("TargetUser=%q", got.TargetUser)
+	}
+	if got.Workstation != "CLIENT-42" {
+		t.Fatalf("Workstation=%q", got.Workstation)
+	}
+	if got.Computer != "DC01.example.local" {
+		t.Fatalf("Computer=%q", got.Computer)
+	}
+}
+
+func TestMetadataOnlyOverlay(t *testing.T) {
+	got := overlayMetadata(NormalizedEvent{}, EventMetadataPayload{
+		Computer:   "DC01",
+		TargetUser: "EXAMPLE\\alice",
+		Device:     "CLIENT-42",
+		SrcIP:      "10.0.0.42",
+	})
+	if got.Workstation != "CLIENT-42" {
+		t.Fatalf("Workstation=%q", got.Workstation)
+	}
+	if normalizeUsername(got.TargetUser) != "alice" {
+		t.Fatalf("normalized user=%q", normalizeUsername(got.TargetUser))
+	}
+}
+
+func TestBucketStart(t *testing.T) {
+	ts := time.Date(2026, 7, 18, 12, 34, 56, 0, time.UTC)
+	got := bucketStart(ts, time.Minute)
+	want := time.Date(2026, 7, 18, 12, 34, 0, 0, time.UTC)
+	if !got.Equal(want) {
+		t.Fatalf("got %s want %s", got, want)
+	}
+}
+
+func TestNewEventIDIgnoresWMIActivityOperational(t *testing.T) {
+	cfg := Config{
+		NewEventIDMode:           "inventory",
+		NewEventIDMinCount:       3,
+		NewEventIDLearningPeriod: 24 * time.Hour,
+		NewEventIDIgnoreChannels: []string{"Microsoft-Windows-WMI-Activity/Operational"},
+	}
+	agentFirst := time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC)
+	eventFirst := time.Date(2026, 7, 18, 18, 0, 0, 0, time.UTC)
+	got := assessNewEventID(cfg, "Microsoft-Windows-WMI-Activity/Operational", 5857, 12, agentFirst, eventFirst)
+	if got.Emit {
+		t.Fatalf("WMI operational event should be ignored: %+v", got)
+	}
+	if got.Reason != "ignored_operational_channel" {
+		t.Fatalf("reason=%q", got.Reason)
+	}
+}
+
+func TestNewEventIDInventoryIsInformationalAndPlausible(t *testing.T) {
+	cfg := Config{
+		NewEventIDMode:           "inventory",
+		NewEventIDMinCount:       3,
+		NewEventIDLearningPeriod: 24 * time.Hour,
+	}
+	agentFirst := time.Date(2026, 7, 1, 10, 0, 0, 0, time.UTC)
+	eventFirst := time.Date(2026, 7, 18, 18, 0, 0, 0, time.UTC)
+	got := assessNewEventID(cfg, "Application", 1000, 3, agentFirst, eventFirst)
+	if !got.Emit || got.Severity != "info" || got.Status != "plausible" {
+		t.Fatalf("unexpected assessment: %+v", got)
+	}
+}
+
+func TestNewEventIDSkipsAgentLearningPeriod(t *testing.T) {
+	cfg := Config{
+		NewEventIDMode:           "alert",
+		NewEventIDMinCount:       1,
+		NewEventIDLearningPeriod: 24 * time.Hour,
+	}
+	agentFirst := time.Date(2026, 7, 18, 17, 30, 0, 0, time.UTC)
+	eventFirst := time.Date(2026, 7, 18, 18, 0, 0, 0, time.UTC)
+	got := assessNewEventID(cfg, "Security", 4719, 4, agentFirst, eventFirst)
+	if got.Emit || got.Reason != "agent_learning_period" {
+		t.Fatalf("unexpected assessment: %+v", got)
+	}
+}
+
+func TestMetadataContextAllowlist(t *testing.T) {
+	cfg := Config{MetadataContextEventIDs: map[uint32]struct{}{4624: {}, 4740: {}}}
+	if !shouldStoreMetadataContext(cfg, "Security", 4740) {
+		t.Fatal("Security/4740 should retain compact context")
+	}
+	if shouldStoreMetadataContext(cfg, "Microsoft-Windows-WMI-Activity/Operational", 4740) {
+		t.Fatal("non-Security channel must not create context rows")
+	}
+	if shouldStoreMetadataContext(cfg, "Security", 4688) {
+		t.Fatal("event outside allowlist must stay count-only")
+	}
+}
+
+func TestCompactLockoutContextDropsNoise(t *testing.T) {
+	norm := NormalizedEvent{
+		TargetUser:    "EXAMPLE\\Alice",
+		SubjectUser:   "DC01$",
+		Workstation:   "CLIENT-42",
+		SrcIP:         "10.0.0.10",
+		LogonType:     "3",
+		StatusText:    "status-varies",
+		FailureReason: "large noisy reason",
+	}
+	target, subject, srcIP, workstation, logonType, statusText, failureReason := compactMetadataContext(4740, norm)
+	if target != "alice" || workstation != "CLIENT-42" {
+		t.Fatalf("unexpected lockout identity target=%q workstation=%q", target, workstation)
+	}
+	if subject != "" || srcIP != "" || logonType != "" || statusText != "" || failureReason != "" {
+		t.Fatalf("lockout context retained high-cardinality noise: subject=%q src=%q logon=%q status=%q reason=%q", subject, srcIP, logonType, statusText, failureReason)
+	}
+}
+
+func TestRetainCompactContextDropsMachineLogons(t *testing.T) {
+	if retainCompactContext(4624, "server01$", "") {
+		t.Fatal("machine-account logon context should not be retained")
+	}
+	if !retainCompactContext(4624, "alice", "") {
+		t.Fatal("human logon context should be retained")
+	}
+}
--- a/schema.legacy.sql
+++ b/schema.legacy.sql
@@ -0,0 +1,101 @@
+CREATE DATABASE IF NOT EXISTS eventcollector
+  CHARACTER SET utf8mb4
+  COLLATE utf8mb4_unicode_ci;
+
+USE eventcollector;
+
+CREATE TABLE IF NOT EXISTS agents (
+    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    hostname VARCHAR(255) NOT NULL,
+    api_key_hash CHAR(64) NOT NULL,
+    first_seen DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+    last_seen DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+    last_ip VARCHAR(64) NOT NULL DEFAULT '',
+    is_enabled TINYINT(1) NOT NULL DEFAULT 1,
+    PRIMARY KEY (id),
+    UNIQUE KEY ux_agents_hostname (hostname),
+    KEY ix_agents_last_seen (last_seen)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+CREATE TABLE IF NOT EXISTS event_logs (
+    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    agent_id BIGINT UNSIGNED NOT NULL,
+    hostname VARCHAR(255) NOT NULL,
+    channel_name VARCHAR(128) NOT NULL,
+    event_id INT UNSIGNED NOT NULL,
+    source VARCHAR(255) NOT NULL,
+    ts DATETIME(6) NOT NULL,
+    received_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+    msg LONGTEXT NOT NULL,
+    msg_sha256 CHAR(64) NOT NULL,
+    PRIMARY KEY (id),
+    KEY ix_event_logs_ts (ts),
+    KEY ix_event_logs_received_at (received_at),
+    KEY ix_event_logs_agent_ts (agent_id, ts),
+    KEY ix_event_logs_eventid_ts (event_id, ts),
+    KEY ix_event_logs_hostname_ts (hostname, ts),
+    KEY ix_event_logs_channel_event_ts (channel_name, event_id, ts),
+    CONSTRAINT fk_event_logs_agent
+        FOREIGN KEY (agent_id) REFERENCES agents(id)
+        ON DELETE RESTRICT
+        ON UPDATE RESTRICT
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+CREATE TABLE IF NOT EXISTS detections (
+    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
+    rule_name VARCHAR(128) NOT NULL,
+    severity VARCHAR(32) NOT NULL,
+    hostname VARCHAR(255) NOT NULL,
+    channel_name VARCHAR(128) NOT NULL DEFAULT '',
+    event_id INT UNSIGNED NOT NULL DEFAULT 0,
+    score DOUBLE NOT NULL DEFAULT 0,
+    window_start DATETIME(6) NOT NULL,
+    window_end DATETIME(6) NOT NULL,
+    summary VARCHAR(512) NOT NULL,
+    details_json JSON NOT NULL,
+    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
+    PRIMARY KEY (id),
+    UNIQUE KEY ux_detection_dedupe (rule_name, hostname, channel_name, event_id, window_start, window_end),
+    KEY ix_detections_created (created_at),
+    KEY ix_detections_rule_host_time (rule_name, hostname, created_at),
+    KEY ix_detections_severity_time (severity, created_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+
+USE eventcollector;
+
+INSERT INTO agents (hostname, api_key_hash)
+VALUES
+  ('client01.domain.local', SHA2('SUPER-LANGER-AGENT-KEY-01', 256)),
+  ('client02.domain.local', SHA2('SUPER-LANGER-AGENT-KEY-02', 256));
+
+  #V2
+
+  ALTER TABLE event_logs
+    ADD COLUMN computer VARCHAR(255) NOT NULL DEFAULT '' AFTER source,
+    ADD COLUMN provider_name VARCHAR(255) NOT NULL DEFAULT '' AFTER computer,
+    ADD COLUMN level_value INT UNSIGNED NOT NULL DEFAULT 0 AFTER provider_name,
+    ADD COLUMN task_value INT UNSIGNED NOT NULL DEFAULT 0 AFTER level_value,
+    ADD COLUMN opcode_value INT UNSIGNED NOT NULL DEFAULT 0 AFTER task_value,
+    ADD COLUMN keywords VARCHAR(255) NOT NULL DEFAULT '' AFTER opcode_value,
+    ADD COLUMN target_user VARCHAR(255) NOT NULL DEFAULT '' AFTER keywords,
+    ADD COLUMN target_domain VARCHAR(255) NOT NULL DEFAULT '' AFTER target_user,
+    ADD COLUMN subject_user VARCHAR(255) NOT NULL DEFAULT '' AFTER target_domain,
+    ADD COLUMN subject_domain VARCHAR(255) NOT NULL DEFAULT '' AFTER subject_user,
+    ADD COLUMN workstation VARCHAR(255) NOT NULL DEFAULT '' AFTER subject_domain,
+    ADD COLUMN src_ip VARCHAR(64) NOT NULL DEFAULT '' AFTER workstation,
+    ADD COLUMN src_port VARCHAR(32) NOT NULL DEFAULT '' AFTER src_ip,
+    ADD COLUMN logon_type VARCHAR(32) NOT NULL DEFAULT '' AFTER src_port,
+    ADD COLUMN process_name VARCHAR(512) NOT NULL DEFAULT '' AFTER logon_type,
+    ADD COLUMN authentication_package VARCHAR(128) NOT NULL DEFAULT '' AFTER process_name,
+    ADD COLUMN logon_process VARCHAR(128) NOT NULL DEFAULT '' AFTER authentication_package,
+    ADD COLUMN status_text VARCHAR(64) NOT NULL DEFAULT '' AFTER logon_process,
+    ADD COLUMN sub_status_text VARCHAR(64) NOT NULL DEFAULT '' AFTER status_text,
+    ADD COLUMN failure_reason VARCHAR(512) NOT NULL DEFAULT '' AFTER sub_status_text;
+
+ALTER TABLE event_logs
+    ADD KEY ix_event_logs_target_user_ts (target_user, ts),
+    ADD KEY ix_event_logs_src_ip_ts (src_ip, ts),
+    ADD KEY ix_event_logs_target_user_src_ip_ts (target_user, src_ip, ts),
+    ADD KEY ix_event_logs_eventid_srcip_ts (event_id, src_ip, ts),
+    ADD KEY ix_event_logs_eventid_targetuser_ts (event_id, target_user, ts),
+    ADD KEY ix_event_logs_eventid_logontype_ts (event_id, logon_type, ts);