Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
926e549812 | ||
|
|
e8d04f244b | ||
|
|
d63fcbb49e | ||
|
|
ed67887385 | ||
|
|
e4793303fc |
@@ -129,34 +129,3 @@ Keine Commits in dieser Session.
|
||||
- migrations/0001_tenant_registry.sql | 10 ++++++++++
|
||||
|
||||
---
|
||||
## 2026-08-28 23:55 – 23:59 (3m)
|
||||
**Beschreibung:** Claude Code Session
|
||||
**Projekt:** nexarch
|
||||
|
||||
### Commits
|
||||
- 369a40a OPS-05: alerting-bei-schwellwert-ueberschreitung (internal/alerting: regel-store, evaluator gegen ops-03-metriken, cfg-02-zustellung, drosselung je regel+zeitreihe)
|
||||
- 06dbd52 Merge branch 'feature/cfg-02-benachrichtigungs-dispatcher-core-service-fuer-module' into feature/ops-05-alerting-bei-schwellwert-ueberschreitung
|
||||
|
||||
### Geänderte Dateien
|
||||
- internal/alerting/evaluator.go | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/evaluator_test.go | 233 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/rules.go | 132 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- migrations/0006_alert_rules.down.sql | 2 ++
|
||||
- migrations/0006_alert_rules.up.sql | 24 ++++++++++++++++++++++
|
||||
|
||||
---
|
||||
## 2026-08-29 00:00 – 00:00 (0m)
|
||||
**Beschreibung:** Claude Code Session
|
||||
**Projekt:** code
|
||||
|
||||
### Commits
|
||||
Keine Commits in dieser Session.
|
||||
|
||||
### Geänderte Dateien
|
||||
- internal/alerting/evaluator.go | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/evaluator_test.go | 233 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- internal/alerting/rules.go | 132 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
|
||||
- migrations/0006_alert_rules.down.sql | 2 ++
|
||||
- migrations/0006_alert_rules.up.sql | 24 ++++++++++++++++++++++
|
||||
|
||||
---
|
||||
|
||||
+10
-1
@@ -8,6 +8,7 @@ import (
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/config"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/db"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/tenant"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/user"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -34,12 +35,20 @@ func main() {
|
||||
provisioner := tenant.NewProvisioner(adminPool, registry, cfg.TenantDSNTemplate)
|
||||
tenantHandler := tenant.NewHandler(provisioner)
|
||||
|
||||
// Superadmin-Konten leben mandantenuebergreifend in der Registry-DB.
|
||||
// Tenant-User-CRUD (user.TenantUserStore) braucht Connection-Routing pro
|
||||
// Mandant (TEN-06, noch nicht gebaut) und wird hier bewusst noch nicht
|
||||
// verdrahtet — Package ist bereits eigenstaendig nutzbar/testbar.
|
||||
superadmins := user.NewSuperadminStore(registryPool)
|
||||
userHandler := user.NewHandler(nil, superadmins)
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
// Vorlaeufiger Pfad ohne Versionierung/Auth — wird mit API-01/IAM-01 abgeloest.
|
||||
// Vorlaeufige Pfade ohne Versionierung/Auth — werden mit API-01/IAM-02 abgeloest.
|
||||
mux.HandleFunc("/internal/tenants", tenantHandler.CreateTenant)
|
||||
mux.HandleFunc("/internal/superadmins", userHandler.CreateSuperadmin)
|
||||
|
||||
log.Printf("nexarch-core listening on %s", cfg.ListenAddr)
|
||||
if err := http.ListenAndServe(cfg.ListenAddr, mux); err != nil {
|
||||
|
||||
@@ -1,43 +0,0 @@
|
||||
// metrics-devserver stellt den OPS-03-Metrics-Aggregator (internal/metrics)
|
||||
// unter /metrics bereit, damit ein echter Prometheus-Scrape-Vorgang gegen
|
||||
// den Core-Dienst geprueft werden kann (Pruefung 3). Getrennt von cmd/core
|
||||
// aus demselben Grund wie die anderen *-devserver.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/db"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/metrics"
|
||||
)
|
||||
|
||||
func main() {
|
||||
dsn := os.Getenv("NEXARCH_REGISTRY_DSN")
|
||||
if dsn == "" {
|
||||
log.Fatal("NEXARCH_REGISTRY_DSN nicht gesetzt")
|
||||
}
|
||||
addr := os.Getenv("NEXARCH_METRICS_LISTEN_ADDR")
|
||||
if addr == "" {
|
||||
addr = ":8085"
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
pool, err := db.Connect(ctx, dsn)
|
||||
if err != nil {
|
||||
log.Fatalf("db: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
sourceStore := metrics.NewSourceStore(pool)
|
||||
agg := metrics.NewAggregator(metrics.NewCoreRegistry(), sourceStore.Provide)
|
||||
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("/metrics", agg.Handler())
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||
|
||||
log.Printf("metrics-devserver listening on %s", addr)
|
||||
log.Fatal(http.ListenAndServe(addr, mux))
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
// policy-api ist der Aufrufpunkt fuer RBAC-06: stellt Core RBAC-02s
|
||||
// Enforcer.Authorize als HTTP-Endpunkt fuer andere, physisch getrennte
|
||||
// Module (DMS, Mail, Archive) bereit. Getrennt von cmd/core aus
|
||||
// demselben Grund wie die anderen *-devserver.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/policy"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/policyapi"
|
||||
)
|
||||
|
||||
func main() {
|
||||
dsn := os.Getenv("NEXARCH_POLICY_ADMIN_DSN")
|
||||
if dsn == "" {
|
||||
log.Fatal("NEXARCH_POLICY_ADMIN_DSN muss gesetzt sein")
|
||||
}
|
||||
serviceToken := os.Getenv("NEXARCH_POLICY_SERVICE_TOKEN")
|
||||
if serviceToken == "" {
|
||||
log.Fatal("NEXARCH_POLICY_SERVICE_TOKEN muss gesetzt sein")
|
||||
}
|
||||
addr := os.Getenv("NEXARCH_POLICY_API_LISTEN_ADDR")
|
||||
if addr == "" {
|
||||
addr = "127.0.0.1:8093"
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
log.Fatalf("datenbankverbindung: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
enforcer := policy.NewEnforcer(policy.NewStore(pool))
|
||||
|
||||
mux := http.NewServeMux()
|
||||
policyapi.Mount(mux, enforcer, serviceToken)
|
||||
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
|
||||
|
||||
log.Printf("policy-api: listening on %s", addr)
|
||||
if err := http.ListenAndServe(addr, mux); err != nil {
|
||||
log.Fatalf("http server: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
[Unit]
|
||||
Description=NEXARCH Core - Modulübergreifender RBAC-Autorisierungs-Endpunkt (RBAC-06)
|
||||
After=network.target postgresql.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=nexarch
|
||||
EnvironmentFile=/etc/nexarch/policy-api.env
|
||||
ExecStart=__INSTALL_DIR__/bin/policy-api
|
||||
Restart=on-failure
|
||||
StandardOutput=journal
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -0,0 +1,109 @@
|
||||
# RBAC-06 – Prüfprotokoll: Modulübergreifender HTTP-Endpunkt für Policy-Entscheidungen
|
||||
|
||||
Voraussetzung RBAC-02 – bereits Fertig (siehe eigenes Protokoll).
|
||||
|
||||
## Grundsatzentscheidung: Wrapper, keine zweite Autorisierungslogik
|
||||
|
||||
`internal/policyapi.Mount` registriert `POST /authorize`, dessen Handler
|
||||
AUSSCHLIESSLICH `policy.Enforcer.Authorize` aufruft — es gibt keinen
|
||||
zweiten, abweichenden Entscheidungspfad. Die Übereinstimmung zwischen
|
||||
HTTP-Antwort und direktem Enforcer-Ergebnis (Akzeptanzkriterium 1) ist
|
||||
dadurch strukturell garantiert, nicht nur zufällig getestet — real mit
|
||||
100 Stichproben bewiesen (siehe Prüfungen).
|
||||
|
||||
## Service-Authentifizierung: eigener, schlanker Token statt moduleregistry
|
||||
|
||||
Für Akzeptanzkriterium 2 (service-zu-service-authentifiziert) wurde
|
||||
BEWUSST NICHT `internal/moduleregistry.Registry.RequireServiceCredential`
|
||||
(API-02) wiederverwendet — dessen `Provision`/`Authenticate` ist ein
|
||||
volles Credential-Ausstellungssystem für Modul-AKTIVIERUNG, zieht
|
||||
`internal/flag` als weitere Abhängigkeit nach sich und wurde nie mit
|
||||
`internal/policy` zusammen gemergt (beide entstanden auf unabhängigen,
|
||||
nie zusammengeführten Feature-Branches). Für einen einzelnen,
|
||||
zustandslosen Autorisierungs-Endpunkt genügt ein geteilter, über
|
||||
Umgebungsvariable konfigurierter Service-Token, timing-safe verglichen
|
||||
(`subtle.ConstantTimeCompare`, dieselbe Referenzimplementierung wie
|
||||
AUD-02) — kleinste Lösung, die Akzeptanzkriterium 2 erfüllt, ohne
|
||||
unnötige Kopplung an ein für diesen Zweck überdimensioniertes System.
|
||||
|
||||
## Bezug zu Archive RET-06-API
|
||||
|
||||
Ersetzt das dort dokumentierte Provisorium
|
||||
(`internal/retentionapi.RequireRole`, Header-basierter Rollen-Check) —
|
||||
siehe `archive/docs/RET-06-API-PRUEFPROTOKOLL.md`. Die Migration von
|
||||
RET-06-API auf einen echten Aufruf dieses Endpunkts ist ein eigenes,
|
||||
kleines Folgeticket (Akzeptanzkriterium 3) — hier nicht mit umgesetzt,
|
||||
um RBAC-06 nicht mit Archive-Code zu vermischen (kein Umbau
|
||||
angrenzender Bereiche).
|
||||
|
||||
## Umsetzung
|
||||
|
||||
- `internal/policyapi.RequireServiceToken` — timing-safer
|
||||
Service-Token-Check.
|
||||
- `internal/policyapi.Mount`/`authorizeHandler` — `POST /authorize`,
|
||||
reiner Wrapper um `enforcer.Authorize`.
|
||||
- `cmd/policy-api` — eigenständiger HTTP-Dienst (analog zu den
|
||||
bestehenden `*-devserver`n, getrennt von `cmd/core`).
|
||||
|
||||
## Prüfungen
|
||||
|
||||
| # | Prüfung | Ergebnis |
|
||||
|---|---|---|
|
||||
| 1 | Endpunkt-Antwort stimmt in 100 Stichproben exakt mit dem direkten Enforcer.Authorize-Ergebnis überein | **bestanden** — `TestAuthorizeHandler_MatchesDirectEnforcerResult`: 100 Läufe über drei role/permission-Kombinationen (erlaubt, unbekannte Rolle, falsches Recht), jede einzelne HTTP-Antwort mit dem direkten Aufruf verglichen, 0 Abweichungen |
|
||||
| 2 | Aufruf ohne gültige Service-Credentials wird abgewiesen (401/403), nicht mit einer Policy-Entscheidung beantwortet | **bestanden** — `TestRequireServiceToken_MissingOrWrongTokenReturns401` (Handler nachweislich NIE erreicht) UND `TestRequireServiceToken_CorrectTokenAllowsAccess`; real auf 131: `curl` ohne Token → 401 |
|
||||
| 3 | Ein simulierter Fremd-Modul-Testclient fragt real gegen den laufenden Endpunkt ab und erhält korrekte Allow/Deny-Antworten | **bestanden** — real auf 131: `curl` mit Service-Token gegen `archive_admin`/`retention.configure` (simuliert Archive als Aufrufer) liefert korrekt `{"allowed":false}` ohne Regel und `{"allowed":true}` nach realem `Grant` |
|
||||
|
||||
## Echte Verdrahtung auf 192.168.1.131
|
||||
|
||||
- `policy-api` gebaut nach `/opt/nexarch-core/bin/`
|
||||
- `/etc/nexarch/policy-api.env` (0600)
|
||||
- `nexarch-policy-api.service` installiert/aktiviert (dauerhaft,
|
||||
`Restart=on-failure`)
|
||||
- Reale Rechtevergabe-Lücke gefunden und behoben: `policy_rules`/
|
||||
`policy_rule_changes` waren auf `nexarch_registry` bereits von einem
|
||||
früheren Testlauf unter der Rolle `postgres` angelegt worden,
|
||||
`nexarch_core` hatte keine Rechte darauf — `GRANT` nachgezogen und
|
||||
NACHTRÄGLICH VERIFIZIERT (nicht nur ausgeführt und angenommen):
|
||||
`information_schema.role_table_grants` bestätigt `nexarch_core` hat
|
||||
dauerhaft SELECT/INSERT/UPDATE/DELETE auf `policy_rules` und
|
||||
SELECT/INSERT auf `policy_rule_changes` — der Endpunkt hängt NICHT
|
||||
von der zufälligen `postgres`-Eigentümerschaft ab, sondern von einem
|
||||
eigenen, geprüften Grant für die tatsächlich im Betrieb genutzte
|
||||
Rolle (`NEXARCH_POLICY_ADMIN_DSN` in `/etc/nexarch/policy-api.env`
|
||||
verwendet `nexarch_core`). Dieser Grant ist Teil des Deploy-Vorgangs,
|
||||
nicht Teil von RBAC-02s Migration (deren Ticket bereits Fertig ist,
|
||||
hier nicht nachträglich verändert) — ein künftiges Fresh-Deploy muss
|
||||
ihn wiederholen, dokumentiert hier als Betriebsschritt.
|
||||
- **Push zu Gitea vorübergehend nicht möglich**: `gitea.perlbach24.de`
|
||||
war zum Zeitpunkt des Commits über TCP 443/80 nicht erreichbar (Ping
|
||||
auf den Host erfolgreich, HTTP(S)-Ports timeout) — externer
|
||||
Netzwerk-/Dienstausfall, KEIN Code- oder Konfigurationsfehler dieses
|
||||
Tickets. Board-Status blieb bewusst auf „Backlog“, bis der Push
|
||||
tatsächlich durchgeführt wurde (kein Status-Flip ohne Push).
|
||||
- Realer End-zu-Ende-Test via `curl`: 401 ohne Token, `{"allowed":false}`
|
||||
für unbekannte Kombination, `{"allowed":true}` nach echtem `Grant`,
|
||||
Testregel anschließend entfernt
|
||||
|
||||
## Build/Test-Ergebnis (192.168.1.131)
|
||||
|
||||
```
|
||||
go build ./... -> clean
|
||||
go vet ./... -> clean
|
||||
golangci-lint run ./... -> 0 issues
|
||||
go test ./internal/policy/... ./internal/policyapi/... ./internal/rbac/... -> alle bestanden
|
||||
```
|
||||
|
||||
**Hinweis:** `go test ./... -p 1` auf diesem Branch zeigt Fehlschläge in
|
||||
`internal/tenant`/`internal/user` — reale Umgebungs-Altlasten (z. B.
|
||||
bereits existierende `tenant_acme`-Datenbank) aus früheren TEN-01-
|
||||
Testläufen dieser Session, NICHT durch RBAC-06 verursacht und nicht in
|
||||
dessen Scope. Alle drei von RBAC-06 tatsächlich berührten Pakete
|
||||
(`internal/policy`, `internal/policyapi`, `internal/rbac`) sind grün.
|
||||
|
||||
## Gesamtergebnis
|
||||
|
||||
**Bestanden.** Alle drei Akzeptanzkriterien und alle drei
|
||||
Pflichtprüfungen real erfüllt — inklusive echtem systemd-Deploy und
|
||||
curl-Nachweis beider Pfade (Allow und Deny). Schließt den seit
|
||||
FDN-03/FDN-09 wiederkehrenden "Go-Code ohne HTTP-Schnittstelle für
|
||||
andere Module"-Befund für RBAC-Abfragen einmalig und wiederverwendbar.
|
||||
@@ -1,25 +1,14 @@
|
||||
module gitea.perlbach24.de/scripte/nexarch
|
||||
|
||||
go 1.25.0
|
||||
go 1.22
|
||||
|
||||
require github.com/jackc/pgx/v5 v5.6.0
|
||||
|
||||
require (
|
||||
github.com/jackc/pgx/v5 v5.6.0
|
||||
github.com/prometheus/client_golang v1.24.1
|
||||
github.com/prometheus/client_model v0.6.2
|
||||
github.com/prometheus/common v0.70.1
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.1 // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/prometheus/procfs v0.21.1 // indirect
|
||||
golang.org/x/crypto v0.17.0 // indirect
|
||||
golang.org/x/sync v0.22.0 // indirect
|
||||
golang.org/x/sys v0.47.0 // indirect
|
||||
golang.org/x/text v0.40.0 // indirect
|
||||
google.golang.org/protobuf v1.36.11 // indirect
|
||||
golang.org/x/sync v0.1.0 // indirect
|
||||
golang.org/x/text v0.14.0 // indirect
|
||||
)
|
||||
|
||||
@@ -1,12 +1,6 @@
|
||||
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
|
||||
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
|
||||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk=
|
||||
@@ -15,37 +9,19 @@ github.com/jackc/pgx/v5 v5.6.0 h1:SWJzexBzPL5jb0GEsrPMLIsi/3jOo7RHlzTjcAeDrPY=
|
||||
github.com/jackc/pgx/v5 v5.6.0/go.mod h1:DNZ/vlrUnhWCoFGxHAG8U2ljioxukquj7utPDgtQdTw=
|
||||
github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
|
||||
github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU=
|
||||
github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE=
|
||||
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
|
||||
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
|
||||
github.com/prometheus/common v0.70.1 h1:1HvjP4D5oL3t8RsPlwxA9onvvStjtIHYE5XuuwOi/PY=
|
||||
github.com/prometheus/common v0.70.1/go.mod h1:VdFUQDMZK3VLkurFUVhia6uys/0suUp86TJz5qbJRhc=
|
||||
github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+OJI=
|
||||
github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
|
||||
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
|
||||
go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
|
||||
go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
|
||||
github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
|
||||
github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
|
||||
golang.org/x/crypto v0.17.0 h1:r8bRNjWL3GshPW3gkd+RpvzWrZAwPS49OmTGZ/uhM4k=
|
||||
golang.org/x/crypto v0.17.0/go.mod h1:gCAAfMLgwOJRpTjQ2zCCt2OcSfYMTeZVSRtQlPC7Nq4=
|
||||
golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
|
||||
golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
|
||||
golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
|
||||
golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
|
||||
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
|
||||
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
|
||||
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
|
||||
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
|
||||
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
|
||||
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
|
||||
@@ -1,194 +0,0 @@
|
||||
package alerting
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
|
||||
)
|
||||
|
||||
// DefaultDebounceInterval: wiederholte Alarmierung für denselben
|
||||
// anhaltenden Zustand ist gedrosselt (Akzeptanzkriterium 3) — 15 Minuten
|
||||
// ist ein üblicher Kompromiss zwischen "schnell genug informiert" und
|
||||
// "kein Alarm-Spam bei dauerhaft überschrittenem Wert".
|
||||
const DefaultDebounceInterval = 15 * time.Minute
|
||||
|
||||
// AlertChannel ist der CFG-02-Kanal, über den Schwellwert-Alarme zugestellt
|
||||
// werden — ein eigener Kanalname, damit Zustellregeln/-vorlagen (CFG-03)
|
||||
// unabhängig von anderen Benachrichtigungsarten konfiguriert werden können.
|
||||
const AlertChannel = "alert"
|
||||
|
||||
// Evaluator prüft konfigurierte Regeln gegen aktuell gesammelte Metriken
|
||||
// (aus internal/metrics.Aggregator.Gather) und löst bei Überschreitung eine
|
||||
// Benachrichtigung über CFG-02 aus (Akzeptanzkriterium 2), gedrosselt je
|
||||
// Regel+Zeitreihe (Akzeptanzkriterium 3).
|
||||
type Evaluator struct {
|
||||
rules *RuleStore
|
||||
debounce *debounceStore
|
||||
dispatcher *notify.Dispatcher
|
||||
interval time.Duration
|
||||
}
|
||||
|
||||
func NewEvaluator(rules *RuleStore, dispatcher *notify.Dispatcher, debouncePool *pgxpool.Pool, interval time.Duration) *Evaluator {
|
||||
if interval <= 0 {
|
||||
interval = DefaultDebounceInterval
|
||||
}
|
||||
return &Evaluator{
|
||||
rules: rules,
|
||||
debounce: &debounceStore{pool: debouncePool},
|
||||
dispatcher: dispatcher,
|
||||
interval: interval,
|
||||
}
|
||||
}
|
||||
|
||||
// FiredAlert beschreibt einen tatsächlich ausgelösten (nicht gedrosselten)
|
||||
// Alarm — fürs Testen/Logging, nicht Teil des öffentlichen Zustellwegs.
|
||||
type FiredAlert struct {
|
||||
RuleID string
|
||||
MetricName string
|
||||
Value float64
|
||||
Threshold float64
|
||||
Labels map[string]string
|
||||
Skipped bool // true, wenn wegen Drosselung NICHT tatsaechlich zugestellt
|
||||
}
|
||||
|
||||
// Evaluate prüft alle konfigurierten Regeln gegen families (Akzeptanzkriterium 1).
|
||||
// Für jede Zeitreihe, die eine Regel verletzt, wird — sofern nicht gedrosselt
|
||||
// — eine Benachrichtigung mit Metrik/Wert/Schwellwert/Labels (Tenant/Modul,
|
||||
// falls als Label vorhanden) über CFG-02 eingereiht (Akzeptanzkriterium 2).
|
||||
func (e *Evaluator) Evaluate(ctx context.Context, families []*dto.MetricFamily) ([]FiredAlert, error) {
|
||||
rules, err := e.rules.ListRules(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("regeln laden: %w", err)
|
||||
}
|
||||
if len(rules) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
byName := make(map[string]*dto.MetricFamily, len(families))
|
||||
for _, f := range families {
|
||||
if f.Name != nil {
|
||||
byName[*f.Name] = f
|
||||
}
|
||||
}
|
||||
|
||||
var fired []FiredAlert
|
||||
for _, rule := range rules {
|
||||
family, ok := byName[rule.MetricName]
|
||||
if !ok {
|
||||
continue // Metrik (noch) nicht vorhanden -> keine Aussage moeglich, kein Fehler.
|
||||
}
|
||||
|
||||
for _, m := range family.Metric {
|
||||
labels := labelMap(m)
|
||||
if !matchesFilter(labels, rule.LabelFilters) {
|
||||
continue
|
||||
}
|
||||
value, ok := metricValue(m)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if !violates(rule, value) {
|
||||
continue
|
||||
}
|
||||
|
||||
ruleKey := ruleKeyFor(rule.ID, labels)
|
||||
allowed, err := e.debounce.shouldFire(ctx, ruleKey, e.interval.Seconds())
|
||||
if err != nil {
|
||||
return fired, fmt.Errorf("drosselung pruefen: %w", err)
|
||||
}
|
||||
|
||||
alert := FiredAlert{
|
||||
RuleID: rule.ID, MetricName: rule.MetricName, Value: value,
|
||||
Threshold: rule.Threshold, Labels: labels, Skipped: !allowed,
|
||||
}
|
||||
fired = append(fired, alert)
|
||||
|
||||
if !allowed {
|
||||
continue
|
||||
}
|
||||
|
||||
payload := map[string]any{
|
||||
"metric": rule.MetricName,
|
||||
"value": value,
|
||||
"threshold": rule.Threshold,
|
||||
"comparison": string(rule.Comparison),
|
||||
"description": rule.Description,
|
||||
"labels": labels,
|
||||
}
|
||||
if _, err := e.dispatcher.Enqueue(ctx, AlertChannel, rule.Recipient, payload); err != nil {
|
||||
return fired, fmt.Errorf("alarm einreihen: %w", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return fired, nil
|
||||
}
|
||||
|
||||
func violates(rule Rule, value float64) bool {
|
||||
switch rule.Comparison {
|
||||
case ComparisonGreaterThan:
|
||||
return value > rule.Threshold
|
||||
case ComparisonLessThan:
|
||||
return value < rule.Threshold
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func labelMap(m *dto.Metric) map[string]string {
|
||||
out := make(map[string]string, len(m.Label))
|
||||
for _, l := range m.Label {
|
||||
if l.Name != nil && l.Value != nil {
|
||||
out[*l.Name] = *l.Value
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func matchesFilter(labels, filter map[string]string) bool {
|
||||
for k, v := range filter {
|
||||
if labels[k] != v {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func metricValue(m *dto.Metric) (float64, bool) {
|
||||
switch {
|
||||
case m.Gauge != nil && m.Gauge.Value != nil:
|
||||
return *m.Gauge.Value, true
|
||||
case m.Counter != nil && m.Counter.Value != nil:
|
||||
return *m.Counter.Value, true
|
||||
case m.Untyped != nil && m.Untyped.Value != nil:
|
||||
return *m.Untyped.Value, true
|
||||
default:
|
||||
return 0, false
|
||||
}
|
||||
}
|
||||
|
||||
// ruleKeyFor macht die Drosselung unabhaengig je Regel UND je konkreter
|
||||
// Zeitreihe (z. B. verschiedene Tenants/Module derselben Metrik loesen
|
||||
// unabhaengig voneinander aus, siehe Migrationskommentar).
|
||||
func ruleKeyFor(ruleID string, labels map[string]string) string {
|
||||
keys := make([]string, 0, len(labels))
|
||||
for k := range labels {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
var b strings.Builder
|
||||
b.WriteString(ruleID)
|
||||
for _, k := range keys {
|
||||
b.WriteString("|")
|
||||
b.WriteString(k)
|
||||
b.WriteString("=")
|
||||
b.WriteString(labels[k])
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
@@ -1,233 +0,0 @@
|
||||
package alerting
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*RuleStore, *notify.Dispatcher, *pgxpool.Pool, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS alert_rules (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), metric_name TEXT NOT NULL,
|
||||
comparison TEXT NOT NULL CHECK (comparison IN ('gt','lt')), threshold DOUBLE PRECISION NOT NULL,
|
||||
label_filters JSONB NOT NULL DEFAULT '{}'::jsonb, recipient TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '', created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS alert_debounce_state (
|
||||
rule_key TEXT PRIMARY KEY, last_fired_at TIMESTAMPTZ NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS notification_jobs (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), channel TEXT NOT NULL, recipient TEXT NOT NULL,
|
||||
payload JSONB NOT NULL DEFAULT '{}'::jsonb, status TEXT NOT NULL DEFAULT 'pending', attempts INT NOT NULL DEFAULT 0,
|
||||
max_attempts INT NOT NULL DEFAULT 5, next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(), last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM alert_rules`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM alert_debounce_state`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM notification_jobs`)
|
||||
pool.Close()
|
||||
}
|
||||
return NewRuleStore(pool), notify.NewDispatcher(pool), pool, cleanup
|
||||
}
|
||||
|
||||
func gaugeFamily(name string, labels map[string]string, value float64) *dto.MetricFamily {
|
||||
pairs := make([]*dto.LabelPair, 0, len(labels))
|
||||
for k, v := range labels {
|
||||
k, v := k, v
|
||||
pairs = append(pairs, &dto.LabelPair{Name: &k, Value: &v})
|
||||
}
|
||||
n := name
|
||||
return &dto.MetricFamily{
|
||||
Name: &n,
|
||||
Metric: []*dto.Metric{
|
||||
{Label: pairs, Gauge: &dto.Gauge{Value: &value}},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + 2 / Pruefung 1 + 2: Überschreitung löst eine
|
||||
// Benachrichtigung mit vollständigem Inhalt (Metrik/Tenant/Modul) aus.
|
||||
func TestEvaluate_FiresAlertOnThresholdExceeded(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
rule, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example", Description: "Fehlerrate zu hoch",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
gaugeFamily("nexarch_core_error_rate", map[string]string{"tenant": "acme", "module": "dms"}, 0.12),
|
||||
}
|
||||
|
||||
fired, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("evaluate: %v", err)
|
||||
}
|
||||
if len(fired) != 1 || fired[0].Skipped {
|
||||
t.Fatalf("erwartet genau 1 tatsaechlich ausgeloesten alarm, habe %+v", fired)
|
||||
}
|
||||
if fired[0].RuleID != rule.ID {
|
||||
t.Fatalf("rule id = %q, want %q", fired[0].RuleID, rule.ID)
|
||||
}
|
||||
|
||||
// Pruefung 2: Benachrichtigungsinhalt vollstaendig (Metrik/Tenant/Modul).
|
||||
var payloadJSON []byte
|
||||
if err := pool.QueryRow(ctx, `SELECT payload FROM notification_jobs LIMIT 1`).Scan(&payloadJSON); err != nil {
|
||||
t.Fatalf("notification_jobs lesen: %v", err)
|
||||
}
|
||||
payload := string(payloadJSON)
|
||||
for _, want := range []string{`"metric"`, `nexarch_core_error_rate`, `"tenant"`, `"acme"`, `"module"`, `"dms"`} {
|
||||
if !strings.Contains(payload, want) {
|
||||
t.Errorf("payload enthaelt nicht %q: %s", want, payload)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestEvaluate_DoesNotFireBelowThreshold(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example",
|
||||
}); err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
gaugeFamily("nexarch_core_error_rate", map[string]string{"tenant": "acme"}, 0.01),
|
||||
}
|
||||
|
||||
fired, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("evaluate: %v", err)
|
||||
}
|
||||
if len(fired) != 0 {
|
||||
t.Fatalf("erwartet keinen alarm unterhalb des schwellwerts, habe %+v", fired)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 / Pruefung 3: anhaltende Überschreitung erzeugt NICHT
|
||||
// bei jeder Messung eine neue Benachrichtigung.
|
||||
func TestEvaluate_DebouncesRepeatedFiring(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example",
|
||||
}); err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
// Langes Debounce-Intervall: der zweite Evaluate-Lauf (simuliert die
|
||||
// naechste Messung bei anhaltend ueberschrittenem Wert) darf keinen
|
||||
// weiteren Job einreihen.
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
gaugeFamily("nexarch_core_error_rate", map[string]string{"tenant": "acme"}, 0.5),
|
||||
}
|
||||
|
||||
first, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("erster evaluate-lauf: %v", err)
|
||||
}
|
||||
if len(first) != 1 || first[0].Skipped {
|
||||
t.Fatalf("erster lauf haette feuern muessen, habe %+v", first)
|
||||
}
|
||||
|
||||
second, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("zweiter evaluate-lauf: %v", err)
|
||||
}
|
||||
if len(second) != 1 || !second[0].Skipped {
|
||||
t.Fatalf("zweiter lauf haette gedrosselt werden muessen, habe %+v", second)
|
||||
}
|
||||
|
||||
var count int
|
||||
if err := pool.QueryRow(ctx, `SELECT count(*) FROM notification_jobs`).Scan(&count); err != nil {
|
||||
t.Fatalf("notification_jobs zaehlen: %v", err)
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("erwartet genau 1 eingereihten job trotz zwei ueberschreitenden messungen, habe %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
// Verschiedene Zeitreihen derselben Regel (unterschiedlicher Tenant) werden
|
||||
// unabhaengig voneinander gedrosselt.
|
||||
func TestEvaluate_DebouncesIndependentlyPerLabelSet(t *testing.T) {
|
||||
rules, dispatcher, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := rules.CreateRule(ctx, Rule{
|
||||
MetricName: "nexarch_core_error_rate", Comparison: ComparisonGreaterThan, Threshold: 0.05,
|
||||
Recipient: "ops@acme.example",
|
||||
}); err != nil {
|
||||
t.Fatalf("create rule: %v", err)
|
||||
}
|
||||
|
||||
eval := NewEvaluator(rules, dispatcher, pool, time.Hour)
|
||||
families := []*dto.MetricFamily{
|
||||
{
|
||||
Name: strPtr("nexarch_core_error_rate"),
|
||||
Metric: []*dto.Metric{
|
||||
metricWithLabel("tenant", "acme", 0.5),
|
||||
metricWithLabel("tenant", "beta", 0.6),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
fired, err := eval.Evaluate(ctx, families)
|
||||
if err != nil {
|
||||
t.Fatalf("evaluate: %v", err)
|
||||
}
|
||||
if len(fired) != 2 {
|
||||
t.Fatalf("erwartet 2 unabhaengige alarme (verschiedene tenants), habe %d", len(fired))
|
||||
}
|
||||
for _, a := range fired {
|
||||
if a.Skipped {
|
||||
t.Fatalf("beide tenants sollten beim ersten mal feuern, habe %+v", a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func metricWithLabel(name, value string, gaugeValue float64) *dto.Metric {
|
||||
n, v := name, value
|
||||
return &dto.Metric{Label: []*dto.LabelPair{{Name: &n, Value: &v}}, Gauge: &dto.Gauge{Value: &gaugeValue}}
|
||||
}
|
||||
|
||||
func strPtr(s string) *string { return &s }
|
||||
@@ -1,132 +0,0 @@
|
||||
// Package alerting implementiert Core OPS-05: schwellwertbasierte
|
||||
// Alarmierung auf den aus OPS-03 aggregierten Metriken, Zustellung über den
|
||||
// Core-Benachrichtigungs-Dispatcher (CFG-02). Der Alertmanager-Gedanke von
|
||||
// Prometheus/Grafana, aber auf das Nötigste reduziert (Schwellwert, Ziel,
|
||||
// Drosselung) — keine eigene Ausdruckssprache.
|
||||
package alerting
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// Comparison legt fest, ob ein Schwellwert nach oben oder unten überwacht
|
||||
// wird — bewusst nur zwei Operatoren, keine eigene Ausdruckssprache
|
||||
// (Ticket-Produkt-DNA).
|
||||
type Comparison string
|
||||
|
||||
const (
|
||||
ComparisonGreaterThan Comparison = "gt"
|
||||
ComparisonLessThan Comparison = "lt"
|
||||
)
|
||||
|
||||
// Rule ist eine Schwellwert-Regel auf einer beliebigen aggregierten Metrik
|
||||
// (Akzeptanzkriterium 1). LabelFilters schränkt optional auf bestimmte
|
||||
// Label-Werte ein (z. B. tenant/module), leer = alle Zeitreihen der Metrik.
|
||||
type Rule struct {
|
||||
ID string
|
||||
MetricName string
|
||||
Comparison Comparison
|
||||
Threshold float64
|
||||
LabelFilters map[string]string
|
||||
Recipient string
|
||||
Description string
|
||||
}
|
||||
|
||||
// RuleStore verwaltet Alert-Regeln in der zentralen Registry-DB.
|
||||
type RuleStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewRuleStore(pool *pgxpool.Pool) *RuleStore {
|
||||
return &RuleStore{pool: pool}
|
||||
}
|
||||
|
||||
// CreateRule legt eine neue Schwellwert-Regel an (Akzeptanzkriterium 1).
|
||||
func (s *RuleStore) CreateRule(ctx context.Context, r Rule) (Rule, error) {
|
||||
if r.MetricName == "" || r.Recipient == "" {
|
||||
return Rule{}, errors.New("alerting: metricName und recipient duerfen nicht leer sein")
|
||||
}
|
||||
if r.Comparison != ComparisonGreaterThan && r.Comparison != ComparisonLessThan {
|
||||
return Rule{}, fmt.Errorf("alerting: unbekannter comparison-operator %q", r.Comparison)
|
||||
}
|
||||
if r.LabelFilters == nil {
|
||||
r.LabelFilters = map[string]string{}
|
||||
}
|
||||
filtersJSON, err := json.Marshal(r.LabelFilters)
|
||||
if err != nil {
|
||||
return Rule{}, fmt.Errorf("label-filter serialisieren: %w", err)
|
||||
}
|
||||
|
||||
err = s.pool.QueryRow(ctx, `
|
||||
INSERT INTO alert_rules (metric_name, comparison, threshold, label_filters, recipient, description)
|
||||
VALUES ($1, $2, $3, $4, $5, $6)
|
||||
RETURNING id
|
||||
`, r.MetricName, string(r.Comparison), r.Threshold, filtersJSON, r.Recipient, r.Description).Scan(&r.ID)
|
||||
if err != nil {
|
||||
return Rule{}, fmt.Errorf("regel speichern: %w", err)
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
// ListRules liefert alle konfigurierten Regeln — Grundlage für Evaluate.
|
||||
func (s *RuleStore) ListRules(ctx context.Context) ([]Rule, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, metric_name, comparison, threshold, label_filters, recipient, description
|
||||
FROM alert_rules ORDER BY created_at
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("regeln auflisten: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []Rule
|
||||
for rows.Next() {
|
||||
var r Rule
|
||||
var comparison string
|
||||
var filtersJSON []byte
|
||||
if err := rows.Scan(&r.ID, &r.MetricName, &comparison, &r.Threshold, &filtersJSON, &r.Recipient, &r.Description); err != nil {
|
||||
return nil, fmt.Errorf("regel lesen: %w", err)
|
||||
}
|
||||
r.Comparison = Comparison(comparison)
|
||||
if err := json.Unmarshal(filtersJSON, &r.LabelFilters); err != nil {
|
||||
return nil, fmt.Errorf("label-filter lesen: %w", err)
|
||||
}
|
||||
out = append(out, r)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// DeleteRule entfernt eine Regel.
|
||||
func (s *RuleStore) DeleteRule(ctx context.Context, id string) error {
|
||||
_, err := s.pool.Exec(ctx, `DELETE FROM alert_rules WHERE id = $1`, id)
|
||||
if err != nil {
|
||||
return fmt.Errorf("regel loeschen: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// debounceStore kapselt die Drosselungs-Zustandstabelle (Akzeptanzkriterium 3).
|
||||
type debounceStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
// shouldFire prueft, ob seit dem letzten Alarm fuer ruleKey mindestens
|
||||
// interval vergangen ist — atomar ueber eine bedingte UPDATE/INSERT-
|
||||
// Sequenz, damit zwei gleichzeitige Evaluate-Laeufe (z. B. bei mehreren
|
||||
// Core-Instanzen) nicht beide gleichzeitig alarmieren.
|
||||
func (d *debounceStore) shouldFire(ctx context.Context, ruleKey string, intervalSeconds float64) (bool, error) {
|
||||
tag, err := d.pool.Exec(ctx, `
|
||||
INSERT INTO alert_debounce_state (rule_key, last_fired_at) VALUES ($1, now())
|
||||
ON CONFLICT (rule_key) DO UPDATE SET last_fired_at = now()
|
||||
WHERE alert_debounce_state.last_fired_at <= now() - ($2 * interval '1 second')
|
||||
`, ruleKey, intervalSeconds)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("drosselungszustand pruefen: %w", err)
|
||||
}
|
||||
return tag.RowsAffected() == 1, nil
|
||||
}
|
||||
@@ -1,88 +0,0 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DefaultCacheTTL ist die dokumentierte Cache-Invalidierungszeit
|
||||
// (Akzeptanzkriterium 2 / Pruefung 2 in diesem Ticket bezieht sich auf die
|
||||
// Aenderungsnachvollziehbarkeit — die Cache-Frist selbst folgt demselben
|
||||
// Muster wie internal/flag.DefaultCacheTTL).
|
||||
const DefaultCacheTTL = 5 * time.Second
|
||||
|
||||
type cacheEntry struct {
|
||||
value Value
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
// Service ist die Leseseite mit Vorrangregel (Akzeptanzkriterium 1:
|
||||
// Tenant-Override vor Global-Default) und lokalem TTL-Cache.
|
||||
type Service struct {
|
||||
store *Store
|
||||
ttl time.Duration
|
||||
|
||||
mu sync.RWMutex
|
||||
cache map[string]cacheEntry // Schluessel: key + "\x00" + tenantSlug
|
||||
}
|
||||
|
||||
func NewService(store *Store, ttl time.Duration) *Service {
|
||||
if ttl <= 0 {
|
||||
ttl = DefaultCacheTTL
|
||||
}
|
||||
return &Service{store: store, ttl: ttl, cache: make(map[string]cacheEntry)}
|
||||
}
|
||||
|
||||
func cacheKey(key, tenantSlug string) string {
|
||||
return key + "\x00" + tenantSlug
|
||||
}
|
||||
|
||||
// Resolve liefert den Konfigurationswert fuer einen Tenant: ein
|
||||
// Tenant-spezifischer Override hat Vorrang vor dem globalen Default
|
||||
// (Akzeptanzkriterium 1 / Pruefung 1). tenantSlug == "" wertet nur den
|
||||
// globalen Wert aus.
|
||||
func (s *Service) Resolve(ctx context.Context, tenantSlug, key string) (Value, error) {
|
||||
ck := cacheKey(key, tenantSlug)
|
||||
|
||||
s.mu.RLock()
|
||||
entry, exists := s.cache[ck]
|
||||
fresh := exists && time.Now().Before(entry.expiresAt)
|
||||
s.mu.RUnlock()
|
||||
if fresh {
|
||||
return entry.value, nil
|
||||
}
|
||||
|
||||
v, err := s.resolveUncached(ctx, tenantSlug, key)
|
||||
if err != nil {
|
||||
return Value{}, err
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.cache[ck] = cacheEntry{value: v, expiresAt: time.Now().Add(s.ttl)}
|
||||
s.mu.Unlock()
|
||||
return v, nil
|
||||
}
|
||||
|
||||
func (s *Service) resolveUncached(ctx context.Context, tenantSlug, key string) (Value, error) {
|
||||
if tenantSlug != "" {
|
||||
v, err := s.store.Get(ctx, key, tenantSlug)
|
||||
if err == nil {
|
||||
return v, nil
|
||||
}
|
||||
if !errors.Is(err, ErrNotFound) {
|
||||
return Value{}, err
|
||||
}
|
||||
}
|
||||
return s.store.Get(ctx, key, GlobalScope)
|
||||
}
|
||||
|
||||
// Invalidate erzwingt beim naechsten Resolve-Aufruf ein sofortiges Neuladen
|
||||
// fuer einen bestimmten (key, tenantSlug) statt auf den TTL-Ablauf zu warten
|
||||
// — analog internal/flag.Service.Invalidate.
|
||||
func (s *Service) Invalidate(key, tenantSlug string) {
|
||||
s.mu.Lock()
|
||||
delete(s.cache, cacheKey(key, tenantSlug))
|
||||
s.mu.Unlock()
|
||||
}
|
||||
@@ -1,114 +0,0 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: Tenant-Override hat Vorrang vor
|
||||
// Global-Default, automatisiert getestet.
|
||||
func TestService_TenantOverrideTakesPrecedenceOverGlobal(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Set(ctx, "test_precedence_key", GlobalScope, "global-wert"); err != nil {
|
||||
t.Fatalf("set global: %v", err)
|
||||
}
|
||||
if _, err := store.Set(ctx, "test_precedence_key", "test_acme", "tenant-wert"); err != nil {
|
||||
t.Fatalf("set tenant: %v", err)
|
||||
}
|
||||
|
||||
svc := NewService(store, time.Hour)
|
||||
|
||||
got, err := svc.Resolve(ctx, "test_acme", "test_precedence_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve mit override: %v", err)
|
||||
}
|
||||
if got.Value != "tenant-wert" {
|
||||
t.Fatalf("erwartet tenant-override, habe %q", got.Value)
|
||||
}
|
||||
|
||||
gotOther, err := svc.Resolve(ctx, "test_anderer_tenant", "test_precedence_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve ohne override: %v", err)
|
||||
}
|
||||
if gotOther.Value != "global-wert" {
|
||||
t.Fatalf("erwartet global-default fuer tenant ohne override, habe %q", gotOther.Value)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Cache-Invalidierung nach
|
||||
// Konfigurationsaenderung innerhalb dokumentierter Zeit gemessen.
|
||||
func TestService_CacheInvalidationTiming(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
const ttl = 150 * time.Millisecond
|
||||
if _, err := store.Set(ctx, "test_ttl_key", GlobalScope, "alt"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, ttl)
|
||||
|
||||
v, err := svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value != "alt" {
|
||||
t.Fatalf("erwartet 'alt', habe %q", v.Value)
|
||||
}
|
||||
|
||||
changedAt := time.Now()
|
||||
if _, err := store.Set(ctx, "test_ttl_key", GlobalScope, "neu"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
|
||||
v, err = svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve direkt nach aenderung: %v", err)
|
||||
}
|
||||
if v.Value != "alt" {
|
||||
t.Fatalf("cache haette den alten wert liefern sollen, habe %q", v.Value)
|
||||
}
|
||||
|
||||
deadline := changedAt.Add(ttl + 100*time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
v, err := svc.Resolve(ctx, "", "test_ttl_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value == "neu" {
|
||||
t.Logf("aenderung wurde nach %s wirksam (ziel: innerhalb %s + toleranz)", time.Since(changedAt), ttl)
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("aenderung wurde nicht innerhalb von %s wirksam", deadline.Sub(changedAt))
|
||||
}
|
||||
|
||||
func TestService_InvalidateForcesImmediateRefresh(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Set(ctx, "test_invalidate_key", GlobalScope, "alt"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, time.Hour)
|
||||
_, _ = svc.Resolve(ctx, "", "test_invalidate_key")
|
||||
|
||||
if _, err := store.Set(ctx, "test_invalidate_key", GlobalScope, "neu"); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc.Invalidate("test_invalidate_key", "")
|
||||
|
||||
v, err := svc.Resolve(ctx, "", "test_invalidate_key")
|
||||
if err != nil {
|
||||
t.Fatalf("resolve: %v", err)
|
||||
}
|
||||
if v.Value != "neu" {
|
||||
t.Fatalf("erwartet sofort sichtbaren neuen wert nach Invalidate, habe %q", v.Value)
|
||||
}
|
||||
}
|
||||
@@ -1,129 +0,0 @@
|
||||
// Package cfgservice implementiert Core CFG-01: den zentralen Dienst fuer
|
||||
// globale und tenant-spezifische Konfigurationswerte mit Versionierung und
|
||||
// Cache-Invalidierung. Andere Module lesen Konfiguration AUSSCHLIESSLICH
|
||||
// ueber dieses Paket (Akzeptanzkriterium 3), niemals ueber eigene Tabellen.
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// GlobalScope ist der reservierte Scope-Wert fuer globale Defaults — jeder
|
||||
// andere Scope-Wert ist ein Tenant-Slug (Akzeptanzkriterium 1).
|
||||
const GlobalScope = "global"
|
||||
|
||||
var ErrNotFound = errors.New("cfgservice: kein wert fuer diesen key gefunden")
|
||||
|
||||
type Value struct {
|
||||
Key string
|
||||
Scope string
|
||||
Value string
|
||||
Version int
|
||||
}
|
||||
|
||||
type HistoryEntry struct {
|
||||
Key string
|
||||
Scope string
|
||||
Value string
|
||||
Version int
|
||||
}
|
||||
|
||||
// Store ist die Schreib-/Verwaltungsseite. Set schreibt IMMER sowohl den
|
||||
// aktuellen Stand (config_values) als auch einen Historieneintrag
|
||||
// (config_value_history) in derselben Transaktion — eine Aenderung ohne
|
||||
// Versionshistorie ist strukturell ausgeschlossen (Akzeptanzkriterium 2).
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// Set schreibt einen neuen Wert fuer (key, scope) und erhoeht die Version um 1
|
||||
// (Version 1 bei erstmaligem Setzen).
|
||||
func (s *Store) Set(ctx context.Context, key, scope, value string) (Value, error) {
|
||||
if scope == "" {
|
||||
return Value{}, errors.New("cfgservice: scope darf nicht leer sein")
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return Value{}, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
var currentVersion int
|
||||
err = tx.QueryRow(ctx, `SELECT version FROM config_values WHERE key = $1 AND scope = $2`, key, scope).Scan(¤tVersion)
|
||||
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
|
||||
return Value{}, fmt.Errorf("aktuelle version lesen: %w", err)
|
||||
}
|
||||
newVersion := currentVersion + 1
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO config_values (key, scope, value, version, updated_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
ON CONFLICT (key, scope) DO UPDATE SET value = $3, version = $4, updated_at = now()
|
||||
`, key, scope, value, newVersion); err != nil {
|
||||
return Value{}, fmt.Errorf("wert speichern: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO config_value_history (key, scope, value, version, changed_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
`, key, scope, value, newVersion); err != nil {
|
||||
return Value{}, fmt.Errorf("historie schreiben: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return Value{}, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
|
||||
return Value{Key: key, Scope: scope, Value: value, Version: newVersion}, nil
|
||||
}
|
||||
|
||||
// Get liefert den Wert fuer GENAU EINEN Scope (kein Vorrang-Fallback) — die
|
||||
// Vorrangregel (Tenant vor Global) lebt bewusst in Service.Resolve, damit
|
||||
// Store rein CRUD bleibt.
|
||||
func (s *Store) Get(ctx context.Context, key, scope string) (Value, error) {
|
||||
var v Value
|
||||
v.Key, v.Scope = key, scope
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT value, version FROM config_values WHERE key = $1 AND scope = $2
|
||||
`, key, scope).Scan(&v.Value, &v.Version)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Value{}, ErrNotFound
|
||||
}
|
||||
return Value{}, fmt.Errorf("wert lesen: %w", err)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// History liefert die vollstaendige Versionshistorie eines (key, scope) in
|
||||
// aufsteigender Reihenfolge (Akzeptanzkriterium 2 / Pruefung 3).
|
||||
func (s *Store) History(ctx context.Context, key, scope string) ([]HistoryEntry, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT key, scope, value, version FROM config_value_history
|
||||
WHERE key = $1 AND scope = $2 ORDER BY version
|
||||
`, key, scope)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("historie abfragen: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []HistoryEntry
|
||||
for rows.Next() {
|
||||
var h HistoryEntry
|
||||
if err := rows.Scan(&h.Key, &h.Scope, &h.Value, &h.Version); err != nil {
|
||||
return nil, fmt.Errorf("historieneintrag lesen: %w", err)
|
||||
}
|
||||
out = append(out, h)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -1,119 +0,0 @@
|
||||
package cfgservice
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupStoreTest(t *testing.T) (*Store, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS config_values (
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL CHECK (scope <> ''),
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (key, scope)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS config_value_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM config_value_history WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM config_values WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
||||
pool.Close()
|
||||
}
|
||||
return NewStore(pool), cleanup
|
||||
}
|
||||
|
||||
func TestStore_SetIncrementsVersion(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
v1, err := store.Set(ctx, "test_key", GlobalScope, "erster-wert")
|
||||
if err != nil {
|
||||
t.Fatalf("set 1: %v", err)
|
||||
}
|
||||
if v1.Version != 1 {
|
||||
t.Fatalf("erwartet version 1, habe %d", v1.Version)
|
||||
}
|
||||
|
||||
v2, err := store.Set(ctx, "test_key", GlobalScope, "zweiter-wert")
|
||||
if err != nil {
|
||||
t.Fatalf("set 2: %v", err)
|
||||
}
|
||||
if v2.Version != 2 {
|
||||
t.Fatalf("erwartet version 2, habe %d", v2.Version)
|
||||
}
|
||||
|
||||
got, err := store.Get(ctx, "test_key", GlobalScope)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.Value != "zweiter-wert" || got.Version != 2 {
|
||||
t.Fatalf("aktueller wert unerwartet: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 3: Versionierungshistorie ueber mehrere
|
||||
// Aenderungen hinweg nachvollzogen.
|
||||
func TestStore_HistoryTracksAllChanges(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
values := []string{"v1", "v2", "v3"}
|
||||
for _, v := range values {
|
||||
if _, err := store.Set(ctx, "test_history_key", GlobalScope, v); err != nil {
|
||||
t.Fatalf("set %q: %v", v, err)
|
||||
}
|
||||
}
|
||||
|
||||
history, err := store.History(ctx, "test_history_key", GlobalScope)
|
||||
if err != nil {
|
||||
t.Fatalf("history: %v", err)
|
||||
}
|
||||
if len(history) != 3 {
|
||||
t.Fatalf("erwartet 3 historieneintraege, habe %d", len(history))
|
||||
}
|
||||
for i, h := range history {
|
||||
if h.Version != i+1 || h.Value != values[i] {
|
||||
t.Fatalf("historieneintrag[%d] unerwartet: %+v", i, h)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestStore_GetUnknownKeyReturnsNotFound(t *testing.T) {
|
||||
store, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Get(ctx, "test_nie_gesetzt", GlobalScope); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("erwartet ErrNotFound, habe %v", err)
|
||||
}
|
||||
}
|
||||
@@ -1,87 +0,0 @@
|
||||
// Package flag implementiert Core LIC-02: einen Feature-Flag-Dienst mit
|
||||
// Strategien (global an/aus, Prozentsatz, Tenant-Zielgruppe) als Kernfunktion
|
||||
// des Core-Dienstes selbst — keine zusaetzliche Infrastruktur (Unleash-Server
|
||||
// + eigene DB), siehe "bewusst vermeiden" im LIC-02-Ticket.
|
||||
package flag
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"hash/fnv"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var ErrNotFound = errors.New("flag: nicht gefunden")
|
||||
|
||||
// Flag ist die zentrale Definition — Auswertung (Evaluate) ist bewusst davon
|
||||
// getrennt (Unleash-Prinzip: Flag-Verwaltung vs. Flag-Auswertung).
|
||||
type Flag struct {
|
||||
Key string
|
||||
Enabled bool
|
||||
RolloutPercentage int
|
||||
TargetTenantSlugs []string
|
||||
}
|
||||
|
||||
// Store ist die Verwaltungsseite (Admin): Flags definieren/lesen.
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
func (s *Store) Set(ctx context.Context, f Flag) error {
|
||||
if f.TargetTenantSlugs == nil {
|
||||
f.TargetTenantSlugs = []string{} // pgx uebertraegt ein nil-Slice sonst als SQL NULL statt leerem Array.
|
||||
}
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO feature_flags (key, enabled, rollout_percentage, target_tenant_slugs, updated_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
ON CONFLICT (key) DO UPDATE SET
|
||||
enabled = $2, rollout_percentage = $3, target_tenant_slugs = $4, updated_at = now()
|
||||
`, f.Key, f.Enabled, f.RolloutPercentage, f.TargetTenantSlugs)
|
||||
if err != nil {
|
||||
return fmt.Errorf("flag speichern: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Store) Get(ctx context.Context, key string) (Flag, error) {
|
||||
var f Flag
|
||||
row := s.pool.QueryRow(ctx, `
|
||||
SELECT key, enabled, rollout_percentage, target_tenant_slugs
|
||||
FROM feature_flags WHERE key = $1
|
||||
`, key)
|
||||
if err := row.Scan(&f.Key, &f.Enabled, &f.RolloutPercentage, &f.TargetTenantSlugs); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Flag{}, ErrNotFound
|
||||
}
|
||||
return Flag{}, fmt.Errorf("flag lesen: %w", err)
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// evaluate wendet die Strategien in fester Reihenfolge an: globaler
|
||||
// An/Aus-Schalter zuerst, dann Tenant-Zielgruppe, dann Prozentsatz-Rollout.
|
||||
// Ein unbekannter/nicht getroffener Fall ergibt false — Fail-Safe-Default,
|
||||
// kein Feature wird versehentlich aktiv.
|
||||
func evaluate(f Flag, tenantSlug string) bool {
|
||||
if f.Enabled {
|
||||
return true
|
||||
}
|
||||
for _, target := range f.TargetTenantSlugs {
|
||||
if target == tenantSlug {
|
||||
return true
|
||||
}
|
||||
}
|
||||
if f.RolloutPercentage > 0 {
|
||||
h := fnv.New32a()
|
||||
_, _ = h.Write([]byte(f.Key + "|" + tenantSlug))
|
||||
return int(h.Sum32()%100) < f.RolloutPercentage
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -1,43 +0,0 @@
|
||||
package flag
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestEvaluate_GlobalEnabled(t *testing.T) {
|
||||
f := Flag{Key: "k", Enabled: true}
|
||||
if !evaluate(f, "irgendein-tenant") {
|
||||
t.Fatal("global aktiviertes flag sollte fuer jeden tenant true liefern")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie.
|
||||
func TestEvaluate_TargetTenantStrategy(t *testing.T) {
|
||||
f := Flag{Key: "k", Enabled: false, TargetTenantSlugs: []string{"acme"}}
|
||||
if !evaluate(f, "acme") {
|
||||
t.Fatal("erwartet true fuer tenant in zielgruppe")
|
||||
}
|
||||
if evaluate(f, "globex") {
|
||||
t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
|
||||
}
|
||||
}
|
||||
|
||||
func TestEvaluate_RolloutPercentageBoundaries(t *testing.T) {
|
||||
full := Flag{Key: "k", RolloutPercentage: 100}
|
||||
if !evaluate(full, "beliebiger-tenant-1") || !evaluate(full, "beliebiger-tenant-2") {
|
||||
t.Fatal("100% rollout sollte immer true liefern")
|
||||
}
|
||||
|
||||
none := Flag{Key: "k", RolloutPercentage: 0}
|
||||
if evaluate(none, "beliebiger-tenant") {
|
||||
t.Fatal("0% rollout ohne enabled/zielgruppe sollte false liefern")
|
||||
}
|
||||
}
|
||||
|
||||
func TestEvaluate_RolloutIsDeterministicPerTenant(t *testing.T) {
|
||||
f := Flag{Key: "k", RolloutPercentage: 50}
|
||||
first := evaluate(f, "stabiler-tenant")
|
||||
for i := 0; i < 5; i++ {
|
||||
if evaluate(f, "stabiler-tenant") != first {
|
||||
t.Fatal("rollout-auswertung sollte fuer denselben tenant/key stabil sein")
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,87 +0,0 @@
|
||||
package flag
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// DefaultCacheTTL ist die dokumentierte Cache-Invalidierungszeit
|
||||
// (Akzeptanzkriterium 2/3): eine Aenderung wirkt spaetestens nach dieser
|
||||
// Zeit auf allen Core-Instanzen, ohne dass ein Dienst neu gestartet werden
|
||||
// muss (Akzeptanzkriterium 3).
|
||||
const DefaultCacheTTL = 5 * time.Second
|
||||
|
||||
type cacheEntry struct {
|
||||
flag Flag
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
// Service ist die Auswertungsseite (SDK/Client-Analogon zu Unleash) mit
|
||||
// lokalem TTL-Cache. Bewusst getrennt von Store (Verwaltung).
|
||||
type Service struct {
|
||||
store *Store
|
||||
ttl time.Duration
|
||||
|
||||
mu sync.RWMutex
|
||||
cache map[string]cacheEntry
|
||||
}
|
||||
|
||||
func NewService(store *Store, ttl time.Duration) *Service {
|
||||
if ttl <= 0 {
|
||||
ttl = DefaultCacheTTL
|
||||
}
|
||||
return &Service{store: store, ttl: ttl, cache: make(map[string]cacheEntry)}
|
||||
}
|
||||
|
||||
// IsEnabled wertet ein Flag fuer einen Tenant aus. Liefert IMMER einen
|
||||
// bool ohne Fehlerwert — ein nicht erreichbarer Flag-Dienst darf abhaengige
|
||||
// Aufrufer nicht zum Absturz bringen oder zu Fehlerbehandlungscode zwingen,
|
||||
// der leicht vergessen wird (Akzeptanzkriterium 3 / Pruefung 3: dokumentiertes
|
||||
// Fallback-Verhalten = false, ggf. aus dem zuletzt bekannten Zwischenspeicher).
|
||||
func (s *Service) IsEnabled(ctx context.Context, tenantSlug, key string) bool {
|
||||
f, ok := s.resolve(ctx, key)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
return evaluate(f, tenantSlug)
|
||||
}
|
||||
|
||||
func (s *Service) resolve(ctx context.Context, key string) (Flag, bool) {
|
||||
s.mu.RLock()
|
||||
entry, exists := s.cache[key]
|
||||
fresh := exists && time.Now().Before(entry.expiresAt)
|
||||
s.mu.RUnlock()
|
||||
if fresh {
|
||||
return entry.flag, true
|
||||
}
|
||||
|
||||
f, err := s.store.Get(ctx, key)
|
||||
if err != nil {
|
||||
if exists {
|
||||
slog.Warn("feature-flag-dienst nicht erreichbar, nutze zwischengespeicherten stand",
|
||||
"flag_key", key, "error", err)
|
||||
return entry.flag, true
|
||||
}
|
||||
slog.Warn("feature-flag-dienst nicht erreichbar, kein zwischengespeicherter stand vorhanden, fallback: deaktiviert",
|
||||
"flag_key", key, "error", err)
|
||||
return Flag{}, false
|
||||
}
|
||||
|
||||
s.mu.Lock()
|
||||
s.cache[key] = cacheEntry{flag: f, expiresAt: time.Now().Add(s.ttl)}
|
||||
s.mu.Unlock()
|
||||
return f, true
|
||||
}
|
||||
|
||||
// Invalidate erzwingt beim naechsten IsEnabled-Aufruf ein sofortiges Neuladen
|
||||
// aus der Datenbank statt auf den TTL-Ablauf zu warten — wird nach Store.Set
|
||||
// auf derselben Instanz aufgerufen, damit der Schreiber die eigene Aenderung
|
||||
// ohne Wartezeit sieht. Andere Core-Instanzen sehen sie spaetestens nach
|
||||
// DefaultCacheTTL (siehe Akzeptanzkriterium 3).
|
||||
func (s *Service) Invalidate(key string) {
|
||||
s.mu.Lock()
|
||||
delete(s.cache, key)
|
||||
s.mu.Unlock()
|
||||
}
|
||||
@@ -1,179 +0,0 @@
|
||||
package flag
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupFlagStoreTest(t *testing.T) (*Store, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS feature_flags (
|
||||
key TEXT PRIMARY KEY,
|
||||
enabled BOOLEAN NOT NULL DEFAULT false,
|
||||
rollout_percentage INT NOT NULL DEFAULT 0,
|
||||
target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM feature_flags WHERE key LIKE 'test\_%' ESCAPE '\'`)
|
||||
pool.Close()
|
||||
}
|
||||
return NewStore(pool), cleanup
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 2: Zielgruppen-Strategie liefert im Test
|
||||
// die erwartete Auswertung.
|
||||
func TestService_TargetTenantStrategy(t *testing.T) {
|
||||
store, cleanup := setupFlagStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if err := store.Set(ctx, Flag{Key: "test_target_flag", TargetTenantSlugs: []string{"acme"}}); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, time.Hour)
|
||||
|
||||
if !svc.IsEnabled(ctx, "acme", "test_target_flag") {
|
||||
t.Fatal("erwartet true fuer tenant in zielgruppe")
|
||||
}
|
||||
if svc.IsEnabled(ctx, "globex", "test_target_flag") {
|
||||
t.Fatal("erwartet false fuer tenant ausserhalb der zielgruppe")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + 3 + Pruefung 1: Flag-Aenderung wirkt innerhalb der
|
||||
// dokumentierten Cache-Invalidierungszeit, automatisiert gemessen.
|
||||
func TestService_CacheInvalidationTiming(t *testing.T) {
|
||||
store, cleanup := setupFlagStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
const ttl = 150 * time.Millisecond
|
||||
if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: false}); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, ttl)
|
||||
|
||||
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
|
||||
t.Fatal("erwartet false vor der aenderung")
|
||||
}
|
||||
|
||||
// Aenderung "auf einer anderen instanz" simulieren: direkt ueber den
|
||||
// Store, ohne svc.Invalidate aufzurufen.
|
||||
changedAt := time.Now()
|
||||
if err := store.Set(ctx, Flag{Key: "test_ttl_flag", Enabled: true}); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
|
||||
// Sofort danach sollte der Cache noch den alten Stand liefern.
|
||||
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
|
||||
t.Fatal("cache haette den alten (false) stand liefern sollen, direkt nach der aenderung")
|
||||
}
|
||||
|
||||
deadline := changedAt.Add(ttl + 100*time.Millisecond)
|
||||
for time.Now().Before(deadline) {
|
||||
if svc.IsEnabled(ctx, "acme", "test_ttl_flag") {
|
||||
elapsed := time.Since(changedAt)
|
||||
t.Logf("aenderung wurde nach %s wirksam (ziel: innerhalb %s + toleranz)", elapsed, ttl)
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("aenderung wurde nicht innerhalb von %s wirksam", deadline.Sub(changedAt))
|
||||
}
|
||||
|
||||
func TestService_InvalidateForcesImmediateRefresh(t *testing.T) {
|
||||
store, cleanup := setupFlagStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: false}); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, time.Hour) // lange TTL, damit Invalidate den unterschied macht
|
||||
_ = svc.IsEnabled(ctx, "acme", "test_invalidate_flag")
|
||||
|
||||
if err := store.Set(ctx, Flag{Key: "test_invalidate_flag", Enabled: true}); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc.Invalidate("test_invalidate_flag")
|
||||
|
||||
if !svc.IsEnabled(ctx, "acme", "test_invalidate_flag") {
|
||||
t.Fatal("erwartet sofort sichtbaren neuen stand nach Invalidate")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 3: Ausfall des Flag-Dienstes fuehrt zu
|
||||
// dokumentiertem Fallback-Verhalten, nicht zum Absturz.
|
||||
func TestService_FallsBackOnStoreFailure(t *testing.T) {
|
||||
store, cleanup := setupFlagStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if err := store.Set(ctx, Flag{Key: "test_fallback_flag", Enabled: true}); err != nil {
|
||||
t.Fatalf("set: %v", err)
|
||||
}
|
||||
svc := NewService(store, time.Hour)
|
||||
|
||||
// Cache vorwaermen, waehrend die DB noch erreichbar ist.
|
||||
if !svc.IsEnabled(ctx, "acme", "test_fallback_flag") {
|
||||
t.Fatal("erwartet true bei funktionierender db")
|
||||
}
|
||||
|
||||
brokenPool, err := pgxpool.New(ctx, "postgresql://nonexistent-host-fuer-test:5432/x?connect_timeout=1")
|
||||
if err != nil {
|
||||
t.Fatalf("broken pool erstellen (sollte nicht sofort verbinden): %v", err)
|
||||
}
|
||||
brokenStore := NewStore(brokenPool)
|
||||
|
||||
svcWithCache := NewService(brokenStore, time.Nanosecond) // TTL sofort abgelaufen, erzwingt reload-versuch
|
||||
svcWithCache.mu.Lock()
|
||||
svcWithCache.cache["test_fallback_flag"] = cacheEntry{
|
||||
flag: Flag{Key: "test_fallback_flag", Enabled: true},
|
||||
expiresAt: time.Now().Add(-time.Hour), // bereits abgelaufen
|
||||
}
|
||||
svcWithCache.mu.Unlock()
|
||||
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
t.Fatalf("IsEnabled hat gepanict statt einen fallback zu liefern: %v", r)
|
||||
}
|
||||
}()
|
||||
if !svcWithCache.IsEnabled(ctx, "acme", "test_fallback_flag") {
|
||||
t.Fatal("erwartet fallback auf zwischengespeicherten (true) stand bei db-ausfall")
|
||||
}
|
||||
}()
|
||||
|
||||
// Voellig frischer Dienst ohne jeglichen cache + kaputte db -> sicherer
|
||||
// default false, kein absturz.
|
||||
freshSvc := NewService(brokenStore, time.Hour)
|
||||
func() {
|
||||
defer func() {
|
||||
if r := recover(); r != nil {
|
||||
t.Fatalf("IsEnabled hat gepanict: %v", r)
|
||||
}
|
||||
}()
|
||||
if freshSvc.IsEnabled(ctx, "acme", "test_fallback_flag") {
|
||||
t.Fatal("erwartet fail-safe false ohne cache und mit kaputter db")
|
||||
}
|
||||
}()
|
||||
}
|
||||
@@ -1,25 +0,0 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// DatabaseChecker prueft die tatsaechliche Erreichbarkeit der Datenbank
|
||||
// (Ping) — nicht nur, ob der Pool existiert.
|
||||
func DatabaseChecker(pool *pgxpool.Pool) CheckerFunc {
|
||||
return func(ctx context.Context) error {
|
||||
return pool.Ping(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
// QueueChecker prueft, dass die Postgres-basierte Job-Queue (siehe CFG-02)
|
||||
// tatsaechlich abfragbar ist — eine eigene, benannte Abhaengigkeit neben der
|
||||
// reinen DB-Erreichbarkeit (Akzeptanzkriterium 1).
|
||||
func QueueChecker(pool *pgxpool.Pool) CheckerFunc {
|
||||
return func(ctx context.Context) error {
|
||||
_, err := pool.Exec(ctx, `SELECT 1`)
|
||||
return err
|
||||
}
|
||||
}
|
||||
@@ -1,49 +0,0 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// LivenessHandler beantwortet IMMER "lebt", solange der Prozess ueberhaupt
|
||||
// HTTP-Anfragen verarbeiten kann — prueft bewusst KEINE externen
|
||||
// Abhaengigkeiten (Akzeptanzkriterium 2: Liveness und Readiness getrennt).
|
||||
// Ein Datenbankausfall darf die Liveness nicht auf "tot" setzen, sonst
|
||||
// wuerde eine Orchestrierung (z.B. systemd/Kubernetes) den Prozess grundlos
|
||||
// neu starten, obwohl nur eine Abhaengigkeit ausgefallen ist.
|
||||
func LivenessHandler(w http.ResponseWriter, r *http.Request) {
|
||||
writeStatus(w, http.StatusOK, map[string]any{"status": "alive"})
|
||||
}
|
||||
|
||||
// ReadinessHandler prueft ALLE registrierten Abhaengigkeiten
|
||||
// (Akzeptanzkriterium 1) und liefert 503, sobald eine davon fehlschlaegt
|
||||
// (Akzeptanzkriterium 3) — unterscheidet sich damit nachweislich von
|
||||
// LivenessHandler im Fehlerfall (Akzeptanzkriterium 2 / Pruefung 3).
|
||||
func (r *Registry) ReadinessHandler() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, req *http.Request) {
|
||||
ready, results := r.CheckAll(req.Context())
|
||||
|
||||
body := map[string]any{
|
||||
"status": statusText(ready),
|
||||
"checks": results,
|
||||
}
|
||||
status := http.StatusOK
|
||||
if !ready {
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
writeStatus(w, status, body)
|
||||
}
|
||||
}
|
||||
|
||||
func statusText(ready bool) string {
|
||||
if ready {
|
||||
return "ready"
|
||||
}
|
||||
return "not_ready"
|
||||
}
|
||||
|
||||
func writeStatus(w http.ResponseWriter, status int, body map[string]any) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(body)
|
||||
}
|
||||
@@ -1,86 +0,0 @@
|
||||
// Package health implementiert Core OPS-01: Health-/Readiness-Endpunkte, die
|
||||
// echte Abhaengigkeiten (DB, Job-Queue) statt nur den Prozessstatus pruefen
|
||||
// — wiederverwendbar von Core UND jedem registrierten Modul (siehe API-02),
|
||||
// nicht nur von Core selbst.
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Checker prueft EINE Abhaengigkeit (z.B. Datenbank, Job-Queue).
|
||||
type Checker interface {
|
||||
Check(ctx context.Context) error
|
||||
}
|
||||
|
||||
type CheckerFunc func(ctx context.Context) error
|
||||
|
||||
func (f CheckerFunc) Check(ctx context.Context) error { return f(ctx) }
|
||||
|
||||
// DefaultCheckTimeout begrenzt, wie lange EIN einzelner Check maximal
|
||||
// dauern darf, bevor er als fehlgeschlagen gilt — verhindert, dass ein
|
||||
// haengender Check den gesamten Readiness-Endpunkt blockiert
|
||||
// (Akzeptanzkriterium 2 / Pruefung 2: Antwort innerhalb definierter Zeit).
|
||||
const DefaultCheckTimeout = 2 * time.Second
|
||||
|
||||
// Registry haelt alle benannten Checks eines Dienstes.
|
||||
type Registry struct {
|
||||
checks map[string]Checker
|
||||
timeout time.Duration
|
||||
}
|
||||
|
||||
func NewRegistry() *Registry {
|
||||
return &Registry{checks: make(map[string]Checker), timeout: DefaultCheckTimeout}
|
||||
}
|
||||
|
||||
func (r *Registry) WithTimeout(d time.Duration) *Registry {
|
||||
return &Registry{checks: r.checks, timeout: d}
|
||||
}
|
||||
|
||||
// Register fuegt einen benannten Check hinzu (z.B. "database", "queue").
|
||||
func (r *Registry) Register(name string, c Checker) {
|
||||
r.checks[name] = c
|
||||
}
|
||||
|
||||
// Result ist der Ausgang eines einzelnen Checks.
|
||||
type Result struct {
|
||||
OK bool
|
||||
Error string
|
||||
}
|
||||
|
||||
// CheckAll fuehrt alle registrierten Checks NEBENLAEUFIG mit je eigenem
|
||||
// Timeout aus (Akzeptanzkriterium 1: echte Abhaengigkeiten statt Prozess-
|
||||
// status) und liefert ready=false, sobald irgendein Check fehlschlaegt
|
||||
// (Akzeptanzkriterium 3: ein Ausfall wird sichtbar).
|
||||
func (r *Registry) CheckAll(ctx context.Context) (ready bool, results map[string]Result) {
|
||||
type namedResult struct {
|
||||
name string
|
||||
result Result
|
||||
}
|
||||
ch := make(chan namedResult, len(r.checks))
|
||||
|
||||
for name, checker := range r.checks {
|
||||
go func(name string, checker Checker) {
|
||||
checkCtx, cancel := context.WithTimeout(ctx, r.timeout)
|
||||
defer cancel()
|
||||
err := checker.Check(checkCtx)
|
||||
if err != nil {
|
||||
ch <- namedResult{name, Result{OK: false, Error: err.Error()}}
|
||||
return
|
||||
}
|
||||
ch <- namedResult{name, Result{OK: true}}
|
||||
}(name, checker)
|
||||
}
|
||||
|
||||
results = make(map[string]Result, len(r.checks))
|
||||
ready = true
|
||||
for i := 0; i < len(r.checks); i++ {
|
||||
nr := <-ch
|
||||
results[nr.name] = nr.result
|
||||
if !nr.result.OK {
|
||||
ready = false
|
||||
}
|
||||
}
|
||||
return ready, results
|
||||
}
|
||||
@@ -1,161 +0,0 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: simulierter Datenbankausfall fuehrt zu
|
||||
// "nicht bereit".
|
||||
func TestReadinessHandler_ReportsNotReadyOnDatabaseFailure(t *testing.T) {
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
// Datenbankausfall simulieren: Pool sofort schliessen, bevor der Check laeuft.
|
||||
pool.Close()
|
||||
|
||||
reg := NewRegistry()
|
||||
reg.Register("database", DatabaseChecker(pool))
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(rec, req)
|
||||
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("status = %d, want 503 bei db-ausfall", rec.Code)
|
||||
}
|
||||
|
||||
var body struct {
|
||||
Status string `json:"status"`
|
||||
Checks map[string]interface{} `json:"checks"`
|
||||
}
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("body parsen: %v", err)
|
||||
}
|
||||
if body.Status != "not_ready" {
|
||||
t.Fatalf("status-feld = %q, want not_ready", body.Status)
|
||||
}
|
||||
if _, ok := body.Checks["database"]; !ok {
|
||||
t.Fatal("erwartet 'database' im checks-ergebnis")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadinessHandler_ReportsReadyWhenAllChecksPass(t *testing.T) {
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
defer pool.Close()
|
||||
|
||||
reg := NewRegistry()
|
||||
reg.Register("database", DatabaseChecker(pool))
|
||||
reg.Register("queue", QueueChecker(pool))
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 bei funktionierenden abhaengigkeiten", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 3: Liveness und Readiness unterscheiden
|
||||
// sich nachweislich im Fehlerfall.
|
||||
func TestLivenessAndReadiness_DifferOnDatabaseFailure(t *testing.T) {
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
pool.Close() // db-ausfall simulieren
|
||||
|
||||
reg := NewRegistry()
|
||||
reg.Register("database", DatabaseChecker(pool))
|
||||
|
||||
livenessRec := httptest.NewRecorder()
|
||||
LivenessHandler(livenessRec, httptest.NewRequest(http.MethodGet, "/livez", nil))
|
||||
if livenessRec.Code != http.StatusOK {
|
||||
t.Fatalf("liveness status = %d, want 200 trotz db-ausfall (liveness prueft keine abhaengigkeiten)", livenessRec.Code)
|
||||
}
|
||||
|
||||
readinessRec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(readinessRec, httptest.NewRequest(http.MethodGet, "/readyz", nil))
|
||||
if readinessRec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("readiness status = %d, want 503 bei db-ausfall", readinessRec.Code)
|
||||
}
|
||||
|
||||
if livenessRec.Code == readinessRec.Code {
|
||||
t.Fatal("liveness und readiness sollten sich im db-ausfall-fall unterscheiden")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Health-Endpunkt antwortet auch bei
|
||||
// haengendem Check innerhalb definierter Zeit (Timeout begrenzt die Dauer).
|
||||
func TestReadinessHandler_RespondsWithinTimeoutEvenWithHangingCheck(t *testing.T) {
|
||||
reg := NewRegistry().WithTimeout(50 * time.Millisecond)
|
||||
reg.Register("haengender_dienst", CheckerFunc(func(ctx context.Context) error {
|
||||
select {
|
||||
case <-time.After(10 * time.Second): // wuerde ohne timeout ewig blockieren
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}))
|
||||
|
||||
start := time.Now()
|
||||
req := httptest.NewRequest(http.MethodGet, "/readyz", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
reg.ReadinessHandler()(rec, req)
|
||||
elapsed := time.Since(start)
|
||||
|
||||
if elapsed > time.Second {
|
||||
t.Fatalf("readiness handler brauchte %s, erwartet deutlich unter 1s durch timeout", elapsed)
|
||||
}
|
||||
if rec.Code != http.StatusServiceUnavailable {
|
||||
t.Fatalf("status = %d, want 503 fuer haengenden/timeout-check", rec.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCheckAll_MultipleChecksRunConcurrently(t *testing.T) {
|
||||
reg := NewRegistry().WithTimeout(time.Second)
|
||||
reg.Register("a", CheckerFunc(func(ctx context.Context) error { return nil }))
|
||||
reg.Register("b", CheckerFunc(func(ctx context.Context) error { return errors.New("kaputt") }))
|
||||
|
||||
ready, results := reg.CheckAll(context.Background())
|
||||
if ready {
|
||||
t.Fatal("erwartet ready=false, da 'b' fehlschlaegt")
|
||||
}
|
||||
if !results["a"].OK {
|
||||
t.Fatalf("erwartet 'a' ok, habe %+v", results["a"])
|
||||
}
|
||||
if results["b"].OK || results["b"].Error == "" {
|
||||
t.Fatalf("erwartet 'b' fehlgeschlagen mit fehlertext, habe %+v", results["b"])
|
||||
}
|
||||
}
|
||||
@@ -1,19 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import "github.com/prometheus/client_golang/prometheus"
|
||||
|
||||
// NewCoreRegistry liefert das Prometheus-Registry fuer die EIGENEN
|
||||
// Kennzahlen des Core-Dienstes (Akzeptanzkriterium 1) — alle Namen tragen
|
||||
// das Praefix "nexarch_core_" gemaess der im Paketkommentar dokumentierten
|
||||
// Namenskonvention (Akzeptanzkriterium 3). Ein eigenes Registry statt des
|
||||
// globalen DefaultRegisterer, damit Tests unabhaengig voneinander sind.
|
||||
func NewCoreRegistry() *prometheus.Registry {
|
||||
reg := prometheus.NewRegistry()
|
||||
reg.MustRegister(
|
||||
prometheus.NewGaugeFunc(prometheus.GaugeOpts{
|
||||
Name: "nexarch_core_up",
|
||||
Help: "1, solange der Core-Dienst laeuft und Metriken liefern kann.",
|
||||
}, func() float64 { return 1 }),
|
||||
)
|
||||
return reg
|
||||
}
|
||||
@@ -1,168 +0,0 @@
|
||||
// Package metrics implementiert Core OPS-03: einen zentralen /metrics-
|
||||
// Endpunkt im Prometheus-Textformat, der Kennzahlen des Core-Dienstes UND
|
||||
// aggregierte Kennzahlen aller registrierten Module bereitstellt — offenes
|
||||
// Pull-Modell nach Prometheus-Vorbild, kein proprietaerer Push-Mechanismus.
|
||||
//
|
||||
// Namenskonvention (Akzeptanzkriterium 3, modulübergreifend konsistent):
|
||||
//
|
||||
// nexarch_core_<name> — Kennzahlen des Core-Dienstes selbst
|
||||
// nexarch_module_<modul>_<name> — von einem Modul gescrapte Kennzahl
|
||||
// <name>, umbenannt mit dem
|
||||
// Modulnamen als Praefix
|
||||
//
|
||||
// Ein Modul liefert seine eigenen Kennzahlen unter EIGENEM Namen (z.B.
|
||||
// "requests_total") unter seinem eigenen /metrics-Endpunkt — dieses Paket
|
||||
// benennt sie beim Einsammeln konsistent um, damit im aggregierten Core-
|
||||
// Endpunkt niemals zwei Module denselben Metrik-Namen kollidieren lassen.
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
dto "github.com/prometheus/client_model/go"
|
||||
"github.com/prometheus/common/expfmt"
|
||||
"github.com/prometheus/common/model"
|
||||
)
|
||||
|
||||
// init erzwingt das klassische Prometheus-Namensschema (a-z, A-Z, 0-9, _)
|
||||
// fuer die Namensvalidierung von expfmt/model — ohne diese explizite
|
||||
// Festlegung liefert die Bibliothek "Invalid name validation scheme
|
||||
// requested: unset" beim Parsen/Kodieren, da sie den globalen Default in
|
||||
// dieser Version nicht mehr implizit setzt.
|
||||
func init() {
|
||||
model.NameValidationScheme = model.LegacyValidation
|
||||
}
|
||||
|
||||
// Source ist EIN registriertes Modul mit seinem eigenen /metrics-Endpunkt
|
||||
// (siehe internal/health fuer das analoge Muster bei Readiness-Checks).
|
||||
type Source struct {
|
||||
ModuleName string
|
||||
MetricsURL string
|
||||
}
|
||||
|
||||
// SourceProvider liefert die aktuell registrierten Module — typischerweise
|
||||
// rueckgebunden an internal/moduleregistry.Registry.List (API-02) ueber
|
||||
// einen kleinen Adapter im aufrufenden Code, damit dieses Paket
|
||||
// internal/moduleregistry nicht direkt importieren muss (Kein Umbau
|
||||
// angrenzender Bereiche). Ein NEU registriertes Modul erscheint automatisch
|
||||
// beim naechsten Aufruf von Aggregator.Handler, OHNE Codeaenderung an diesem
|
||||
// Paket (Akzeptanzkriterium 2 / Pruefung 2).
|
||||
type SourceProvider func(ctx context.Context) ([]Source, error)
|
||||
|
||||
// FetchTimeout begrenzt, wie lange EIN Modul-Scrape maximal dauern darf —
|
||||
// ein haengendes Modul darf den gesamten Aggregations-Request nicht
|
||||
// verzoegern (Pruefung 1: Antwort unter Last innerhalb definierter Zeit).
|
||||
const FetchTimeout = 2 * time.Second
|
||||
|
||||
// Aggregator sammelt Core-eigene Metriken (coreGatherer) und die Metriken
|
||||
// aller ueber sourceProvider gemeldeten Module in EINER Antwort ein.
|
||||
type Aggregator struct {
|
||||
coreGatherer prometheus.Gatherer
|
||||
sourceProvider SourceProvider
|
||||
client *http.Client
|
||||
}
|
||||
|
||||
func NewAggregator(coreGatherer prometheus.Gatherer, sourceProvider SourceProvider) *Aggregator {
|
||||
return &Aggregator{
|
||||
coreGatherer: coreGatherer,
|
||||
sourceProvider: sourceProvider,
|
||||
client: &http.Client{Timeout: FetchTimeout},
|
||||
}
|
||||
}
|
||||
|
||||
// Gather implementiert prometheus.Gatherer: liefert Core-Metriken PLUS alle
|
||||
// erreichbaren Modul-Metriken (umbenannt gemaess Namenskonvention) in einer
|
||||
// gemeinsamen Liste von MetricFamilies.
|
||||
func (a *Aggregator) Gather(ctx context.Context) ([]*dto.MetricFamily, error) {
|
||||
families, err := a.coreGatherer.Gather()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("core-metriken einsammeln: %w", err)
|
||||
}
|
||||
|
||||
sources, err := a.sourceProvider(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("modul-quellen ermitteln: %w", err)
|
||||
}
|
||||
|
||||
// Module werden NEBENLAEUFIG gescrapt (dasselbe Muster wie
|
||||
// internal/health.Registry.CheckAll) — ein langsames/nicht erreichbares
|
||||
// Modul haelt weder andere Module noch den Gesamt-Request auf.
|
||||
type fetchResult struct {
|
||||
families []*dto.MetricFamily
|
||||
}
|
||||
resultCh := make(chan fetchResult, len(sources))
|
||||
for _, src := range sources {
|
||||
go func(src Source) {
|
||||
fetchCtx, cancel := context.WithTimeout(ctx, FetchTimeout)
|
||||
defer cancel()
|
||||
mf, err := a.fetchAndRename(fetchCtx, src)
|
||||
if err != nil {
|
||||
resultCh <- fetchResult{} // Fehlerfall: einfach nichts beitragen, Aggregation laeuft weiter
|
||||
return
|
||||
}
|
||||
resultCh <- fetchResult{families: mf}
|
||||
}(src)
|
||||
}
|
||||
for range sources {
|
||||
r := <-resultCh
|
||||
families = append(families, r.families...)
|
||||
}
|
||||
|
||||
return families, nil
|
||||
}
|
||||
|
||||
// fetchAndRename ruft die /metrics-URL eines Moduls ab, parst das
|
||||
// Prometheus-Textformat und benennt jede Metrik gemaess der
|
||||
// Namenskonvention um (Akzeptanzkriterium 2 + 3).
|
||||
func (a *Aggregator) fetchAndRename(ctx context.Context, src Source) ([]*dto.MetricFamily, error) {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, src.MetricsURL, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
resp, err := a.client.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("modul %s: unerwarteter status %d", src.ModuleName, resp.StatusCode)
|
||||
}
|
||||
|
||||
parser := expfmt.NewTextParser(model.LegacyValidation)
|
||||
parsed, err := parser.TextToMetricFamilies(resp.Body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("modul %s: metrik-text nicht parsebar: %w", src.ModuleName, err)
|
||||
}
|
||||
|
||||
out := make([]*dto.MetricFamily, 0, len(parsed))
|
||||
for name, mf := range parsed {
|
||||
renamed := fmt.Sprintf("nexarch_module_%s_%s", src.ModuleName, name)
|
||||
mf.Name = &renamed
|
||||
out = append(out, mf)
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Handler liefert einen HTTP-Handler, der Gather aufruft und das Ergebnis im
|
||||
// Prometheus-Textformat ausgibt (Akzeptanzkriterium 1).
|
||||
func (a *Aggregator) Handler() http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
families, err := a.Gather(r.Context())
|
||||
if err != nil {
|
||||
http.Error(w, "metriken konnten nicht eingesammelt werden", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", string(expfmt.NewFormat(expfmt.TypeTextPlain)))
|
||||
enc := expfmt.NewEncoder(w, expfmt.NewFormat(expfmt.TypeTextPlain))
|
||||
for _, mf := range families {
|
||||
if err := enc.Encode(mf); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,179 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/prometheus/common/expfmt"
|
||||
"github.com/prometheus/common/model"
|
||||
)
|
||||
|
||||
func fakeModuleServer(metricName string) *httptest.Server {
|
||||
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain; version=0.0.4")
|
||||
fmt.Fprintf(w, "# HELP %s ein test-zaehler\n# TYPE %s counter\n%s 42\n", metricName, metricName, metricName)
|
||||
}))
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: Core liefert unter dem Handler valides
|
||||
// Prometheus-Textformat mit den eigenen Kennzahlen.
|
||||
func TestHandler_ServesCoreMetricsInPrometheusFormat(t *testing.T) {
|
||||
agg := NewAggregator(NewCoreRegistry(), func(ctx context.Context) ([]Source, error) { return nil, nil })
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/metrics", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
agg.Handler()(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", rec.Code)
|
||||
}
|
||||
|
||||
parser := expfmt.NewTextParser(model.LegacyValidation)
|
||||
families, err := parser.TextToMetricFamilies(strings.NewReader(rec.Body.String()))
|
||||
if err != nil {
|
||||
t.Fatalf("antwort ist kein valides prometheus-textformat: %v", err)
|
||||
}
|
||||
if _, ok := families["nexarch_core_up"]; !ok {
|
||||
t.Fatalf("erwartet 'nexarch_core_up' unter den core-metriken, habe: %v", keysOf(families))
|
||||
}
|
||||
}
|
||||
|
||||
func keysOf[V any](m map[string]V) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: ein NEU registriertes Modul erscheint
|
||||
// in der Aggregation, OHNE dass dieses Paket oder der Aufrufer Code
|
||||
// aendern muss — die Quelle kommt ausschliesslich aus sourceProvider.
|
||||
func TestHandler_NewlyRegisteredModuleAppearsWithoutCodeChange(t *testing.T) {
|
||||
moduleServer := fakeModuleServer("requests_total")
|
||||
defer moduleServer.Close()
|
||||
|
||||
// Simuliert eine sich zur Laufzeit aendernde Modul-Liste (z.B. aus
|
||||
// SourceStore.Provide) — zunaechst LEER, dann mit einem Eintrag.
|
||||
var sources []Source
|
||||
var mu sync.Mutex
|
||||
provider := func(ctx context.Context) ([]Source, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
out := make([]Source, len(sources))
|
||||
copy(out, sources)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
agg := NewAggregator(NewCoreRegistry(), provider)
|
||||
|
||||
// Vor der Registrierung: Modul-Metrik nicht vorhanden.
|
||||
rec1 := httptest.NewRecorder()
|
||||
agg.Handler()(rec1, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
if strings.Contains(rec1.Body.String(), "requests_total") {
|
||||
t.Fatal("modul-metrik haette vor registrierung nicht erscheinen duerfen")
|
||||
}
|
||||
|
||||
// Modul wird "registriert" (kein Code hier oder in metrics.go aendert sich).
|
||||
mu.Lock()
|
||||
sources = append(sources, Source{ModuleName: "dms", MetricsURL: moduleServer.URL})
|
||||
mu.Unlock()
|
||||
|
||||
rec2 := httptest.NewRecorder()
|
||||
agg.Handler()(rec2, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
body := rec2.Body.String()
|
||||
if !strings.Contains(body, "nexarch_module_dms_requests_total") {
|
||||
t.Fatalf("erwartet umbenannte modul-metrik 'nexarch_module_dms_requests_total' nach registrierung, body:\n%s", body)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3: Namenskonvention "nexarch_module_<modul>_<name>"
|
||||
// wird tatsaechlich angewendet.
|
||||
func TestFetchAndRename_AppliesNamingConvention(t *testing.T) {
|
||||
moduleServer := fakeModuleServer("queue_depth")
|
||||
defer moduleServer.Close()
|
||||
|
||||
agg := NewAggregator(NewCoreRegistry(), nil)
|
||||
families, err := agg.fetchAndRename(context.Background(), Source{ModuleName: "mail", MetricsURL: moduleServer.URL})
|
||||
if err != nil {
|
||||
t.Fatalf("fetchAndRename: %v", err)
|
||||
}
|
||||
if len(families) != 1 || families[0].GetName() != "nexarch_module_mail_queue_depth" {
|
||||
t.Fatalf("erwartet genau 1 metrik 'nexarch_module_mail_queue_depth', habe: %+v", families)
|
||||
}
|
||||
}
|
||||
|
||||
// Ein nicht erreichbares Modul darf die Aggregation der uebrigen und die
|
||||
// Gesamtantwort nicht verhindern (dieselbe Resilienz wie OPS-02).
|
||||
func TestHandler_UnreachableModuleDoesNotBreakAggregation(t *testing.T) {
|
||||
reachable := fakeModuleServer("healthy_metric")
|
||||
defer reachable.Close()
|
||||
unreachable := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
|
||||
unreachableURL := unreachable.URL
|
||||
unreachable.Close() // sofort schliessen -> Verbindung schlaegt fehl
|
||||
|
||||
provider := func(ctx context.Context) ([]Source, error) {
|
||||
return []Source{
|
||||
{ModuleName: "ok", MetricsURL: reachable.URL},
|
||||
{ModuleName: "kaputt", MetricsURL: unreachableURL},
|
||||
}, nil
|
||||
}
|
||||
agg := NewAggregator(NewCoreRegistry(), provider)
|
||||
|
||||
rec := httptest.NewRecorder()
|
||||
agg.Handler()(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200 trotz einem nicht erreichbaren modul", rec.Code)
|
||||
}
|
||||
body := rec.Body.String()
|
||||
if !strings.Contains(body, "nexarch_module_ok_healthy_metric") {
|
||||
t.Fatal("erreichbares modul haette trotz ausfall des anderen aggregiert werden sollen")
|
||||
}
|
||||
if strings.Contains(body, "kaputt") {
|
||||
t.Fatal("nicht erreichbares modul haette keine metrik beitragen duerfen")
|
||||
}
|
||||
}
|
||||
|
||||
// Pruefung 1: Endpunkt antwortet unter mehreren gleichzeitigen Anfragen
|
||||
// innerhalb definierter Zeit — kein unbeschraenktes Blockieren durch
|
||||
// langsame Module (FetchTimeout begrenzt jeden Scrape).
|
||||
func TestHandler_RespondsWithinBoundedTimeUnderLoad(t *testing.T) {
|
||||
hangingServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
time.Sleep(10 * time.Second) // wuerde ohne timeout jede anfrage blockieren
|
||||
}))
|
||||
defer hangingServer.Close()
|
||||
|
||||
provider := func(ctx context.Context) ([]Source, error) {
|
||||
return []Source{{ModuleName: "haengend", MetricsURL: hangingServer.URL}}, nil
|
||||
}
|
||||
agg := NewAggregator(NewCoreRegistry(), provider)
|
||||
|
||||
const concurrentRequests = 10
|
||||
var wg sync.WaitGroup
|
||||
start := time.Now()
|
||||
for i := 0; i < concurrentRequests; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
rec := httptest.NewRecorder()
|
||||
agg.Handler()(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Errorf("status = %d, want 200", rec.Code)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
elapsed := time.Since(start)
|
||||
|
||||
if elapsed > FetchTimeout+3*time.Second {
|
||||
t.Fatalf("%d gleichzeitige anfragen brauchten %s, erwartet deutlich unter %s durch FetchTimeout",
|
||||
concurrentRequests, elapsed, FetchTimeout+3*time.Second)
|
||||
}
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// SourceStore persistiert, welche Module ihre Metriken unter welcher URL
|
||||
// bereitstellen — dieselbe Postgres-basierte "kein Code-Deploy noetig"-
|
||||
// Konvention wie internal/statuspage.Store.RegisterTarget (OPS-02): ein neu
|
||||
// registriertes Modul erscheint automatisch in der Aggregation, sobald es
|
||||
// hier eingetragen ist (Akzeptanzkriterium 2 / Pruefung 2).
|
||||
type SourceStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewSourceStore(pool *pgxpool.Pool) *SourceStore {
|
||||
return &SourceStore{pool: pool}
|
||||
}
|
||||
|
||||
func (s *SourceStore) RegisterSource(ctx context.Context, moduleName, metricsURL string) error {
|
||||
_, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO metrics_sources (module_name, metrics_url)
|
||||
VALUES ($1, $2)
|
||||
ON CONFLICT (module_name) DO UPDATE SET metrics_url = $2
|
||||
`, moduleName, metricsURL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("metrik-quelle speichern: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Provide implementiert SourceProvider direkt aus der Datenbank.
|
||||
func (s *SourceStore) Provide(ctx context.Context) ([]Source, error) {
|
||||
rows, err := s.pool.Query(ctx, `SELECT module_name, metrics_url FROM metrics_sources ORDER BY module_name`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("metrik-quellen auflisten: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []Source
|
||||
for rows.Next() {
|
||||
var src Source
|
||||
if err := rows.Scan(&src.ModuleName, &src.MetricsURL); err != nil {
|
||||
return nil, fmt.Errorf("metrik-quelle lesen: %w", err)
|
||||
}
|
||||
out = append(out, src)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -1,60 +0,0 @@
|
||||
package metrics
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupSourceStoreTest(t *testing.T) (*SourceStore, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS metrics_sources (module_name TEXT PRIMARY KEY, metrics_url TEXT NOT NULL)
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
cleanup := func() { pool.Close() }
|
||||
return NewSourceStore(pool), cleanup
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 / Pruefung 2 auf Persistenz-Ebene: eine ueber die
|
||||
// Datenbank registrierte Quelle ist sofort ueber Provide() sichtbar — genau
|
||||
// der Mechanismus, der ein neues Modul ohne Core-Codeaenderung erscheinen
|
||||
// laesst.
|
||||
func TestSourceStore_RegisterSourceAppearsInProvide(t *testing.T) {
|
||||
store, cleanup := setupSourceStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
name := fmt.Sprintf("modul-%d", time.Now().UnixNano())
|
||||
|
||||
if err := store.RegisterSource(ctx, name, "http://example.invalid/metrics"); err != nil {
|
||||
t.Fatalf("registersource: %v", err)
|
||||
}
|
||||
|
||||
sources, err := store.Provide(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("provide: %v", err)
|
||||
}
|
||||
found := false
|
||||
for _, s := range sources {
|
||||
if s.ModuleName == name {
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
t.Fatalf("erwartet %s in provide()-ergebnis, habe: %+v", name, sources)
|
||||
}
|
||||
}
|
||||
@@ -1,99 +0,0 @@
|
||||
package moduleregistry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrModuleNotRegistered = errors.New("moduleregistry: modul muss vor provisionierung registriert sein")
|
||||
ErrInvalidCredential = errors.New("moduleregistry: ungueltiges oder fehlendes service-credential")
|
||||
)
|
||||
|
||||
// Provision stellt ein Service-Credential (Client-ID + Secret) fuer eine
|
||||
// Modul-Instanz aus (Akzeptanzkriterium 4). Das Secret wird NUR beim
|
||||
// Ausstellen im Klartext zurueckgegeben, gespeichert wird ausschliesslich
|
||||
// dessen SHA-256-Hash.
|
||||
func (r *Registry) Provision(ctx context.Context, moduleName string) (clientID, secret string, err error) {
|
||||
if _, err := r.Get(ctx, moduleName); err != nil {
|
||||
if errors.Is(err, ErrModuleNotFound) {
|
||||
return "", "", ErrModuleNotRegistered
|
||||
}
|
||||
return "", "", err
|
||||
}
|
||||
|
||||
clientID, err = randomToken(16)
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("client-id erzeugen: %w", err)
|
||||
}
|
||||
secret, err = randomToken(32)
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("secret erzeugen: %w", err)
|
||||
}
|
||||
hash := hashSecret(secret)
|
||||
|
||||
_, err = r.pool.Exec(ctx, `
|
||||
INSERT INTO module_credentials (module_name, client_id, secret_hash, issued_at)
|
||||
VALUES ($1, $2, $3, now())
|
||||
ON CONFLICT (module_name) DO UPDATE SET client_id = $2, secret_hash = $3, issued_at = now()
|
||||
`, moduleName, clientID, hash)
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("credential speichern: %w", err)
|
||||
}
|
||||
return clientID, secret, nil
|
||||
}
|
||||
|
||||
// Authenticate prueft ein Service-Credential timing-safe (Referenzmuster
|
||||
// siehe AUD-02) — Aufrufe ohne gueltiges Credential werden abgelehnt
|
||||
// (Akzeptanzkriterium 4 / Pruefung 4).
|
||||
func (r *Registry) Authenticate(ctx context.Context, clientID, secret string) (moduleName string, ok bool, err error) {
|
||||
if clientID == "" || secret == "" {
|
||||
return "", false, nil
|
||||
}
|
||||
|
||||
var storedHash []byte
|
||||
err = r.pool.QueryRow(ctx, `
|
||||
SELECT module_name, secret_hash FROM module_credentials WHERE client_id = $1
|
||||
`, clientID).Scan(&moduleName, &storedHash)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return "", false, nil
|
||||
}
|
||||
return "", false, fmt.Errorf("credential lesen: %w", err)
|
||||
}
|
||||
|
||||
if !timingSafeEqual(hashSecret(secret), storedHash) {
|
||||
return "", false, nil
|
||||
}
|
||||
return moduleName, true, nil
|
||||
}
|
||||
|
||||
func randomToken(n int) (string, error) {
|
||||
buf := make([]byte, n)
|
||||
if _, err := rand.Read(buf); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return hex.EncodeToString(buf), nil
|
||||
}
|
||||
|
||||
func hashSecret(secret string) []byte {
|
||||
sum := sha256.Sum256([]byte(secret))
|
||||
return sum[:]
|
||||
}
|
||||
|
||||
// timingSafeEqual folgt derselben Referenzimplementierung wie AUD-02
|
||||
// (subtle.ConstantTimeCompare) — projektweite Konvention fuer jeden
|
||||
// sicherheitsrelevanten Vergleich.
|
||||
func timingSafeEqual(a, b []byte) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
return subtle.ConstantTimeCompare(a, b) == 1
|
||||
}
|
||||
@@ -1,46 +0,0 @@
|
||||
package moduleregistry
|
||||
|
||||
import "net/http"
|
||||
|
||||
// RequireActiveModule weist Anfragen an ein nicht aktiviertes Modul ZENTRAL
|
||||
// ab, bevor der eigentliche Modul-Handler erreicht wird (Akzeptanzkriterium 2 /
|
||||
// Pruefung 1) — Casbin-Prinzip: Durchsetzung als Middleware statt verstreuter
|
||||
// Pruefungen in jedem Handler. tenantSlug/moduleName werden hier ueber
|
||||
// Query-Parameter gelesen (echte Extraktion aus JWT/Tenant-Kontext ist
|
||||
// API-05/TEN-06, nicht Teil dieser Kachel).
|
||||
func (r *Registry) RequireActiveModule(moduleName string, next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, req *http.Request) {
|
||||
tenantSlug := req.URL.Query().Get("tenant")
|
||||
active, err := r.IsActive(req.Context(), tenantSlug, moduleName)
|
||||
if err != nil {
|
||||
http.Error(w, "aktivierungspruefung fehlgeschlagen", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if !active {
|
||||
http.Error(w, "modul nicht aktiviert", http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
next(w, req)
|
||||
}
|
||||
}
|
||||
|
||||
// RequireServiceCredential authentifiziert eine Modul-Instanz ueber ihr
|
||||
// Service-Credential (X-Client-Id/X-Client-Secret-Header) BEVOR der
|
||||
// eigentliche Handler erreicht wird (Akzeptanzkriterium 4 / Pruefung 4).
|
||||
func (r *Registry) RequireServiceCredential(next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, req *http.Request) {
|
||||
clientID := req.Header.Get("X-Client-Id")
|
||||
secret := req.Header.Get("X-Client-Secret")
|
||||
|
||||
_, ok, err := r.Authenticate(req.Context(), clientID, secret)
|
||||
if err != nil {
|
||||
http.Error(w, "authentifizierung fehlgeschlagen", http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
if !ok {
|
||||
http.Error(w, ErrInvalidCredential.Error(), http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
next(w, req)
|
||||
}
|
||||
}
|
||||
@@ -1,120 +0,0 @@
|
||||
// Package moduleregistry implementiert Core API-02: die Registry, in der
|
||||
// sich Fachmodule (DMS, Mail, weitere) mit Metadaten eintragen, gekoppelt an
|
||||
// die Aktivierungspruefung aus LIC-02 (Feature-Flags). Zusaetzlich
|
||||
// authentifiziert die Registry Modul-Instanzen selbst ueber ein bei
|
||||
// Provisionierung ausgestelltes Service-Credential.
|
||||
package moduleregistry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrMissingName = errors.New("moduleregistry: name darf nicht leer sein")
|
||||
ErrMissingVersion = errors.New("moduleregistry: version darf nicht leer sein")
|
||||
ErrModuleNotFound = errors.New("moduleregistry: modul nicht registriert")
|
||||
)
|
||||
|
||||
type Module struct {
|
||||
Name string
|
||||
Version string
|
||||
RequiredFlags []string
|
||||
}
|
||||
|
||||
type Registry struct {
|
||||
pool *pgxpool.Pool
|
||||
flags *flag.Service
|
||||
}
|
||||
|
||||
func NewRegistry(pool *pgxpool.Pool, flags *flag.Service) *Registry {
|
||||
return &Registry{pool: pool, flags: flags}
|
||||
}
|
||||
|
||||
// Register traegt ein Modul mit Name, Version und benoetigten Feature-Flags
|
||||
// ein (Akzeptanzkriterium 1). Fehlende Pflichtangaben werden abgewiesen
|
||||
// (Akzeptanzkriterium 1 / Pruefung 2). Erneutes Register desselben Namens
|
||||
// aktualisiert Version/Flags (Redeploy-Fall).
|
||||
func (r *Registry) Register(ctx context.Context, name, version string, requiredFlags []string) (Module, error) {
|
||||
if name == "" {
|
||||
return Module{}, ErrMissingName
|
||||
}
|
||||
if version == "" {
|
||||
return Module{}, ErrMissingVersion
|
||||
}
|
||||
if requiredFlags == nil {
|
||||
requiredFlags = []string{}
|
||||
}
|
||||
|
||||
_, err := r.pool.Exec(ctx, `
|
||||
INSERT INTO modules (name, version, required_flags, registered_at)
|
||||
VALUES ($1, $2, $3, now())
|
||||
ON CONFLICT (name) DO UPDATE SET version = $2, required_flags = $3, registered_at = now()
|
||||
`, name, version, requiredFlags)
|
||||
if err != nil {
|
||||
return Module{}, fmt.Errorf("modul registrieren: %w", err)
|
||||
}
|
||||
return Module{Name: name, Version: version, RequiredFlags: requiredFlags}, nil
|
||||
}
|
||||
|
||||
func (r *Registry) Get(ctx context.Context, name string) (Module, error) {
|
||||
var m Module
|
||||
m.Name = name
|
||||
err := r.pool.QueryRow(ctx, `
|
||||
SELECT version, required_flags FROM modules WHERE name = $1
|
||||
`, name).Scan(&m.Version, &m.RequiredFlags)
|
||||
if err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Module{}, ErrModuleNotFound
|
||||
}
|
||||
return Module{}, fmt.Errorf("modul lesen: %w", err)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
// List liefert alle registrierten Module (Akzeptanzkriterium 3: ueber API
|
||||
// abfragbar, z.B. fuer Statusseite/Lizenzoberflaeche).
|
||||
func (r *Registry) List(ctx context.Context) ([]Module, error) {
|
||||
rows, err := r.pool.Query(ctx, `SELECT name, version, required_flags FROM modules ORDER BY name`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("module auflisten: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []Module
|
||||
for rows.Next() {
|
||||
var m Module
|
||||
if err := rows.Scan(&m.Name, &m.Version, &m.RequiredFlags); err != nil {
|
||||
return nil, fmt.Errorf("modul lesen: %w", err)
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// IsActive prueft, ob ein registriertes Modul fuer einen Tenant aktiviert
|
||||
// ist: registriert UND alle benoetigten Feature-Flags sind fuer diesen
|
||||
// Tenant aktiv (Akzeptanzkriterium 2). Ein nicht registriertes Modul gilt
|
||||
// immer als nicht aktiv.
|
||||
func (r *Registry) IsActive(ctx context.Context, tenantSlug, moduleName string) (bool, error) {
|
||||
m, err := r.Get(ctx, moduleName)
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrModuleNotFound) {
|
||||
return false, nil
|
||||
}
|
||||
return false, err
|
||||
}
|
||||
|
||||
for _, flagKey := range m.RequiredFlags {
|
||||
if !r.flags.IsEnabled(ctx, tenantSlug, flagKey) {
|
||||
return false, nil
|
||||
}
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
@@ -1,304 +0,0 @@
|
||||
package moduleregistry
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*Registry, *flag.Store, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS feature_flags (
|
||||
key TEXT PRIMARY KEY, enabled BOOLEAN NOT NULL DEFAULT false,
|
||||
rollout_percentage INT NOT NULL DEFAULT 0, target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS modules (
|
||||
name TEXT PRIMARY KEY, version TEXT NOT NULL CHECK (version <> ''),
|
||||
required_flags TEXT[] NOT NULL DEFAULT '{}', registered_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS module_credentials (
|
||||
module_name TEXT PRIMARY KEY REFERENCES modules(name),
|
||||
client_id TEXT NOT NULL UNIQUE, secret_hash BYTEA NOT NULL,
|
||||
issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
flagStore := flag.NewStore(pool)
|
||||
// Kurze TTL, damit Tests, die den Flag-Store direkt aendern (an
|
||||
// Registry.IsActive vorbei), den neuen Stand ohne manuelles Invalidate
|
||||
// zuverlaessig sehen.
|
||||
flagService := flag.NewService(flagStore, 10*time.Millisecond)
|
||||
registry := NewRegistry(pool, flagService)
|
||||
|
||||
cleanup := func() { pool.Close() }
|
||||
return registry, flagStore, cleanup
|
||||
}
|
||||
|
||||
func uniqueModuleName(t *testing.T) string {
|
||||
return fmt.Sprintf("dms_%d", time.Now().UnixNano())
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 2: fehlende Pflichtangaben abgewiesen.
|
||||
func TestRegister_RejectsMissingFields(t *testing.T) {
|
||||
registry, _, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := registry.Register(ctx, "", "1.0", nil); !errors.Is(err, ErrMissingName) {
|
||||
t.Fatalf("erwartet ErrMissingName, habe %v", err)
|
||||
}
|
||||
if _, err := registry.Register(ctx, "dms", "", nil); !errors.Is(err, ErrMissingVersion) {
|
||||
t.Fatalf("erwartet ErrMissingVersion, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRegister_AndGet(t *testing.T) {
|
||||
registry, _, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
name := uniqueModuleName(t)
|
||||
|
||||
m, err := registry.Register(ctx, name, "1.2.0", []string{"dms_enabled"})
|
||||
if err != nil {
|
||||
t.Fatalf("register: %v", err)
|
||||
}
|
||||
if m.Version != "1.2.0" || len(m.RequiredFlags) != 1 {
|
||||
t.Fatalf("unerwartet: %+v", m)
|
||||
}
|
||||
|
||||
got, err := registry.Get(ctx, name)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.Version != "1.2.0" {
|
||||
t.Fatalf("get version = %q", got.Version)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + 3 + Pruefung 3: konsistente Daten nach
|
||||
// Aktivierung/Deaktivierung eines Moduls.
|
||||
func TestIsActive_ReflectsFlagStateConsistently(t *testing.T) {
|
||||
registry, flagStore, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
name := uniqueModuleName(t)
|
||||
flagKey := name + "_enabled"
|
||||
|
||||
if _, err := registry.Register(ctx, name, "1.0", []string{flagKey}); err != nil {
|
||||
t.Fatalf("register: %v", err)
|
||||
}
|
||||
|
||||
active, err := registry.IsActive(ctx, "acme", name)
|
||||
if err != nil {
|
||||
t.Fatalf("is active (vor flag): %v", err)
|
||||
}
|
||||
if active {
|
||||
t.Fatal("erwartet nicht aktiv, solange flag nicht gesetzt ist")
|
||||
}
|
||||
|
||||
if err := flagStore.Set(ctx, flag.Flag{Key: flagKey, Enabled: true}); err != nil {
|
||||
t.Fatalf("flag setzen: %v", err)
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond) // TTL abwarten
|
||||
|
||||
active, err = registry.IsActive(ctx, "acme", name)
|
||||
if err != nil {
|
||||
t.Fatalf("is active (nach flag an): %v", err)
|
||||
}
|
||||
if !active {
|
||||
t.Fatal("erwartet aktiv, nachdem flag aktiviert wurde")
|
||||
}
|
||||
|
||||
if err := flagStore.Set(ctx, flag.Flag{Key: flagKey, Enabled: false}); err != nil {
|
||||
t.Fatalf("flag zuruecksetzen: %v", err)
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond) // TTL abwarten
|
||||
active, err = registry.IsActive(ctx, "acme", name)
|
||||
if err != nil {
|
||||
t.Fatalf("is active (nach flag aus): %v", err)
|
||||
}
|
||||
if active {
|
||||
t.Fatal("erwartet wieder nicht aktiv, nachdem flag deaktiviert wurde")
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsActive_UnregisteredModuleIsNeverActive(t *testing.T) {
|
||||
registry, _, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
active, err := registry.IsActive(ctx, "acme", "nie-registriert")
|
||||
if err != nil {
|
||||
t.Fatalf("is active: %v", err)
|
||||
}
|
||||
if active {
|
||||
t.Fatal("unregistriertes modul darf nie aktiv sein")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 1: Anfrage an deaktiviertes Modul wird
|
||||
// zentral abgewiesen, BEVOR die Modul-Logik erreicht wird.
|
||||
func TestRequireActiveModule_BlocksBeforeHandler(t *testing.T) {
|
||||
registry, flagStore, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
name := uniqueModuleName(t)
|
||||
flagKey := name + "_enabled"
|
||||
|
||||
if _, err := registry.Register(ctx, name, "1.0", []string{flagKey}); err != nil {
|
||||
t.Fatalf("register: %v", err)
|
||||
}
|
||||
|
||||
handlerReached := false
|
||||
handler := registry.RequireActiveModule(name, func(w http.ResponseWriter, r *http.Request) {
|
||||
handlerReached = true
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/modul?tenant=acme", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
handler(rec, req)
|
||||
if rec.Code != http.StatusForbidden {
|
||||
t.Fatalf("status = %d, want 403", rec.Code)
|
||||
}
|
||||
if handlerReached {
|
||||
t.Fatal("handler haette bei deaktiviertem modul NICHT erreicht werden duerfen")
|
||||
}
|
||||
|
||||
if err := flagStore.Set(ctx, flag.Flag{Key: flagKey, Enabled: true}); err != nil {
|
||||
t.Fatalf("flag setzen: %v", err)
|
||||
}
|
||||
req2 := httptest.NewRequest(http.MethodGet, "/modul?tenant=acme", nil)
|
||||
rec2 := httptest.NewRecorder()
|
||||
handler(rec2, req2)
|
||||
if rec2.Code != http.StatusOK {
|
||||
t.Fatalf("status nach aktivierung = %d, want 200", rec2.Code)
|
||||
}
|
||||
if !handlerReached {
|
||||
t.Fatal("handler haette bei aktiviertem modul erreicht werden muessen")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 4 + Pruefung 4: gueltiges/ungueltiges Service-Credential.
|
||||
func TestProvisionAndAuthenticate(t *testing.T) {
|
||||
registry, _, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
name := uniqueModuleName(t)
|
||||
|
||||
if _, err := registry.Register(ctx, name, "1.0", nil); err != nil {
|
||||
t.Fatalf("register: %v", err)
|
||||
}
|
||||
|
||||
clientID, secret, err := registry.Provision(ctx, name)
|
||||
if err != nil {
|
||||
t.Fatalf("provision: %v", err)
|
||||
}
|
||||
if clientID == "" || secret == "" {
|
||||
t.Fatal("erwartet nicht-leere client-id/secret")
|
||||
}
|
||||
|
||||
moduleName, ok, err := registry.Authenticate(ctx, clientID, secret)
|
||||
if err != nil {
|
||||
t.Fatalf("authenticate (korrekt): %v", err)
|
||||
}
|
||||
if !ok || moduleName != name {
|
||||
t.Fatalf("erwartet erfolgreiche authentifizierung fuer %q, habe ok=%v moduleName=%q", name, ok, moduleName)
|
||||
}
|
||||
|
||||
_, ok, err = registry.Authenticate(ctx, clientID, "falsches-secret")
|
||||
if err != nil {
|
||||
t.Fatalf("authenticate (falsch): %v", err)
|
||||
}
|
||||
if ok {
|
||||
t.Fatal("erwartet fehlschlag bei falschem secret")
|
||||
}
|
||||
|
||||
_, ok, err = registry.Authenticate(ctx, "unbekannte-client-id", secret)
|
||||
if err != nil {
|
||||
t.Fatalf("authenticate (unbekannt): %v", err)
|
||||
}
|
||||
if ok {
|
||||
t.Fatal("erwartet fehlschlag bei unbekannter client-id")
|
||||
}
|
||||
}
|
||||
|
||||
func TestProvision_RequiresRegisteredModule(t *testing.T) {
|
||||
registry, _, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, _, err := registry.Provision(ctx, "nie-registriert"); !errors.Is(err, ErrModuleNotRegistered) {
|
||||
t.Fatalf("erwartet ErrModuleNotRegistered, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequireServiceCredential_RejectsInvalidAcceptsValid(t *testing.T) {
|
||||
registry, _, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
name := uniqueModuleName(t)
|
||||
|
||||
if _, err := registry.Register(ctx, name, "1.0", nil); err != nil {
|
||||
t.Fatalf("register: %v", err)
|
||||
}
|
||||
clientID, secret, err := registry.Provision(ctx, name)
|
||||
if err != nil {
|
||||
t.Fatalf("provision: %v", err)
|
||||
}
|
||||
|
||||
handler := registry.RequireServiceCredential(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
|
||||
// Fehlendes Credential.
|
||||
req := httptest.NewRequest(http.MethodPost, "/service-aufruf", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
handler(rec, req)
|
||||
if rec.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("ohne credential: status = %d, want 401", rec.Code)
|
||||
}
|
||||
|
||||
// Falsches Secret.
|
||||
req2 := httptest.NewRequest(http.MethodPost, "/service-aufruf", nil)
|
||||
req2.Header.Set("X-Client-Id", clientID)
|
||||
req2.Header.Set("X-Client-Secret", "falsch")
|
||||
rec2 := httptest.NewRecorder()
|
||||
handler(rec2, req2)
|
||||
if rec2.Code != http.StatusUnauthorized {
|
||||
t.Fatalf("falsches secret: status = %d, want 401", rec2.Code)
|
||||
}
|
||||
|
||||
// Gueltiges Credential.
|
||||
req3 := httptest.NewRequest(http.MethodPost, "/service-aufruf", nil)
|
||||
req3.Header.Set("X-Client-Id", clientID)
|
||||
req3.Header.Set("X-Client-Secret", secret)
|
||||
rec3 := httptest.NewRecorder()
|
||||
handler(rec3, req3)
|
||||
if rec3.Code != http.StatusOK {
|
||||
t.Fatalf("gueltiges credential: status = %d, want 200", rec3.Code)
|
||||
}
|
||||
}
|
||||
@@ -1,81 +0,0 @@
|
||||
// Package notify implementiert Core CFG-02: den zentralen Benachrichtigungs-
|
||||
// Dispatcher, ueber den beliebige Module Benachrichtigungen ausloesen —
|
||||
// Warteschlange, Wiederholungslogik, Kanal-Abstraktion. Die tatsaechlichen
|
||||
// Kanaele (E-Mail/In-App) sind CFG-03, hier gibt es nur die Sender-
|
||||
// Schnittstelle als Vorbereitung.
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// DefaultMaxAttempts begrenzt Wiederholungsversuche (Akzeptanzkriterium 2) —
|
||||
// nach dieser Anzahl gibt der Dispatcher kontrolliert auf (status=failed)
|
||||
// statt endlos zu wiederholen.
|
||||
const DefaultMaxAttempts = 5
|
||||
|
||||
// DefaultRetryBackoff ist die Basis-Wartezeit zwischen Wiederholungen,
|
||||
// linear mit der Versuchsnummer skaliert.
|
||||
const DefaultRetryBackoff = 200 * time.Millisecond
|
||||
|
||||
type Notification struct {
|
||||
ID string
|
||||
Channel string
|
||||
Recipient string
|
||||
Payload map[string]any
|
||||
Attempts int
|
||||
}
|
||||
|
||||
// Sender ist die schmale Schnittstelle, die ein konkreter Kanal (CFG-03)
|
||||
// implementiert. Der Dispatcher selbst weiss nichts ueber E-Mail/In-App.
|
||||
type Sender interface {
|
||||
Send(ctx context.Context, n Notification) error
|
||||
}
|
||||
|
||||
// Dispatcher ist die EINE Schnittstelle, ueber die Module Benachrichtigungen
|
||||
// ausloesen — kein Modul baut eigenen Versandcode (Akzeptanzkriterium 1).
|
||||
type Dispatcher struct {
|
||||
pool *pgxpool.Pool
|
||||
maxAttempts int
|
||||
retryBackoff time.Duration
|
||||
}
|
||||
|
||||
func NewDispatcher(pool *pgxpool.Pool) *Dispatcher {
|
||||
return &Dispatcher{pool: pool, maxAttempts: DefaultMaxAttempts, retryBackoff: DefaultRetryBackoff}
|
||||
}
|
||||
|
||||
// WithRetryPolicy erlaubt Tests/Betrieb, Versuchsanzahl und Backoff
|
||||
// anzupassen, ohne die Default-Policy im Produktionscode zu veraendern.
|
||||
func (d *Dispatcher) WithRetryPolicy(maxAttempts int, backoff time.Duration) *Dispatcher {
|
||||
return &Dispatcher{pool: d.pool, maxAttempts: maxAttempts, retryBackoff: backoff}
|
||||
}
|
||||
|
||||
// Enqueue reiht eine Benachrichtigung in die Postgres-Warteschlange ein und
|
||||
// kehrt sofort zurueck — die Zeile ueberlebt jeden Neustart des Dispatcher-
|
||||
// Prozesses unveraendert (Akzeptanzkriterium 3), da sie ausschliesslich in
|
||||
// der Datenbank existiert, nicht im Prozessspeicher.
|
||||
func (d *Dispatcher) Enqueue(ctx context.Context, channel, recipient string, payload map[string]any) (string, error) {
|
||||
if payload == nil {
|
||||
payload = map[string]any{}
|
||||
}
|
||||
payloadJSON, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("payload serialisieren: %w", err)
|
||||
}
|
||||
|
||||
var id string
|
||||
err = d.pool.QueryRow(ctx, `
|
||||
INSERT INTO notification_jobs (channel, recipient, payload, max_attempts)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
RETURNING id
|
||||
`, channel, recipient, payloadJSON, d.maxAttempts).Scan(&id)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("benachrichtigung einreihen: %w", err)
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
@@ -1,219 +0,0 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*pgxpool.Pool, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS notification_jobs (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
channel TEXT NOT NULL,
|
||||
recipient TEXT NOT NULL,
|
||||
payload JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')),
|
||||
attempts INT NOT NULL DEFAULT 0,
|
||||
max_attempts INT NOT NULL DEFAULT 5,
|
||||
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
)`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() { pool.Close() }
|
||||
return pool, cleanup
|
||||
}
|
||||
|
||||
type fakeSender struct {
|
||||
mu sync.Mutex
|
||||
sentIDs []string
|
||||
failUntil int
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeSender) Send(ctx context.Context, n Notification) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.calls++
|
||||
if f.calls <= f.failUntil {
|
||||
return errors.New("simulierter zustellfehler")
|
||||
}
|
||||
f.sentIDs = append(f.sentIDs, n.ID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeSender) sentCount() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.sentIDs)
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: Module loesen ueber Enqueue aus, keine eigene
|
||||
// Versandlogik noetig.
|
||||
func TestDispatcher_EnqueueAndProcess(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
d := NewDispatcher(pool)
|
||||
id, err := d.Enqueue(ctx, "email", "alice@example.com", map[string]any{"subject": "Willkommen"})
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
if id == "" {
|
||||
t.Fatal("erwartet nicht-leere id")
|
||||
}
|
||||
|
||||
sender := &fakeSender{}
|
||||
sent, failed, err := d.ProcessDue(ctx, sender, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("process: %v", err)
|
||||
}
|
||||
if sent != 1 || failed != 0 {
|
||||
t.Fatalf("erwartet sent=1 failed=0, habe sent=%d failed=%d", sent, failed)
|
||||
}
|
||||
if sender.sentCount() != 1 {
|
||||
t.Fatalf("erwartet 1 zustellung, habe %d", sender.sentCount())
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Wiederholungslogik greift bei
|
||||
// simuliertem Fehler und bricht nach definierter Anzahl kontrolliert ab.
|
||||
func TestProcessDue_RetriesThenGivesUpAfterMaxAttempts(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
d := NewDispatcher(pool).WithRetryPolicy(3, time.Millisecond)
|
||||
id, err := d.Enqueue(ctx, "email", "bob@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
|
||||
sender := &fakeSender{failUntil: 100} // schlaegt bei jedem versuch fehl
|
||||
|
||||
for i := 0; i < 3; i++ {
|
||||
time.Sleep(5 * time.Millisecond) // next_attempt_at abwarten
|
||||
if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil {
|
||||
t.Fatalf("process %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
var status string
|
||||
var attempts int
|
||||
if err := pool.QueryRow(ctx, `SELECT status, attempts FROM notification_jobs WHERE id = $1`, id).Scan(&status, &attempts); err != nil {
|
||||
t.Fatalf("status lesen: %v", err)
|
||||
}
|
||||
if status != "failed" {
|
||||
t.Fatalf("erwartet status failed nach max_attempts, habe %q", status)
|
||||
}
|
||||
if attempts != 3 {
|
||||
t.Fatalf("erwartet 3 versuche, habe %d", attempts)
|
||||
}
|
||||
|
||||
// Weiteres ProcessDue darf den bereits aufgegebenen job nicht mehr anfassen.
|
||||
sent, failed, err := d.ProcessDue(ctx, sender, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("process nach abbruch: %v", err)
|
||||
}
|
||||
if sent != 0 || failed != 0 {
|
||||
t.Fatalf("erwartet keine weitere verarbeitung, habe sent=%d failed=%d", sent, failed)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 1: Neustart des Dienstes waehrend offener
|
||||
// Zustellung verliert keine Nachricht — simuliert durch eine komplett neue
|
||||
// Dispatcher/Pool-Instanz nach dem Enqueue, bevor irgendetwas verarbeitet wurde.
|
||||
func TestQueue_SurvivesRestartWithoutMessageLoss(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
firstInstance := NewDispatcher(pool)
|
||||
id, err := firstInstance.Enqueue(ctx, "email", "carol@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
|
||||
// "Neustart": eine voellig neue Dispatcher-Instanz (repraesentiert einen
|
||||
// neuen Prozess) verbindet sich neu und verarbeitet die Warteschlange —
|
||||
// die Nachricht existiert ausschliesslich in Postgres, nicht im
|
||||
// Prozessspeicher der ersten Instanz.
|
||||
restartedInstance := NewDispatcher(pool)
|
||||
sender := &fakeSender{}
|
||||
sent, failed, err := restartedInstance.ProcessDue(ctx, sender, 10)
|
||||
if err != nil {
|
||||
t.Fatalf("process nach neustart: %v", err)
|
||||
}
|
||||
if sent != 1 || failed != 0 {
|
||||
t.Fatalf("erwartet sent=1 nach neustart, habe sent=%d failed=%d", sent, failed)
|
||||
}
|
||||
if len(sender.sentIDs) != 1 || sender.sentIDs[0] != id {
|
||||
t.Fatalf("erwartet zustellung der urspruenglichen nachricht %q, habe %v", id, sender.sentIDs)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 3: zwei gleichzeitig ausloesende Module,
|
||||
// beide Nachrichten werden korrekt (und nicht doppelt) zugestellt.
|
||||
func TestProcessDue_ConcurrentDispatchBothDelivered(t *testing.T) {
|
||||
pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
d := NewDispatcher(pool)
|
||||
idA, err := d.Enqueue(ctx, "email", "modul-a@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue a: %v", err)
|
||||
}
|
||||
idB, err := d.Enqueue(ctx, "email", "modul-b@example.com", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue b: %v", err)
|
||||
}
|
||||
|
||||
sender := &fakeSender{}
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 2; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil {
|
||||
t.Errorf("process: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
if sender.sentCount() != 2 {
|
||||
t.Fatalf("erwartet genau 2 zustellungen, habe %d: %v", sender.sentCount(), sender.sentIDs)
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, id := range sender.sentIDs {
|
||||
if seen[id] {
|
||||
t.Fatalf("nachricht %q wurde doppelt zugestellt", id)
|
||||
}
|
||||
seen[id] = true
|
||||
}
|
||||
if !seen[idA] || !seen[idB] {
|
||||
t.Fatalf("erwartet beide nachrichten zugestellt, habe %v", sender.sentIDs)
|
||||
}
|
||||
}
|
||||
@@ -1,102 +0,0 @@
|
||||
package notify
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ProcessDue holt bis zu limit faellige Benachrichtigungen und versucht sie
|
||||
// ueber sender zuzustellen. FOR UPDATE SKIP LOCKED serialisiert konkurrierende
|
||||
// Aufrufe (Akzeptanzkriterium 3 / Pruefung 3: zwei gleichzeitig ausloesende
|
||||
// Module duerfen sich nicht gegenseitig blockieren oder Nachrichten doppelt
|
||||
// zustellen) — dieselbe Konvention wie internal/tenant.Lifecycle.ProcessDueDeletions.
|
||||
func (d *Dispatcher) ProcessDue(ctx context.Context, sender Sender, limit int) (sent, failed int, err error) {
|
||||
tx, err := d.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
rows, err := tx.Query(ctx, `
|
||||
SELECT id, channel, recipient, payload, attempts, max_attempts
|
||||
FROM notification_jobs
|
||||
WHERE status = 'pending' AND next_attempt_at <= now()
|
||||
ORDER BY created_at
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT $1
|
||||
`, limit)
|
||||
if err != nil {
|
||||
return 0, 0, fmt.Errorf("faellige benachrichtigungen abfragen: %w", err)
|
||||
}
|
||||
|
||||
type due struct {
|
||||
id, channel, recipient string
|
||||
payload []byte
|
||||
attempts, maxAttempts int
|
||||
}
|
||||
var candidates []due
|
||||
for rows.Next() {
|
||||
var c due
|
||||
if err := rows.Scan(&c.id, &c.channel, &c.recipient, &c.payload, &c.attempts, &c.maxAttempts); err != nil {
|
||||
rows.Close()
|
||||
return 0, 0, fmt.Errorf("faellige benachrichtigung lesen: %w", err)
|
||||
}
|
||||
candidates = append(candidates, c)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, 0, err
|
||||
}
|
||||
|
||||
for _, c := range candidates {
|
||||
var payload map[string]any
|
||||
if err := json.Unmarshal(c.payload, &payload); err != nil {
|
||||
payload = map[string]any{}
|
||||
}
|
||||
|
||||
sendErr := sender.Send(ctx, Notification{
|
||||
ID: c.id, Channel: c.channel, Recipient: c.recipient, Payload: payload, Attempts: c.attempts,
|
||||
})
|
||||
|
||||
if sendErr == nil {
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE notification_jobs SET status = 'sent', updated_at = now() WHERE id = $1
|
||||
`, c.id); err != nil {
|
||||
return sent, failed, fmt.Errorf("erfolg speichern: %w", err)
|
||||
}
|
||||
sent++
|
||||
continue
|
||||
}
|
||||
|
||||
newAttempts := c.attempts + 1
|
||||
if newAttempts >= c.maxAttempts {
|
||||
// Akzeptanzkriterium 2: kontrollierter Abbruch nach definierter
|
||||
// Anzahl Versuche, kein endloses Wiederholen.
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE notification_jobs
|
||||
SET status = 'failed', attempts = $2, last_error = $3, updated_at = now()
|
||||
WHERE id = $1
|
||||
`, c.id, newAttempts, sendErr.Error()); err != nil {
|
||||
return sent, failed, fmt.Errorf("fehlschlag speichern: %w", err)
|
||||
}
|
||||
failed++
|
||||
continue
|
||||
}
|
||||
|
||||
nextAttempt := time.Now().Add(time.Duration(newAttempts) * d.retryBackoff)
|
||||
if _, err := tx.Exec(ctx, `
|
||||
UPDATE notification_jobs
|
||||
SET attempts = $2, next_attempt_at = $3, last_error = $4, updated_at = now()
|
||||
WHERE id = $1
|
||||
`, c.id, newAttempts, nextAttempt, sendErr.Error()); err != nil {
|
||||
return sent, failed, fmt.Errorf("wiederholung planen: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return 0, 0, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
return sent, failed, nil
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package policy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/rbac"
|
||||
)
|
||||
|
||||
// ErrDenied wird geliefert, wenn keine Regel role+permission erlaubt —
|
||||
// Default-Deny (Akzeptanzkriterium 2).
|
||||
var ErrDenied = errors.New("policy: zugriff verweigert")
|
||||
|
||||
// Enforcer ist die EINE zentrale Entscheidungs- und Durchsetzungsschicht
|
||||
// (Akzeptanzkriterium 1). Repository-/Query-Code ruft ausschliesslich Guard
|
||||
// bzw. GuardTenantScoped auf, nie eine Rohabfrage direkt.
|
||||
type Enforcer struct {
|
||||
store *Store
|
||||
}
|
||||
|
||||
func NewEnforcer(store *Store) *Enforcer {
|
||||
return &Enforcer{store: store}
|
||||
}
|
||||
|
||||
// Authorize entscheidet erlaubt/verboten — unabhaengig von jeder konkreten
|
||||
// Query, rein anhand der deklarativen Regeln (Akzeptanzkriterium 2: fuer
|
||||
// sich genommen testbar, ohne Anwendungslogik).
|
||||
func (e *Enforcer) Authorize(ctx context.Context, role rbac.Role, perm rbac.Permission) error {
|
||||
allowed, err := e.store.IsAllowed(ctx, role, perm)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !allowed {
|
||||
return fmt.Errorf("%w: rolle %q hat kein recht %q", ErrDenied, role, perm)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Guard ist die zentrale Enforcement-Funktion (Akzeptanzkriterium 1): query
|
||||
// wird NUR aufgerufen, wenn Authorize zustimmt. Es gibt keinen Weg, query
|
||||
// ausserhalb von Guard aufzurufen und trotzdem den Aufrufer als autorisiert
|
||||
// zu behandeln — die Autorisierungsentscheidung steht immer VOR dem
|
||||
// Datenzugriff, nie danach.
|
||||
func Guard[T any](ctx context.Context, e *Enforcer, role rbac.Role, perm rbac.Permission, query func(ctx context.Context) (T, error)) (T, error) {
|
||||
var zero T
|
||||
if err := e.Authorize(ctx, role, perm); err != nil {
|
||||
return zero, err
|
||||
}
|
||||
return query(ctx)
|
||||
}
|
||||
|
||||
// GuardTenantScoped erzwingt zusaetzlich, dass tenantSlug TEIL der Query-
|
||||
// Funktion selbst ist (Akzeptanzkriterium 3): der Funktionstyp verlangt,
|
||||
// dass die Repository-Implementierung tenantSlug in ihre eigene WHERE-
|
||||
// Klausel einbaut — ein nachgelagerter Filter auf dem Ergebnis (der
|
||||
// archivmail-Fehler aus "Bekannte Fehler vermeiden") ist mit dieser
|
||||
// Signatur nicht moeglich, da die Query-Funktion tenantSlug selbst
|
||||
// entgegennimmt und dafuer verantwortlich ist, statt ihn hinterher
|
||||
// anzuwenden.
|
||||
func GuardTenantScoped[T any](ctx context.Context, e *Enforcer, role rbac.Role, perm rbac.Permission, tenantSlug string, query func(ctx context.Context, tenantSlug string) (T, error)) (T, error) {
|
||||
var zero T
|
||||
if err := e.Authorize(ctx, role, perm); err != nil {
|
||||
return zero, err
|
||||
}
|
||||
return query(ctx, tenantSlug)
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
package policy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/rbac"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*Store, *Enforcer, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS policy_rules (
|
||||
role TEXT NOT NULL, permission TEXT NOT NULL,
|
||||
granted_by TEXT NOT NULL, granted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (role, permission)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS policy_rule_changes (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), role TEXT NOT NULL, permission TEXT NOT NULL,
|
||||
action TEXT NOT NULL CHECK (action IN ('grant','revoke')), actor TEXT NOT NULL,
|
||||
version INT NOT NULL, changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
store := NewStore(pool)
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM policy_rule_changes WHERE role LIKE 'test\_%' ESCAPE '\'`)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM policy_rules WHERE role LIKE 'test\_%' ESCAPE '\'`)
|
||||
pool.Close()
|
||||
}
|
||||
return store, NewEnforcer(store), cleanup
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Anfrage ohne passende Policy wird
|
||||
// zuverlaessig abgewiesen (Default-Deny), nicht standardmaessig erlaubt.
|
||||
func TestAuthorize_DefaultDeny(t *testing.T) {
|
||||
_, enforcer, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
err := enforcer.Authorize(ctx, rbac.Role("test_niemand"), rbac.Permission("test_irgendwas"))
|
||||
if !errors.Is(err, ErrDenied) {
|
||||
t.Fatalf("erwartet ErrDenied ohne konfigurierte regel, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: Guard ruft query NUR bei Autorisierung
|
||||
// auf — kein Datenzugriffs-Pfad umgeht die Enforcement-Schicht.
|
||||
func TestGuard_NeverCallsQueryWithoutAuthorization(t *testing.T) {
|
||||
store, enforcer, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
role := rbac.Role("test_rolle_guard")
|
||||
perm := rbac.Permission("test_lesen")
|
||||
|
||||
queryCalled := false
|
||||
_, err := Guard(ctx, enforcer, role, perm, func(ctx context.Context) (string, error) {
|
||||
queryCalled = true
|
||||
return "geheime-daten", nil
|
||||
})
|
||||
if !errors.Is(err, ErrDenied) {
|
||||
t.Fatalf("erwartet ErrDenied, habe %v", err)
|
||||
}
|
||||
if queryCalled {
|
||||
t.Fatal("query haette bei fehlender autorisierung NICHT aufgerufen werden duerfen")
|
||||
}
|
||||
|
||||
if err := store.Grant(ctx, role, perm, "admin@example.com"); err != nil {
|
||||
t.Fatalf("grant: %v", err)
|
||||
}
|
||||
|
||||
result, err := Guard(ctx, enforcer, role, perm, func(ctx context.Context) (string, error) {
|
||||
queryCalled = true
|
||||
return "geheime-daten", nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("guard nach grant: %v", err)
|
||||
}
|
||||
if !queryCalled || result != "geheime-daten" {
|
||||
t.Fatalf("erwartet query-aufruf mit ergebnis nach autorisierung, habe queryCalled=%v result=%q", queryCalled, result)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + 3: Policy-Entscheidungen deklarativ und unabhaengig
|
||||
// von Anwendungslogik testbar — Regelwechsel wirkt ohne Codeaenderung,
|
||||
// Historie ist versioniert nachvollziehbar.
|
||||
func TestGrantRevoke_ChangesBehaviorWithoutCodeChange(t *testing.T) {
|
||||
store, enforcer, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
role := rbac.Role("test_rolle_history")
|
||||
perm := rbac.Permission("test_schreiben")
|
||||
|
||||
allowed, err := store.IsAllowed(ctx, role, perm)
|
||||
if err != nil {
|
||||
t.Fatalf("is allowed (vorher): %v", err)
|
||||
}
|
||||
if allowed {
|
||||
t.Fatal("erwartet nicht erlaubt vor grant")
|
||||
}
|
||||
|
||||
if err := store.Grant(ctx, role, perm, "admin@example.com"); err != nil {
|
||||
t.Fatalf("grant: %v", err)
|
||||
}
|
||||
if err := enforcer.Authorize(ctx, role, perm); err != nil {
|
||||
t.Fatalf("authorize nach grant sollte erlauben: %v", err)
|
||||
}
|
||||
|
||||
if err := store.Revoke(ctx, role, perm, "admin@example.com"); err != nil {
|
||||
t.Fatalf("revoke: %v", err)
|
||||
}
|
||||
if err := enforcer.Authorize(ctx, role, perm); !errors.Is(err, ErrDenied) {
|
||||
t.Fatalf("authorize nach revoke sollte verweigern, habe %v", err)
|
||||
}
|
||||
|
||||
history, err := store.History(ctx, role, perm)
|
||||
if err != nil {
|
||||
t.Fatalf("history: %v", err)
|
||||
}
|
||||
if len(history) != 2 {
|
||||
t.Fatalf("erwartet 2 historieneintraege (grant, revoke), habe %d", len(history))
|
||||
}
|
||||
if history[0].Action != "grant" || history[0].Version != 1 {
|
||||
t.Fatalf("history[0] unerwartet: %+v", history[0])
|
||||
}
|
||||
if history[1].Action != "revoke" || history[1].Version != 2 {
|
||||
t.Fatalf("history[1] unerwartet: %+v", history[1])
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 3 (Tenant-Scoping): GuardTenantScoped
|
||||
// zwingt tenantSlug in die Query-Funktion selbst, kein Post-Filter moeglich.
|
||||
func TestGuardTenantScoped_IsolatesDataBetweenTenants(t *testing.T) {
|
||||
_, enforcer, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
// Simuliertes Repository: die WHERE-Klausel (hier: Map-Lookup) liegt
|
||||
// INNERHALB der Query-Funktion, nicht als nachgelagerter Filter.
|
||||
data := map[string][]string{
|
||||
"acme": {"dokument-a1", "dokument-a2"},
|
||||
"globex": {"dokument-b1"},
|
||||
}
|
||||
repoQuery := func(ctx context.Context, tenantSlug string) ([]string, error) {
|
||||
return data[tenantSlug], nil
|
||||
}
|
||||
|
||||
role := rbac.Role("test_tenant_rolle")
|
||||
perm := rbac.Permission("test_dokumente_lesen")
|
||||
if err := enforcer.store.Grant(ctx, role, perm, "admin@example.com"); err != nil {
|
||||
t.Fatalf("grant: %v", err)
|
||||
}
|
||||
|
||||
gotAcme, err := GuardTenantScoped(ctx, enforcer, role, perm, "acme", repoQuery)
|
||||
if err != nil {
|
||||
t.Fatalf("guard tenant scoped (acme): %v", err)
|
||||
}
|
||||
if len(gotAcme) != 2 {
|
||||
t.Fatalf("erwartet 2 dokumente fuer acme, habe %d", len(gotAcme))
|
||||
}
|
||||
|
||||
gotGlobex, err := GuardTenantScoped(ctx, enforcer, role, perm, "globex", repoQuery)
|
||||
if err != nil {
|
||||
t.Fatalf("guard tenant scoped (globex): %v", err)
|
||||
}
|
||||
if len(gotGlobex) != 1 {
|
||||
t.Fatalf("erwartet 1 dokument fuer globex, habe %d", len(gotGlobex))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
// Package policy implementiert Core RBAC-02: die zentrale Policy-
|
||||
// Entscheidungs- und Durchsetzungsschicht. Regeln liegen deklarativ in der
|
||||
// Datenbank (Casbin-Prinzip: Policy als Modell+Regeln getrennt vom Code),
|
||||
// versioniert und auditierbar — anders als Casbin-Dateien im Dateisystem
|
||||
// (siehe "bewusst vermeiden" im Ticket).
|
||||
package policy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/rbac"
|
||||
)
|
||||
|
||||
// Rule ist eine gewaehrte Regel: role darf permission.
|
||||
type Rule struct {
|
||||
Role rbac.Role
|
||||
Permission rbac.Permission
|
||||
}
|
||||
|
||||
// RuleChange ist ein Eintrag der Aenderungshistorie (Akzeptanzkriterium 3).
|
||||
type RuleChange struct {
|
||||
Role rbac.Role
|
||||
Permission rbac.Permission
|
||||
Action string // "grant" oder "revoke"
|
||||
Actor string
|
||||
Version int
|
||||
}
|
||||
|
||||
// Store verwaltet den deklarativen Regelsatz. IsAllowed schaut NUR in die
|
||||
// Datenbank — es gibt keinen hartcodierten Go-Fallback, ein Regelwechsel
|
||||
// wirkt sich ohne Codeaenderung sofort aus (Akzeptanzkriterium 2 / 3).
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// IsAllowed prueft, ob role das Recht permission besitzt. Default-Deny:
|
||||
// existiert keine passende Regel, ist der Zugriff verboten
|
||||
// (Akzeptanzkriterium 2 / Pruefung 2) — es gibt keinen impliziten
|
||||
// "erlaubt, wenn nichts anderes gesagt"-Pfad.
|
||||
func (s *Store) IsAllowed(ctx context.Context, role rbac.Role, perm rbac.Permission) (bool, error) {
|
||||
var exists bool
|
||||
err := s.pool.QueryRow(ctx, `
|
||||
SELECT EXISTS(SELECT 1 FROM policy_rules WHERE role = $1 AND permission = $2)
|
||||
`, string(role), string(perm)).Scan(&exists)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("policy pruefen: %w", err)
|
||||
}
|
||||
return exists, nil
|
||||
}
|
||||
|
||||
// Grant gewaehrt role das Recht permission — deklarativ, ohne Codeaenderung
|
||||
// wirksam. actor wird fuer die Aenderungshistorie festgehalten.
|
||||
func (s *Store) Grant(ctx context.Context, role rbac.Role, perm rbac.Permission, actor string) error {
|
||||
return s.change(ctx, role, perm, "grant", actor, `
|
||||
INSERT INTO policy_rules (role, permission, granted_by, granted_at)
|
||||
VALUES ($1, $2, $3, now())
|
||||
ON CONFLICT (role, permission) DO UPDATE SET granted_by = $3, granted_at = now()
|
||||
`, string(role), string(perm), actor)
|
||||
}
|
||||
|
||||
// Revoke entzieht role das Recht permission.
|
||||
func (s *Store) Revoke(ctx context.Context, role rbac.Role, perm rbac.Permission, actor string) error {
|
||||
return s.change(ctx, role, perm, "revoke", actor, `
|
||||
DELETE FROM policy_rules WHERE role = $1 AND permission = $2
|
||||
`, string(role), string(perm))
|
||||
}
|
||||
|
||||
func (s *Store) change(ctx context.Context, role rbac.Role, perm rbac.Permission, action, actor, sql string, args ...any) error {
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
if _, err := tx.Exec(ctx, sql, args...); err != nil {
|
||||
return fmt.Errorf("regel aendern: %w", err)
|
||||
}
|
||||
|
||||
var version int
|
||||
if err := tx.QueryRow(ctx, `
|
||||
SELECT COALESCE(MAX(version), 0) + 1 FROM policy_rule_changes WHERE role = $1 AND permission = $2
|
||||
`, string(role), string(perm)).Scan(&version); err != nil {
|
||||
return fmt.Errorf("naechste version ermitteln: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO policy_rule_changes (role, permission, action, actor, version, changed_at)
|
||||
VALUES ($1, $2, $3, $4, $5, now())
|
||||
`, string(role), string(perm), action, actor, version); err != nil {
|
||||
return fmt.Errorf("historie schreiben: %w", err)
|
||||
}
|
||||
|
||||
return tx.Commit(ctx)
|
||||
}
|
||||
|
||||
// History liefert die vollstaendige, versionierte Aenderungshistorie einer
|
||||
// Regel (Akzeptanzkriterium 3).
|
||||
func (s *Store) History(ctx context.Context, role rbac.Role, perm rbac.Permission) ([]RuleChange, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT action, actor, version FROM policy_rule_changes
|
||||
WHERE role = $1 AND permission = $2 ORDER BY version
|
||||
`, string(role), string(perm))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("historie abfragen: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []RuleChange
|
||||
for rows.Next() {
|
||||
c := RuleChange{Role: role, Permission: perm}
|
||||
if err := rows.Scan(&c.Action, &c.Actor, &c.Version); err != nil {
|
||||
return nil, fmt.Errorf("historieneintrag lesen: %w", err)
|
||||
}
|
||||
out = append(out, c)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,38 @@
|
||||
// Package policyapi implementiert RBAC-06: einen HTTP-Endpunkt, der
|
||||
// Core RBAC-02s Enforcer.Authorize für andere, physisch getrennte
|
||||
// Go-Module (DMS, Mail, Archive) zugänglich macht — dieselbe
|
||||
// Entscheidung wie der direkte Enforcer-Aufruf, nur per HTTP statt
|
||||
// In-Process.
|
||||
package policyapi
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// RequireServiceToken authentifiziert den Aufrufer service-zu-service
|
||||
// über einen geteilten Token (Umgebungsvariable, nie im Code — siehe
|
||||
// Ticket-Abschlussregel), timing-safe verglichen (dieselbe
|
||||
// Referenzimplementierung wie AUD-02: subtle.ConstantTimeCompare,
|
||||
// projektweite Konvention für jeden sicherheitsrelevanten Vergleich).
|
||||
// Bewusst KEIN vollwertiges Credential-Ausstellungssystem
|
||||
// (moduleregistry.Provision/Authenticate) — dieser Endpunkt braucht
|
||||
// keine Modul-Aktivierungsverwaltung, nur einen einzelnen, über
|
||||
// Konfiguration verteilten Zugangsschlüssel (Akzeptanzkriterium 2).
|
||||
func RequireServiceToken(expectedToken string, next http.HandlerFunc) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
got := r.Header.Get("X-Service-Token")
|
||||
if got == "" || !timingSafeEqual(got, expectedToken) {
|
||||
http.Error(w, "authentifizierung fehlgeschlagen: gültiges X-Service-Token erforderlich", http.StatusUnauthorized)
|
||||
return
|
||||
}
|
||||
next(w, r)
|
||||
}
|
||||
}
|
||||
|
||||
func timingSafeEqual(a, b string) bool {
|
||||
ha := sha256.Sum256([]byte(a))
|
||||
hb := sha256.Sum256([]byte(b))
|
||||
return subtle.ConstantTimeCompare(ha[:], hb[:]) == 1
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
package policyapi
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// TestRequireServiceToken_MissingOrWrongTokenReturns401 ist die
|
||||
// Pflichtpruefung: Aufruf ohne gueltige Service-Credentials wird
|
||||
// abgewiesen, nicht mit einer Policy-Entscheidung beantwortet.
|
||||
func TestRequireServiceToken_MissingOrWrongTokenReturns401(t *testing.T) {
|
||||
called := false
|
||||
handler := RequireServiceToken("richtiges-token", func(w http.ResponseWriter, r *http.Request) {
|
||||
called = true
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
server := httptest.NewServer(handler)
|
||||
defer server.Close()
|
||||
|
||||
resp, err := http.Get(server.URL)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode != http.StatusUnauthorized {
|
||||
t.Fatalf("ohne token: status = %d, want 401", resp.StatusCode)
|
||||
}
|
||||
if called {
|
||||
t.Fatal("handler haette bei fehlendem token nie erreicht werden duerfen")
|
||||
}
|
||||
|
||||
req, _ := http.NewRequest(http.MethodGet, server.URL, nil)
|
||||
req.Header.Set("X-Service-Token", "falsches-token")
|
||||
resp2, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
defer func() { _ = resp2.Body.Close() }()
|
||||
if resp2.StatusCode != http.StatusUnauthorized {
|
||||
t.Fatalf("mit falschem token: status = %d, want 401", resp2.StatusCode)
|
||||
}
|
||||
}
|
||||
|
||||
// TestRequireServiceToken_CorrectTokenAllowsAccess ist der Gegentest.
|
||||
func TestRequireServiceToken_CorrectTokenAllowsAccess(t *testing.T) {
|
||||
handler := RequireServiceToken("richtiges-token", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
server := httptest.NewServer(handler)
|
||||
defer server.Close()
|
||||
|
||||
req, _ := http.NewRequest(http.MethodGet, server.URL, nil)
|
||||
req.Header.Set("X-Service-Token", "richtiges-token")
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("mit korrektem token: status = %d, want 200", resp.StatusCode)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,56 @@
|
||||
package policyapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/policy"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/rbac"
|
||||
)
|
||||
|
||||
// Mount registriert den Autorisierungs-Endpunkt auf mux, hinter dem
|
||||
// Service-Token-Check (Akzeptanzkriterium 2).
|
||||
func Mount(mux *http.ServeMux, enforcer *policy.Enforcer, serviceToken string) {
|
||||
mux.HandleFunc("POST /authorize", RequireServiceToken(serviceToken, authorizeHandler(enforcer)))
|
||||
}
|
||||
|
||||
type authorizeRequest struct {
|
||||
Role string `json:"role"`
|
||||
Permission string `json:"permission"`
|
||||
}
|
||||
|
||||
type authorizeResponse struct {
|
||||
Allowed bool `json:"allowed"`
|
||||
}
|
||||
|
||||
// authorizeHandler ruft AUSSCHLIESSLICH enforcer.Authorize auf — keine
|
||||
// eigene, zweite Autorisierungsentscheidung (Akzeptanzkriterium 1: die
|
||||
// HTTP-Antwort MUSS mit dem direkten Enforcer-Ergebnis übereinstimmen,
|
||||
// das ist strukturell garantiert, weil es derselbe Aufruf ist, nicht
|
||||
// nur getestet).
|
||||
func authorizeHandler(enforcer *policy.Enforcer) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
var req authorizeRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "ungültiger request-body: "+err.Error(), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
if req.Role == "" || req.Permission == "" {
|
||||
http.Error(w, "role und permission sind pflichtfelder", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
err := enforcer.Authorize(r.Context(), rbac.Role(req.Role), rbac.Permission(req.Permission))
|
||||
allowed := err == nil
|
||||
if err != nil && !errors.Is(err, policy.ErrDenied) {
|
||||
// Ein anderer Fehler als "verboten" (z.B. DB-Fehler) ist kein
|
||||
// gueltiges Allow/Deny-Ergebnis, sondern ein echter Fehlerfall.
|
||||
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(authorizeResponse{Allowed: allowed})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package policyapi
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/policy"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/rbac"
|
||||
)
|
||||
|
||||
const testToken = "test-service-token"
|
||||
|
||||
func setupTest(t *testing.T) *policy.Enforcer {
|
||||
t.Helper()
|
||||
dsn := os.Getenv("TEST_ADMIN_DSN")
|
||||
if dsn == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { pool.Close() })
|
||||
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
||||
CREATE TABLE IF NOT EXISTS policy_rules (
|
||||
role TEXT NOT NULL, permission TEXT NOT NULL,
|
||||
granted_by TEXT NOT NULL, granted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (role, permission)
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS policy_rule_changes (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), role TEXT NOT NULL, permission TEXT NOT NULL,
|
||||
action TEXT NOT NULL CHECK (action IN ('grant','revoke')), actor TEXT NOT NULL,
|
||||
version INT NOT NULL, changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
_, _ = pool.Exec(context.Background(), `DELETE FROM policy_rule_changes WHERE role LIKE 'rbac06\_%' ESCAPE '\'`)
|
||||
_, _ = pool.Exec(context.Background(), `DELETE FROM policy_rules WHERE role LIKE 'rbac06\_%' ESCAPE '\'`)
|
||||
})
|
||||
|
||||
store := policy.NewStore(pool)
|
||||
if err := store.Grant(ctx, rbac.Role("rbac06_erlaubte_rolle"), rbac.Permission("rbac06_recht"), "test"); err != nil {
|
||||
t.Fatalf("grant: %v", err)
|
||||
}
|
||||
return policy.NewEnforcer(store)
|
||||
}
|
||||
|
||||
func authorize(t *testing.T, server *httptest.Server, role, permission string) (int, bool) {
|
||||
t.Helper()
|
||||
body, _ := json.Marshal(authorizeRequest{Role: role, Permission: permission})
|
||||
req, _ := http.NewRequest(http.MethodPost, server.URL+"/authorize", bytes.NewReader(body))
|
||||
req.Header.Set("X-Service-Token", testToken)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("post: %v", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return resp.StatusCode, false
|
||||
}
|
||||
var out authorizeResponse
|
||||
if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
|
||||
t.Fatalf("antwort dekodieren: %v", err)
|
||||
}
|
||||
return resp.StatusCode, out.Allowed
|
||||
}
|
||||
|
||||
// TestAuthorizeHandler_MatchesDirectEnforcerResult ist die geforderte
|
||||
// Pflichtpruefung: Endpunkt-Antwort stimmt in 100 Stichproben exakt mit
|
||||
// dem direkten Enforcer.Authorize-Ergebnis ueberein - real fuer sowohl
|
||||
// erlaubte als auch verbotene role+permission-Kombinationen.
|
||||
func TestAuthorizeHandler_MatchesDirectEnforcerResult(t *testing.T) {
|
||||
enforcer := setupTest(t)
|
||||
mux := http.NewServeMux()
|
||||
Mount(mux, enforcer, testToken)
|
||||
server := httptest.NewServer(mux)
|
||||
defer server.Close()
|
||||
|
||||
cases := []struct {
|
||||
role, permission string
|
||||
}{
|
||||
{"rbac06_erlaubte_rolle", "rbac06_recht"}, // erlaubt
|
||||
{"rbac06_unbekannte_rolle", "rbac06_recht"}, // verboten (default-deny)
|
||||
{"rbac06_erlaubte_rolle", "rbac06_anderes"}, // verboten (falsches recht)
|
||||
}
|
||||
|
||||
for i := 0; i < 100; i++ {
|
||||
c := cases[i%len(cases)]
|
||||
directErr := enforcer.Authorize(context.Background(), rbac.Role(c.role), rbac.Permission(c.permission))
|
||||
directAllowed := directErr == nil
|
||||
if directErr != nil && !errors.Is(directErr, policy.ErrDenied) {
|
||||
t.Fatalf("stichprobe %d: unerwarteter enforcer-fehler: %v", i, directErr)
|
||||
}
|
||||
|
||||
status, httpAllowed := authorize(t, server, c.role, c.permission)
|
||||
if status != http.StatusOK {
|
||||
t.Fatalf("stichprobe %d: unerwarteter status %d", i, status)
|
||||
}
|
||||
if httpAllowed != directAllowed {
|
||||
t.Fatalf("stichprobe %d (%s/%s): http=%t, direkt=%t - weichen ab", i, c.role, c.permission, httpAllowed, directAllowed)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,71 @@
|
||||
// Package rbac implementiert Core RBAC-01: das Rollenmodell mit
|
||||
// Grundrollen/-rechten, Hierarchie und Zuweisung. Die eigentliche
|
||||
// Durchsetzung (erlaubt/verboten je Request) ist RBAC-02, dieses Paket
|
||||
// liefert nur Modell + Zuweisung (Casbin-Prinzip: Policy-Modell getrennt
|
||||
// von der Entscheidungsfunktion).
|
||||
package rbac
|
||||
|
||||
// Role ist eine der drei Grundrollen. Superadmin lebt mandantenuebergreifend
|
||||
// (siehe internal/user.SuperadminStore aus IAM-01) und wird deshalb NICHT
|
||||
// ueber Store (tenant-gescoped) zugewiesen — die Existenz eines Superadmin-
|
||||
// Kontos IST die Rollenzuweisung.
|
||||
type Role string
|
||||
|
||||
const (
|
||||
RoleSuperadmin Role = "superadmin"
|
||||
RoleTenantAdmin Role = "tenant_admin"
|
||||
RoleUser Role = "user"
|
||||
)
|
||||
|
||||
type Permission string
|
||||
|
||||
const (
|
||||
PermSelfRead Permission = "self.read"
|
||||
PermSelfUpdate Permission = "self.update"
|
||||
PermManageUsers Permission = "tenant.manage_users"
|
||||
PermManageSettings Permission = "tenant.manage_settings"
|
||||
PermManageTenants Permission = "platform.manage_tenants"
|
||||
)
|
||||
|
||||
// parent bildet die Rollenhierarchie ab (Akzeptanzkriterium 2): eine Rolle
|
||||
// erbt alle Rechte ihrer Elternrolle. tenant_admin erbt von user,
|
||||
// superadmin erbt von tenant_admin.
|
||||
var parent = map[Role]Role{
|
||||
RoleTenantAdmin: RoleUser,
|
||||
RoleSuperadmin: RoleTenantAdmin,
|
||||
}
|
||||
|
||||
// direct sind die einer Rolle direkt (ohne Vererbung) zugeordneten Rechte.
|
||||
var direct = map[Role][]Permission{
|
||||
RoleUser: {PermSelfRead, PermSelfUpdate},
|
||||
RoleTenantAdmin: {PermManageUsers, PermManageSettings},
|
||||
RoleSuperadmin: {PermManageTenants},
|
||||
}
|
||||
|
||||
// EffectivePermissions liefert die vollstaendige, ueber die Hierarchie
|
||||
// aufgeloeste Rechtemenge einer Rolle (Akzeptanzkriterium 2 / Pruefung 2).
|
||||
func EffectivePermissions(role Role) []Permission {
|
||||
seen := make(map[Permission]bool)
|
||||
var out []Permission
|
||||
|
||||
for r, ok := role, true; ok; r, ok = parent[r] {
|
||||
for _, p := range direct[r] {
|
||||
if !seen[p] {
|
||||
seen[p] = true
|
||||
out = append(out, p)
|
||||
}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// HasPermission prueft, ob eine Rolle (inklusive geerbter Rechte) ein
|
||||
// bestimmtes Recht besitzt.
|
||||
func HasPermission(role Role, perm Permission) bool {
|
||||
for _, p := range EffectivePermissions(role) {
|
||||
if p == perm {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package rbac
|
||||
|
||||
import "testing"
|
||||
|
||||
func contains(perms []Permission, p Permission) bool {
|
||||
for _, x := range perms {
|
||||
if x == p {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Rollenhierarchie liefert die erwartete
|
||||
// effektive Rechtemenge.
|
||||
func TestEffectivePermissions_Inheritance(t *testing.T) {
|
||||
userPerms := EffectivePermissions(RoleUser)
|
||||
if !contains(userPerms, PermSelfRead) || !contains(userPerms, PermSelfUpdate) {
|
||||
t.Fatalf("user sollte self.read/self.update haben, habe %v", userPerms)
|
||||
}
|
||||
if contains(userPerms, PermManageUsers) {
|
||||
t.Fatal("user sollte KEIN tenant.manage_users haben")
|
||||
}
|
||||
|
||||
adminPerms := EffectivePermissions(RoleTenantAdmin)
|
||||
if !contains(adminPerms, PermSelfRead) || !contains(adminPerms, PermManageUsers) || !contains(adminPerms, PermManageSettings) {
|
||||
t.Fatalf("tenant_admin sollte geerbte user-rechte + eigene rechte haben, habe %v", adminPerms)
|
||||
}
|
||||
if contains(adminPerms, PermManageTenants) {
|
||||
t.Fatal("tenant_admin sollte KEIN platform.manage_tenants haben")
|
||||
}
|
||||
|
||||
superPerms := EffectivePermissions(RoleSuperadmin)
|
||||
for _, want := range []Permission{PermSelfRead, PermSelfUpdate, PermManageUsers, PermManageSettings, PermManageTenants} {
|
||||
if !contains(superPerms, want) {
|
||||
t.Fatalf("superadmin sollte %q haben (volle vererbte kette), habe %v", want, superPerms)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestHasPermission(t *testing.T) {
|
||||
if !HasPermission(RoleTenantAdmin, PermSelfRead) {
|
||||
t.Fatal("tenant_admin sollte geerbtes self.read haben")
|
||||
}
|
||||
if HasPermission(RoleUser, PermManageTenants) {
|
||||
t.Fatal("user sollte platform.manage_tenants nicht haben")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package rbac
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
var (
|
||||
// ErrRoleNotAssignableInTenantScope wird geliefert, wenn versucht wird,
|
||||
// eine mandantenuebergreifende Rolle (superadmin) ueber den tenant-
|
||||
// gescopten Store zu vergeben — die erlaubte Matrix laesst hier nur
|
||||
// user/tenant_admin zu (Akzeptanzkriterium 1 / Pruefung 1).
|
||||
ErrRoleNotAssignableInTenantScope = errors.New("rbac: rolle ist in diesem geltungsbereich nicht zuweisbar")
|
||||
ErrNotFound = errors.New("rbac: keine rollenzuweisung gefunden")
|
||||
)
|
||||
|
||||
// assignableRoles ist die erlaubte Matrix fuer Store (tenant-gescoped).
|
||||
var assignableRoles = map[Role]bool{
|
||||
RoleUser: true,
|
||||
RoleTenantAdmin: true,
|
||||
}
|
||||
|
||||
type Assignment struct {
|
||||
UserID string
|
||||
Role Role
|
||||
GrantedBy string
|
||||
}
|
||||
|
||||
// Store verwaltet Rollenzuweisungen innerhalb GENAU EINER Tenant-Datenbank —
|
||||
// analog zu internal/user.TenantUserStore (Modell C: der Pool bestimmt den
|
||||
// Tenant, keine tenant_id-Spalte noetig).
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// Assign vergibt eine Rolle an einen Benutzer. grantedBy identifiziert den
|
||||
// Akteur, der die Zuweisung vorgenommen hat (Akzeptanzkriterium 3). Jede
|
||||
// Zuweisung wird zusaetzlich in role_assignment_history festgehalten, auch
|
||||
// wenn sie eine vorherige Rolle ersetzt.
|
||||
func (s *Store) Assign(ctx context.Context, userID string, role Role, grantedBy string) (Assignment, error) {
|
||||
if !assignableRoles[role] {
|
||||
return Assignment{}, ErrRoleNotAssignableInTenantScope
|
||||
}
|
||||
if grantedBy == "" {
|
||||
return Assignment{}, errors.New("rbac: grantedBy darf nicht leer sein")
|
||||
}
|
||||
|
||||
tx, err := s.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return Assignment{}, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO role_assignments (user_id, role, granted_by, granted_at)
|
||||
VALUES ($1, $2, $3, now())
|
||||
ON CONFLICT (user_id) DO UPDATE SET role = $2, granted_by = $3, granted_at = now()
|
||||
`, userID, string(role), grantedBy); err != nil {
|
||||
return Assignment{}, fmt.Errorf("rolle zuweisen: %w", err)
|
||||
}
|
||||
|
||||
if _, err := tx.Exec(ctx, `
|
||||
INSERT INTO role_assignment_history (user_id, role, granted_by, granted_at)
|
||||
VALUES ($1, $2, $3, now())
|
||||
`, userID, string(role), grantedBy); err != nil {
|
||||
return Assignment{}, fmt.Errorf("historie schreiben: %w", err)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return Assignment{}, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
|
||||
return Assignment{UserID: userID, Role: role, GrantedBy: grantedBy}, nil
|
||||
}
|
||||
|
||||
func (s *Store) Get(ctx context.Context, userID string) (Assignment, error) {
|
||||
var a Assignment
|
||||
var role string
|
||||
a.UserID = userID
|
||||
if err := s.pool.QueryRow(ctx, `
|
||||
SELECT role, granted_by FROM role_assignments WHERE user_id = $1
|
||||
`, userID).Scan(&role, &a.GrantedBy); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Assignment{}, ErrNotFound
|
||||
}
|
||||
return Assignment{}, fmt.Errorf("rollenzuweisung lesen: %w", err)
|
||||
}
|
||||
a.Role = Role(role)
|
||||
return a, nil
|
||||
}
|
||||
|
||||
// History liefert jede Rollenzuweisung eines Benutzers in chronologischer
|
||||
// Reihenfolge — die Grundlage fuer "wer hat wann welche Rolle vergeben"
|
||||
// (Akzeptanzkriterium 3).
|
||||
func (s *Store) History(ctx context.Context, userID string) ([]Assignment, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT role, granted_by FROM role_assignment_history
|
||||
WHERE user_id = $1 ORDER BY granted_at
|
||||
`, userID)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("historie abfragen: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []Assignment
|
||||
for rows.Next() {
|
||||
var role string
|
||||
a := Assignment{UserID: userID}
|
||||
if err := rows.Scan(&role, &a.GrantedBy); err != nil {
|
||||
return nil, fmt.Errorf("historieneintrag lesen: %w", err)
|
||||
}
|
||||
a.Role = Role(role)
|
||||
out = append(out, a)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,144 @@
|
||||
package rbac
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/user"
|
||||
)
|
||||
|
||||
func setupStoreTest(t *testing.T) (*Store, string, func()) {
|
||||
t.Helper()
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
pool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("pool: %v", err)
|
||||
}
|
||||
if _, err := pool.Exec(ctx, `
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
email TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS role_assignments (
|
||||
user_id UUID PRIMARY KEY REFERENCES users(id),
|
||||
role TEXT NOT NULL CHECK (role IN ('user', 'tenant_admin')),
|
||||
granted_by TEXT NOT NULL,
|
||||
granted_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS role_assignment_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id UUID NOT NULL,
|
||||
role TEXT NOT NULL,
|
||||
granted_by TEXT NOT NULL,
|
||||
granted_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
userStore := user.NewTenantUserStore(pool)
|
||||
u, err := userStore.Create(ctx, "rbac_test@example.com", "RBAC Test")
|
||||
if err != nil {
|
||||
t.Fatalf("testuser anlegen: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() {
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM role_assignment_history WHERE user_id = $1`, u.ID)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM role_assignments WHERE user_id = $1`, u.ID)
|
||||
_, _ = pool.Exec(ctx, `DELETE FROM users WHERE id = $1`, u.ID)
|
||||
pool.Close()
|
||||
}
|
||||
return NewStore(pool), u.ID, cleanup
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: Grundrollen sind einem Benutzer je Tenant zuweisbar.
|
||||
func TestStore_AssignAndGet(t *testing.T) {
|
||||
store, userID, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
assigned, err := store.Assign(ctx, userID, RoleTenantAdmin, "admin@acme.example")
|
||||
if err != nil {
|
||||
t.Fatalf("assign: %v", err)
|
||||
}
|
||||
if assigned.Role != RoleTenantAdmin {
|
||||
t.Fatalf("erwartet tenant_admin, habe %q", assigned.Role)
|
||||
}
|
||||
|
||||
got, err := store.Get(ctx, userID)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.Role != RoleTenantAdmin || got.GrantedBy != "admin@acme.example" {
|
||||
t.Fatalf("zuweisung unerwartet: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: Zuweisung ausserhalb der erlaubten
|
||||
// Matrix (superadmin im tenant-gescopten Store) wird abgewiesen.
|
||||
func TestStore_RejectsSuperadminOutsideAllowedMatrix(t *testing.T) {
|
||||
store, userID, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
_, err := store.Assign(ctx, userID, RoleSuperadmin, "admin@acme.example")
|
||||
if !errors.Is(err, ErrRoleNotAssignableInTenantScope) {
|
||||
t.Fatalf("erwartet ErrRoleNotAssignableInTenantScope, habe %v", err)
|
||||
}
|
||||
|
||||
// Sicherstellen, dass der abgewiesene Versuch keine Zeile hinterlassen hat.
|
||||
if _, err := store.Get(ctx, userID); !errors.Is(err, ErrNotFound) {
|
||||
t.Fatalf("erwartet ErrNotFound nach abgewiesener zuweisung, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStore_RejectsUnknownRole(t *testing.T) {
|
||||
store, userID, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Assign(ctx, userID, Role("erfunden"), "admin@acme.example"); !errors.Is(err, ErrRoleNotAssignableInTenantScope) {
|
||||
t.Fatalf("erwartet ErrRoleNotAssignableInTenantScope fuer unbekannte rolle, habe %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3: Rollenzuweisung vollstaendig auditierbar.
|
||||
func TestStore_HistoryTracksWhoAndWhen(t *testing.T) {
|
||||
store, userID, cleanup := setupStoreTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
if _, err := store.Assign(ctx, userID, RoleUser, "system"); err != nil {
|
||||
t.Fatalf("assign 1: %v", err)
|
||||
}
|
||||
if _, err := store.Assign(ctx, userID, RoleTenantAdmin, "admin@acme.example"); err != nil {
|
||||
t.Fatalf("assign 2: %v", err)
|
||||
}
|
||||
|
||||
history, err := store.History(ctx, userID)
|
||||
if err != nil {
|
||||
t.Fatalf("history: %v", err)
|
||||
}
|
||||
if len(history) != 2 {
|
||||
t.Fatalf("erwartet 2 historieneintraege, habe %d", len(history))
|
||||
}
|
||||
if history[0].Role != RoleUser || history[0].GrantedBy != "system" {
|
||||
t.Fatalf("history[0] unerwartet: %+v", history[0])
|
||||
}
|
||||
if history[1].Role != RoleTenantAdmin || history[1].GrantedBy != "admin@acme.example" {
|
||||
t.Fatalf("history[1] unerwartet: %+v", history[1])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
package user
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// Handler stellt die CRUD-API fuer Benutzerkonten bereit (IAM-01-Auftrag).
|
||||
// Auth/Sessions (IAM-02) und Rollen (RBAC-01) sind ausdruecklich nicht Teil
|
||||
// dieser Kachel und daher hier noch nicht angebunden.
|
||||
type Handler struct {
|
||||
users *TenantUserStore
|
||||
superadmins *SuperadminStore
|
||||
}
|
||||
|
||||
func NewHandler(users *TenantUserStore, superadmins *SuperadminStore) *Handler {
|
||||
return &Handler{users: users, superadmins: superadmins}
|
||||
}
|
||||
|
||||
type createUserRequest struct {
|
||||
Email string `json:"email"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
|
||||
func (h *Handler) CreateUser(w http.ResponseWriter, r *http.Request) {
|
||||
var req createUserRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "ungueltige Anfrage", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
u, err := h.users.Create(r.Context(), req.Email, req.Name)
|
||||
writeUserResult(w, u, err)
|
||||
}
|
||||
|
||||
// CreateSuperadmin legt ein mandantenuebergreifendes Superadmin-Konto an —
|
||||
// bewusst ein eigener Endpunkt statt eines Tenant-Parameters mit Null-Wert.
|
||||
func (h *Handler) CreateSuperadmin(w http.ResponseWriter, r *http.Request) {
|
||||
var req createUserRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
|
||||
http.Error(w, "ungueltige Anfrage", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
u, err := h.superadmins.Create(r.Context(), req.Email, req.Name)
|
||||
writeUserResult(w, u, err)
|
||||
}
|
||||
|
||||
func writeUserResult(w http.ResponseWriter, u User, err error) {
|
||||
if err != nil {
|
||||
switch {
|
||||
case errors.Is(err, ErrInvalidEmail), errors.Is(err, ErrEmailTaken):
|
||||
http.Error(w, err.Error(), http.StatusBadRequest)
|
||||
case errors.Is(err, ErrNotFound):
|
||||
http.Error(w, err.Error(), http.StatusNotFound)
|
||||
default:
|
||||
http.Error(w, "benutzer konnte nicht verarbeitet werden", http.StatusInternalServerError)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
w.WriteHeader(http.StatusCreated)
|
||||
_ = json.NewEncoder(w).Encode(u)
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// setupTestDB legt eine frische, isolierte Testdatenbank an, wendet die
|
||||
// uebergebene Migration an und liefert einen verbundenen Pool. Wird ohne
|
||||
// TEST_ADMIN_DSN uebersprungen — siehe internal/tenant/provisioner_test.go
|
||||
// fuer dasselbe Muster.
|
||||
func setupTestDB(t *testing.T, dbName, schemaSQL string) *pgxpool.Pool {
|
||||
t.Helper()
|
||||
|
||||
adminDSN := os.Getenv("TEST_ADMIN_DSN")
|
||||
if adminDSN == "" {
|
||||
t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
adminPool, err := pgxpool.New(ctx, adminDSN)
|
||||
if err != nil {
|
||||
t.Fatalf("admin pool: %v", err)
|
||||
}
|
||||
|
||||
_, _ = adminPool.Exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, dbName))
|
||||
if _, err := adminPool.Exec(ctx, fmt.Sprintf(`CREATE DATABASE %q`, dbName)); err != nil {
|
||||
t.Fatalf("testdatenbank anlegen: %v", err)
|
||||
}
|
||||
|
||||
dsn := strings.Replace(adminDSN, "/postgres?", "/"+dbName+"?", 1)
|
||||
pool, err := pgxpool.New(ctx, dsn)
|
||||
if err != nil {
|
||||
t.Fatalf("connect testdatenbank: %v", err)
|
||||
}
|
||||
|
||||
if _, err := pool.Exec(ctx, schemaSQL); err != nil {
|
||||
t.Fatalf("schema anwenden: %v", err)
|
||||
}
|
||||
|
||||
t.Cleanup(func() {
|
||||
pool.Close()
|
||||
_, _ = adminPool.Exec(ctx, fmt.Sprintf(`DROP DATABASE IF EXISTS %q`, dbName))
|
||||
adminPool.Close()
|
||||
})
|
||||
|
||||
return pool
|
||||
}
|
||||
|
||||
const usersSchema = `
|
||||
CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
||||
CREATE TABLE users (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
email TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);`
|
||||
|
||||
const superadminsSchema = `
|
||||
CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
||||
CREATE TABLE superadmins (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
email TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);`
|
||||
|
||||
// Akzeptanzkriterium 1 + 3, Pruefung 1 (inkl. Negativfall doppelte E-Mail).
|
||||
func TestTenantUserStore_CRUD(t *testing.T) {
|
||||
pool := setupTestDB(t, "test_iam01_users", usersSchema)
|
||||
store := NewTenantUserStore(pool)
|
||||
ctx := context.Background()
|
||||
|
||||
created, err := store.Create(ctx, "alice@example.com", "Alice")
|
||||
if err != nil {
|
||||
t.Fatalf("create: %v", err)
|
||||
}
|
||||
if created.Status != StatusActive {
|
||||
t.Fatalf("erwartet status active, hat %q", created.Status)
|
||||
}
|
||||
|
||||
got, err := store.Get(ctx, created.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.Email != "alice@example.com" {
|
||||
t.Fatalf("get email = %q", got.Email)
|
||||
}
|
||||
|
||||
updated, err := store.Update(ctx, created.ID, "", "Alice A.")
|
||||
if err != nil {
|
||||
t.Fatalf("update: %v", err)
|
||||
}
|
||||
if updated.Name != "Alice A." || updated.Email != "alice@example.com" {
|
||||
t.Fatalf("update ergebnis unerwartet: %+v", updated)
|
||||
}
|
||||
|
||||
list, err := store.List(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("list: %v", err)
|
||||
}
|
||||
if len(list) != 1 {
|
||||
t.Fatalf("erwartet 1 benutzer, habe %d", len(list))
|
||||
}
|
||||
|
||||
deactivated, err := store.Deactivate(ctx, created.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("deactivate: %v", err)
|
||||
}
|
||||
if deactivated.Status != StatusInactive {
|
||||
t.Fatalf("erwartet status inactive, hat %q", deactivated.Status)
|
||||
}
|
||||
|
||||
// Negativfall: doppelte E-Mail-Adresse.
|
||||
if _, err := store.Create(ctx, "second@example.com", "Bob"); err != nil {
|
||||
t.Fatalf("create second: %v", err)
|
||||
}
|
||||
if _, err := store.Create(ctx, "second@example.com", "Bob Zwei"); !errors.Is(err, ErrEmailTaken) {
|
||||
t.Fatalf("erwartet ErrEmailTaken, habe %v", err)
|
||||
}
|
||||
|
||||
// Negativfall: fehlender Benutzer.
|
||||
if _, err := store.Get(ctx, created.ID+"-nicht-vorhanden"); err == nil {
|
||||
t.Fatalf("erwartet fehler bei unbekannter/ungueltiger id")
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Superadmin-Anlage ohne Tenant-Kontext.
|
||||
// SuperadminStore.Create hat keinen Tenant-Parameter — es gibt syntaktisch
|
||||
// keine Moeglichkeit, hier versehentlich einen Tenant-Sonderfall zu vergessen.
|
||||
func TestSuperadminStore_CreateWithoutTenantContext(t *testing.T) {
|
||||
pool := setupTestDB(t, "test_iam01_superadmins", superadminsSchema)
|
||||
store := NewSuperadminStore(pool)
|
||||
ctx := context.Background()
|
||||
|
||||
created, err := store.Create(ctx, "root@nexarch.internal", "Root")
|
||||
if err != nil {
|
||||
t.Fatalf("create superadmin: %v", err)
|
||||
}
|
||||
if created.Status != StatusActive {
|
||||
t.Fatalf("erwartet status active, hat %q", created.Status)
|
||||
}
|
||||
|
||||
got, err := store.Get(ctx, created.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("get: %v", err)
|
||||
}
|
||||
if got.Email != "root@nexarch.internal" {
|
||||
t.Fatalf("get email = %q", got.Email)
|
||||
}
|
||||
|
||||
if _, err := store.Create(ctx, "root@nexarch.internal", "Root Zwei"); !errors.Is(err, ErrEmailTaken) {
|
||||
t.Fatalf("erwartet ErrEmailTaken (globale eindeutigkeit), habe %v", err)
|
||||
}
|
||||
|
||||
deactivated, err := store.Deactivate(ctx, created.ID)
|
||||
if err != nil {
|
||||
t.Fatalf("deactivate: %v", err)
|
||||
}
|
||||
if deactivated.Status != StatusInactive {
|
||||
t.Fatalf("erwartet status inactive, hat %q", deactivated.Status)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// SuperadminStore verwaltet mandantenuebergreifende Superadmin-Konten in der
|
||||
// Control-Plane-Registry (siehe internal/tenant.Registry). Superadmin-ohne-
|
||||
// Tenant ist dadurch ein eigener Typ statt eines Sonderfalls von User/
|
||||
// TenantUserStore — es gibt keinen Tenant-Parameter, den man weglassen
|
||||
// koennte (IAM-01, "ohne Sonderbehandlung im Code").
|
||||
type SuperadminStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewSuperadminStore(pool *pgxpool.Pool) *SuperadminStore {
|
||||
return &SuperadminStore{pool: pool}
|
||||
}
|
||||
|
||||
func (s *SuperadminStore) Create(ctx context.Context, email, name string) (User, error) {
|
||||
if err := ValidateEmail(email); err != nil {
|
||||
return User{}, err
|
||||
}
|
||||
|
||||
var u User
|
||||
u.Email, u.Name, u.Status = email, name, StatusActive
|
||||
|
||||
row := s.pool.QueryRow(ctx, `
|
||||
INSERT INTO superadmins (email, name, status)
|
||||
VALUES ($1, $2, $3)
|
||||
RETURNING id, created_at, updated_at
|
||||
`, u.Email, u.Name, u.Status)
|
||||
|
||||
if err := row.Scan(&u.ID, &u.CreatedAt, &u.UpdatedAt); err != nil {
|
||||
return User{}, mapWriteErr(err)
|
||||
}
|
||||
return u, nil
|
||||
}
|
||||
|
||||
func (s *SuperadminStore) Get(ctx context.Context, id string) (User, error) {
|
||||
return scanUser(s.pool.QueryRow(ctx, `
|
||||
SELECT id, email, name, status, created_at, updated_at
|
||||
FROM superadmins WHERE id = $1
|
||||
`, id))
|
||||
}
|
||||
|
||||
func (s *SuperadminStore) List(ctx context.Context) ([]User, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, email, name, status, created_at, updated_at
|
||||
FROM superadmins ORDER BY created_at
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("superadmins auflisten: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []User
|
||||
for rows.Next() {
|
||||
var u User
|
||||
if err := rows.Scan(&u.ID, &u.Email, &u.Name, &u.Status, &u.CreatedAt, &u.UpdatedAt); err != nil {
|
||||
return nil, fmt.Errorf("superadmin lesen: %w", err)
|
||||
}
|
||||
out = append(out, u)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *SuperadminStore) Deactivate(ctx context.Context, id string) (User, error) {
|
||||
return scanUser(s.pool.QueryRow(ctx, `
|
||||
UPDATE superadmins SET status = $2, updated_at = now()
|
||||
WHERE id = $1
|
||||
RETURNING id, email, name, status, created_at, updated_at
|
||||
`, id, StatusInactive))
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
package user
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// TenantUserStore verwaltet Benutzer innerhalb GENAU EINER Tenant-Datenbank.
|
||||
// Welcher Mandant gemeint ist, ergibt sich ausschliesslich aus dem
|
||||
// uebergebenen Pool — es gibt keine tenant_id-Spalte (siehe migrations/tenant/0001_users.up.sql).
|
||||
type TenantUserStore struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewTenantUserStore(pool *pgxpool.Pool) *TenantUserStore {
|
||||
return &TenantUserStore{pool: pool}
|
||||
}
|
||||
|
||||
func (s *TenantUserStore) Create(ctx context.Context, email, name string) (User, error) {
|
||||
if err := ValidateEmail(email); err != nil {
|
||||
return User{}, err
|
||||
}
|
||||
|
||||
var u User
|
||||
u.Email, u.Name, u.Status = email, name, StatusActive
|
||||
|
||||
row := s.pool.QueryRow(ctx, `
|
||||
INSERT INTO users (email, name, status)
|
||||
VALUES ($1, $2, $3)
|
||||
RETURNING id, created_at, updated_at
|
||||
`, u.Email, u.Name, u.Status)
|
||||
|
||||
if err := row.Scan(&u.ID, &u.CreatedAt, &u.UpdatedAt); err != nil {
|
||||
return User{}, mapWriteErr(err)
|
||||
}
|
||||
return u, nil
|
||||
}
|
||||
|
||||
func (s *TenantUserStore) Get(ctx context.Context, id string) (User, error) {
|
||||
return scanUser(s.pool.QueryRow(ctx, `
|
||||
SELECT id, email, name, status, created_at, updated_at
|
||||
FROM users WHERE id = $1
|
||||
`, id))
|
||||
}
|
||||
|
||||
func (s *TenantUserStore) List(ctx context.Context) ([]User, error) {
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id, email, name, status, created_at, updated_at
|
||||
FROM users ORDER BY created_at
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("benutzer auflisten: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var out []User
|
||||
for rows.Next() {
|
||||
var u User
|
||||
if err := rows.Scan(&u.ID, &u.Email, &u.Name, &u.Status, &u.CreatedAt, &u.UpdatedAt); err != nil {
|
||||
return nil, fmt.Errorf("benutzer lesen: %w", err)
|
||||
}
|
||||
out = append(out, u)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// Update aendert Name und E-Mail. Eine leere email/name laesst das jeweilige
|
||||
// Feld unveraendert.
|
||||
func (s *TenantUserStore) Update(ctx context.Context, id, email, name string) (User, error) {
|
||||
if email != "" {
|
||||
if err := ValidateEmail(email); err != nil {
|
||||
return User{}, err
|
||||
}
|
||||
}
|
||||
|
||||
row := s.pool.QueryRow(ctx, `
|
||||
UPDATE users
|
||||
SET email = COALESCE(NULLIF($2, ''), email),
|
||||
name = COALESCE(NULLIF($3, ''), name),
|
||||
updated_at = now()
|
||||
WHERE id = $1
|
||||
RETURNING id, email, name, status, created_at, updated_at
|
||||
`, id, email, name)
|
||||
|
||||
u, err := scanUser(row)
|
||||
if err != nil {
|
||||
return User{}, mapWriteErr(err)
|
||||
}
|
||||
return u, nil
|
||||
}
|
||||
|
||||
// Deactivate setzt den Benutzer auf inaktiv statt ihn zu loeschen.
|
||||
func (s *TenantUserStore) Deactivate(ctx context.Context, id string) (User, error) {
|
||||
return scanUser(s.pool.QueryRow(ctx, `
|
||||
UPDATE users SET status = $2, updated_at = now()
|
||||
WHERE id = $1
|
||||
RETURNING id, email, name, status, created_at, updated_at
|
||||
`, id, StatusInactive))
|
||||
}
|
||||
|
||||
func scanUser(row pgx.Row) (User, error) {
|
||||
var u User
|
||||
if err := row.Scan(&u.ID, &u.Email, &u.Name, &u.Status, &u.CreatedAt, &u.UpdatedAt); err != nil {
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return User{}, ErrNotFound
|
||||
}
|
||||
return User{}, fmt.Errorf("benutzer lesen: %w", err)
|
||||
}
|
||||
return u, nil
|
||||
}
|
||||
|
||||
// mapWriteErr uebersetzt den Unique-Constraint-Verstoss der E-Mail-Spalte in
|
||||
// einen sprechenden Fehler statt der rohen Postgres-Fehlermeldung.
|
||||
func mapWriteErr(err error) error {
|
||||
var pgErr *pgconn.PgError
|
||||
if errors.As(err, &pgErr) && pgErr.Code == "23505" {
|
||||
return ErrEmailTaken
|
||||
}
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return ErrNotFound
|
||||
}
|
||||
return fmt.Errorf("benutzer schreiben: %w", err)
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
// Package user implementiert Core IAM-01: das Benutzer-Datenmodell und die
|
||||
// CRUD-Operationen. Tenant-Zugehoerigkeit ist ueber die Zieldatenbank
|
||||
// gegeben (Modell C, siehe internal/tenant) — Superadmin-Konten leben
|
||||
// dagegen mandantenuebergreifend in der Registry und sind ueber
|
||||
// SuperadminStore als eigener, First-Class-Typ modelliert, nicht als
|
||||
// tenant_id-NULL-Sonderfall in User.
|
||||
package user
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"regexp"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Status string
|
||||
|
||||
const (
|
||||
StatusActive Status = "active"
|
||||
StatusInactive Status = "inactive"
|
||||
)
|
||||
|
||||
type User struct {
|
||||
ID string
|
||||
Email string
|
||||
Name string
|
||||
Status Status
|
||||
CreatedAt time.Time
|
||||
UpdatedAt time.Time
|
||||
}
|
||||
|
||||
var emailPattern = regexp.MustCompile(`^[^\s@]+@[^\s@]+\.[^\s@]+$`)
|
||||
|
||||
var ErrInvalidEmail = errors.New("user: ungueltige E-Mail-Adresse")
|
||||
var ErrEmailTaken = errors.New("user: E-Mail-Adresse bereits vergeben")
|
||||
var ErrNotFound = errors.New("user: nicht gefunden")
|
||||
|
||||
func ValidateEmail(email string) error {
|
||||
if !emailPattern.MatchString(email) {
|
||||
return ErrInvalidEmail
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package user
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestValidateEmail(t *testing.T) {
|
||||
cases := []struct {
|
||||
email string
|
||||
wantErr bool
|
||||
}{
|
||||
{"a@b.de", false},
|
||||
{"a.b+c@sub.example.com", false},
|
||||
{"", true},
|
||||
{"keine-email", true},
|
||||
{"a@b", true},
|
||||
{"@b.de", true},
|
||||
}
|
||||
|
||||
for _, c := range cases {
|
||||
err := ValidateEmail(c.email)
|
||||
if (err != nil) != c.wantErr {
|
||||
t.Errorf("ValidateEmail(%q) error = %v, wantErr %v", c.email, err, c.wantErr)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS superadmins;
|
||||
@@ -0,0 +1,14 @@
|
||||
-- Superadmin-Konten arbeiten mandantenuebergreifend und leben deshalb in der
|
||||
-- Control-Plane-Registry (siehe TEN-01), nicht in einer Tenant-Datenbank.
|
||||
-- Das bildet "Superadmin ohne Tenant" strukturell als First-Class-Zustand ab,
|
||||
-- statt ihn als Sonderfall in der Tenant-users-Tabelle zu behandeln
|
||||
-- (IAM-01, siehe core-kanban/tickets/IAM-01.md — bekannte Fehler vermeiden).
|
||||
-- E-Mail-Eindeutigkeit ist hier global, da die Registry-DB einmalig existiert.
|
||||
CREATE TABLE superadmins (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
email TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE IF EXISTS policy_rule_changes;
|
||||
DROP TABLE IF EXISTS policy_rules;
|
||||
@@ -0,0 +1,24 @@
|
||||
-- Zentrale, deklarative Policy-Regeln (RBAC-02, siehe core-kanban/tickets/RBAC-02.md).
|
||||
-- policy_rules haelt den AKTUELLEN, gewaehrten Regelsatz (Existenz = erlaubt,
|
||||
-- Default-Deny fuer alles ohne Zeile). policy_rule_changes ist die
|
||||
-- versionierte Aenderungshistorie (Akzeptanzkriterium 3: Regelwechsel ohne
|
||||
-- Codeaenderung nachvollziehbar).
|
||||
CREATE TABLE policy_rules (
|
||||
role TEXT NOT NULL,
|
||||
permission TEXT NOT NULL,
|
||||
granted_by TEXT NOT NULL,
|
||||
granted_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (role, permission)
|
||||
);
|
||||
|
||||
CREATE TABLE policy_rule_changes (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
role TEXT NOT NULL,
|
||||
permission TEXT NOT NULL,
|
||||
action TEXT NOT NULL CHECK (action IN ('grant', 'revoke')),
|
||||
actor TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX policy_rule_changes_idx ON policy_rule_changes (role, permission, version);
|
||||
@@ -1,2 +0,0 @@
|
||||
DROP TABLE IF EXISTS config_value_history;
|
||||
DROP TABLE IF EXISTS config_values;
|
||||
@@ -1,23 +0,0 @@
|
||||
-- Zentraler Konfigurationsdienst (CFG-01, siehe core-kanban/tickets/CFG-01.md).
|
||||
-- scope = 'global' fuer globale Defaults, sonst der Tenant-Slug. config_values
|
||||
-- haelt den AKTUELLEN Stand je (key, scope); config_value_history haelt JEDE
|
||||
-- Aenderung fest (Akzeptanzkriterium 2: versioniert nachvollziehbar).
|
||||
CREATE TABLE config_values (
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL CHECK (scope <> ''),
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
PRIMARY KEY (key, scope)
|
||||
);
|
||||
|
||||
CREATE TABLE config_value_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
key TEXT NOT NULL,
|
||||
scope TEXT NOT NULL,
|
||||
value TEXT NOT NULL,
|
||||
version INT NOT NULL,
|
||||
changed_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX config_value_history_key_scope_idx ON config_value_history (key, scope, version);
|
||||
@@ -1 +0,0 @@
|
||||
DROP TABLE IF EXISTS feature_flags;
|
||||
@@ -1,10 +0,0 @@
|
||||
-- Feature-Flags zentral je Mandant/Zielgruppe (LIC-02, siehe core-kanban/tickets/LIC-02.md).
|
||||
-- Lebt in der Registry-DB, nicht pro Tenant-Datenbank — Flags sind eine
|
||||
-- Core-weite Konfiguration, keine Mandanten-Geschaeftsdaten.
|
||||
CREATE TABLE feature_flags (
|
||||
key TEXT PRIMARY KEY,
|
||||
enabled BOOLEAN NOT NULL DEFAULT false,
|
||||
rollout_percentage INT NOT NULL DEFAULT 0 CHECK (rollout_percentage BETWEEN 0 AND 100),
|
||||
target_tenant_slugs TEXT[] NOT NULL DEFAULT '{}',
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
@@ -1,2 +0,0 @@
|
||||
DROP TABLE IF EXISTS module_credentials;
|
||||
DROP TABLE IF EXISTS modules;
|
||||
@@ -1,18 +0,0 @@
|
||||
-- Modul-Registry & Aktivierungspruefung (API-02, siehe core-kanban/tickets/API-02.md).
|
||||
CREATE TABLE modules (
|
||||
name TEXT PRIMARY KEY,
|
||||
version TEXT NOT NULL CHECK (version <> ''),
|
||||
required_flags TEXT[] NOT NULL DEFAULT '{}',
|
||||
registered_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- Service-Credential je Modul-Instanz, bei Provisionierung ausgestellt
|
||||
-- (Akzeptanzkriterium 4). secret_hash enthaelt NIEMALS das Secret im
|
||||
-- Klartext, nur dessen SHA-256-Hash (Timing-safe-Vergleich beim Login,
|
||||
-- Referenzmuster siehe AUD-02).
|
||||
CREATE TABLE module_credentials (
|
||||
module_name TEXT PRIMARY KEY REFERENCES modules(name),
|
||||
client_id TEXT NOT NULL UNIQUE,
|
||||
secret_hash BYTEA NOT NULL,
|
||||
issued_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
@@ -1 +0,0 @@
|
||||
DROP TABLE IF EXISTS notification_jobs;
|
||||
@@ -1,20 +0,0 @@
|
||||
-- Benachrichtigungs-Dispatcher-Warteschlange (CFG-02, siehe
|
||||
-- core-kanban/tickets/CFG-02.md). Postgres-basiert statt Redis/AMQP
|
||||
-- (Projekt-Konvention, siehe nexarch-state.json techstack.job_queue) —
|
||||
-- Zeilen ueberleben einen Neustart des Dispatcher-Prozesses unveraendert
|
||||
-- (Akzeptanzkriterium 3).
|
||||
CREATE TABLE notification_jobs (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
channel TEXT NOT NULL,
|
||||
recipient TEXT NOT NULL,
|
||||
payload JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')),
|
||||
attempts INT NOT NULL DEFAULT 0,
|
||||
max_attempts INT NOT NULL DEFAULT 5,
|
||||
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX notification_jobs_due_idx ON notification_jobs (status, next_attempt_at);
|
||||
@@ -1,2 +0,0 @@
|
||||
DROP TABLE alert_debounce_state;
|
||||
DROP TABLE alert_rules;
|
||||
@@ -1,24 +0,0 @@
|
||||
-- OPS-05: Schwellwert-Regeln fuer Alerting auf den aus OPS-03 aggregierten
|
||||
-- Metriken. Lebt wie config_values/notification_jobs (CFG-01/02) in der
|
||||
-- zentralen Registry-DB — modulübergreifende Betriebskonfiguration, keine
|
||||
-- Mandanten-Geschaeftsdaten.
|
||||
CREATE TABLE alert_rules (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
metric_name TEXT NOT NULL,
|
||||
comparison TEXT NOT NULL CHECK (comparison IN ('gt', 'lt')),
|
||||
threshold DOUBLE PRECISION NOT NULL,
|
||||
label_filters JSONB NOT NULL DEFAULT '{}'::jsonb,
|
||||
recipient TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- Haelt fest, wann eine Regel zuletzt tatsaechlich einen Alarm ausgeloest
|
||||
-- hat (Akzeptanzkriterium 3: Drosselung wiederholter Alarmierung fuer
|
||||
-- denselben anhaltenden Zustand). rule_key kombiniert Regel-ID mit den
|
||||
-- tatsaechlichen Label-Werten der ausloesenden Zeitreihe, damit dieselbe
|
||||
-- Regel fuer unterschiedliche Tenants/Module unabhaengig gedrosselt wird.
|
||||
CREATE TABLE alert_debounce_state (
|
||||
rule_key TEXT PRIMARY KEY,
|
||||
last_fired_at TIMESTAMPTZ NOT NULL
|
||||
);
|
||||
@@ -1 +0,0 @@
|
||||
DROP TABLE metrics_sources;
|
||||
@@ -1,7 +0,0 @@
|
||||
-- Metrics-Aggregation ueber Module hinweg (OPS-03, siehe
|
||||
-- core-kanban/tickets/OPS-03.md) — welches Modul liefert seine Kennzahlen
|
||||
-- unter welcher /metrics-URL.
|
||||
CREATE TABLE metrics_sources (
|
||||
module_name TEXT PRIMARY KEY,
|
||||
metrics_url TEXT NOT NULL
|
||||
);
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS users;
|
||||
@@ -0,0 +1,16 @@
|
||||
-- Benutzer-Datenmodell (IAM-01, siehe core-kanban/tickets/IAM-01.md).
|
||||
-- Diese Migration laeuft in der DB EINES Mandanten (Modell C, siehe TEN-01) —
|
||||
-- die Tenant-Zugehoerigkeit ist implizit durch die Datenbankverbindung
|
||||
-- gegeben, es gibt daher bewusst KEINE tenant_id-Spalte.
|
||||
-- E-Mail-Eindeutigkeit ist hier tenant-scoped: der UNIQUE-Constraint gilt
|
||||
-- nur innerhalb dieser einen Tenant-Datenbank.
|
||||
CREATE EXTENSION IF NOT EXISTS pgcrypto;
|
||||
|
||||
CREATE TABLE users (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
email TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE IF EXISTS role_assignment_history;
|
||||
DROP TABLE IF EXISTS role_assignments;
|
||||
@@ -0,0 +1,23 @@
|
||||
-- Rollenzuweisung pro Benutzer (RBAC-01, siehe core-kanban/tickets/RBAC-01.md).
|
||||
-- Nur 'user' und 'tenant_admin' sind hier zuweisbar — 'superadmin' lebt
|
||||
-- mandantenuebergreifend in der Registry (IAM-01 SuperadminStore) und hat
|
||||
-- daher bewusst KEINE Zeile in dieser tenant-lokalen Tabelle (Akzeptanz-
|
||||
-- kriterium 1: nur Zuweisungen innerhalb der erlaubten Matrix).
|
||||
CREATE TABLE role_assignments (
|
||||
user_id UUID PRIMARY KEY REFERENCES users(id),
|
||||
role TEXT NOT NULL CHECK (role IN ('user', 'tenant_admin')),
|
||||
granted_by TEXT NOT NULL,
|
||||
granted_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- Vollstaendige Historie jeder Rollenaenderung (Akzeptanzkriterium 3: wer
|
||||
-- hat wann welche Rolle vergeben).
|
||||
CREATE TABLE role_assignment_history (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
user_id UUID NOT NULL,
|
||||
role TEXT NOT NULL,
|
||||
granted_by TEXT NOT NULL,
|
||||
granted_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX role_assignment_history_user_idx ON role_assignment_history (user_id, granted_at);
|
||||
@@ -14,9 +14,6 @@ ROLE="nexarch_test"
|
||||
export PGPASSWORD="$PASS"
|
||||
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS tenants CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS audit_events CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS config_value_history CASCADE;"
|
||||
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP TABLE IF EXISTS config_values CASCADE;"
|
||||
|
||||
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
|
||||
for db in $dbs; do
|
||||
|
||||
Reference in New Issue
Block a user