Compare commits

..
Author SHA1 Message Date
sysops d6e9412043 API-11: wiederanlauf-nachsynchronisierungs-endpunkt-starten
- Merge API-02 (bereits Vorfahre), AUD-01, LIC-03 in API-06-Branch, um
  echte Produktionsimplementierungen fuer resync.Handler zu erhalten
  (moduleregistry.Registry, audit.Log, usage.Store) - go.mod/go.sum/
  scripts/reset-test-env.sh additiv zusammengefuehrt, keine Logikkonflikte
- cmd/resync-api: startet internal/resync.Handler (API-06) als
  eigenstaendigen HTTP-Dienst, kleiner auditAdapter fuer Signatur-
  Anpassung (kein neuer Fachcode)
- reines Wiring, kein Diff an internal/resync|audit|usage|moduleregistry
  (verifiziert)
- real deployed auf 131, end-zu-ende per curl mit echtem, ueber
  moduleregistry.Registry.Provision ausgestelltem Service-Credential:
  angewendetes usage-delta real in usage_counters bestaetigt, falsches
  Credential -> 401
- reale Grant-Luecke gefunden und behoben (nexarch_core auf modules/
  module_credentials/audit_events/feature_flags/usage_counters/
  resync_*_buffer), ueber information_schema verifiziert

Pruefungen siehe docs/API-11-PRUEFPROTOKOLL.md
2026-08-30 20:14:07 +02:00
sysops e6892535ed Merge branch 'feature/lic-03-nutzungszaehler-quotas' into feature/api-11-wiederanlauf-nachsynchronisierungs-endpunkt-starten
# Conflicts:
#	go.mod
#	scripts/reset-test-env.sh
2026-08-30 20:09:41 +02:00
sysops 1c41b65a47 Merge branch 'feature/aud-01-zentrales-audit-log-modell' into feature/api-11-wiederanlauf-nachsynchronisierungs-endpunkt-starten
# Conflicts:
#	go.mod
#	go.sum
#	scripts/reset-test-env.sh
2026-08-30 20:09:26 +02:00
sysops ec9bb27bb3 API-06: go.sum/go.mod aktualisieren (golang-jwt/jwt/v5 fuer moduletrust) 2026-08-28 09:22:56 +02:00
sysops 0fd9856b10 API-06: wiederanlauf-nachsynchronisierung-nach-core-ausfall (postgres-puffer, service-credential, sofort-invalidate) 2026-08-28 09:22:08 +02:00
sysopsandClaude Sonnet 5 b23cd1961f API-02: modul-registry-aktivierungspruefung
internal/moduleregistry: Registry.Register traegt Fachmodule mit Name,
Version und benoetigten Feature-Flags ein (Akzeptanzkriterium 1), fehlende
Pflichtangaben werden abgewiesen. IsActive kombiniert Registrierung + LIC-02
Feature-Flag-Auswertung (ALLE benoetigten Flags muessen fuer den Tenant
aktiv sein) — ein nicht registriertes Modul ist nie aktiv. List liefert alle
Module fuer Statusseite/Lizenzoberflaeche (Akzeptanzkriterium 3).

RequireActiveModule ist die zentrale Durchsetzungs-Middleware (Casbin-
Prinzip): weist Anfragen an ein deaktiviertes Modul ab, BEVOR der
Modul-Handler ueberhaupt aufgerufen wird (Akzeptanzkriterium 2) —
Pruefung per Test belegt, dass der Handler bei Deaktivierung nachweislich
nicht erreicht wird.

Service-Credentials (Akzeptanzkriterium 4): Registry.Provision stellt pro
Modul-Instanz Client-ID + Secret aus, gespeichert wird nur der SHA-256-Hash
des Secrets. Registry.Authenticate vergleicht timing-safe (dieselbe
subtle.ConstantTimeCompare-Referenzimplementierung wie AUD-02).
RequireServiceCredential-Middleware liest X-Client-Id/X-Client-Secret und
weist Aufrufe ohne gueltiges Credential mit 401 ab, bevor der Core-seitige
Endpunkt (z.B. Audit-Nachlieferung, Nutzungsmeldung) erreicht wird.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Anfrage an deaktiviertes Modul nachweislich vor Modul-Logik abgewiesen —
   TestRequireActiveModule_BlocksBeforeHandler: handlerReached bleibt false
   bei 403, wird true erst nach Aktivierung bei 200. PASS.
2. Registrierung mit fehlenden Pflichtangaben abgewiesen —
   TestRegister_RejectsMissingFields (leerer Name, leere Version). PASS.
3. Registry-Abfrage liefert konsistente Daten nach Aktivierung/Deaktivierung —
   TestIsActive_ReflectsFlagStateConsistently: aus/an/aus-Zyklus, IsActive
   folgt dem Flag-Zustand korrekt. PASS.
4. Aufruf mit ungueltigem/fehlendem Service-Credential abgewiesen, mit
   gueltigem angenommen — TestRequireServiceCredential_RejectsInvalidAcceptsValid
   und TestProvisionAndAuthenticate (falsches Secret, unbekannte Client-ID,
   korrektes Credential). PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 21:17:50 +02:00
sysopsandClaude Sonnet 5 d447869246 LIC-03: nutzungszaehler-quotas
internal/usage: Store.Increment aktualisiert Zaehlerstaende ueber ein
einziges atomares SQL-UPSERT (value = value + delta) statt Read-Modify-Write
in Go — haelt Zaehlerstaende bei parallelen Schreibzugriffen konsistent
(Akzeptanzkriterium 1), ganz ohne Anwendungs-Lock. Quotas sind Konfiguration
(usage_quotas-Tabelle je Tenant+Metrik), kein Hardcode.

Check/Enforce leiten aus Zaehlerstand + Quota eine definierte Reaktion ab
(StatusOK/Warning bei 80%/Exceeded, Akzeptanzkriterium 2) — Enforce ruft eine
uebergebene Reaction-Funktion auf, wenn der Status nicht OK ist; die
konkrete Sperr-/Benachrichtigungslogik bleibt beim Aufrufer (z.B. TEN-02 vor
Benutzeranlage), Enforce garantiert nur zuverlaessiges Ausloesen. Fehlende
Quota-Konfiguration bedeutet unbegrenzt (StatusOK), kein Fehler.

RunPeriodicAggregation ist das Aggregations-Grundgerüst (Akzeptanzkriterium 1:
"periodisch aggregiert") — dieselbe In-Prozess-Worker-Goroutine-Konvention
wie internal/tenant.Lifecycle.RunSweeper. Die konkrete Aggregationsquelle
(Zeilen zaehlen in Modul-Tabellen) haengt vom jeweiligen Modul ab und ist
nicht Teil dieser Kachel.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Quota-Ueberschreitung automatisiert erkannt, definierte Reaktion
   ausgeloest — TestEnforce_TriggersReactionOnExceeded: Reaction-Callback
   wird mit StatusExceeded aufgerufen. PASS.
2. Aggregationsjob liefert bei parallelen Schreibzugriffen konsistente
   Zaehlerstaende — TestIncrement_ConsistentUnderConcurrentWrites: 50
   nebenlaeufige Increments, Endstand exakt 50 (kein Lost Update). PASS.
3. Zaehlerstand eines Tenants beeinflusst nicht den eines anderen —
   TestIncrement_IsolatedBetweenTenants. PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 21:10:11 +02:00
sysopsandClaude Sonnet 5 b12d53f469 AUD-01: zentrales-audit-log-modell
internal/audit: eigenes, strukturiertes Audit-Datenmodell (Akteur, Aktion,
Zielobjekt, Zeitpunkt, Tenant) in der Registry-DB, getrennt von jedem
allgemeinen Anwendungs-Log (eigenes Paket, eigene Tabelle audit_events,
kein Logging-Framework). Log.Record ist der EINE zentrale Schreibpfad —
es gibt keine zweite Schreibmoeglichkeit, ueber die ein Handler die
Validierung umgehen koennte.

Fehlender Tenant-Bezug wird zweifach verhindert (Akzeptanzkriterium 2):
Log.Record weist leeren TenantSlug direkt ab (ErrMissingTenant), zusaetzlich
erzwingt eine CHECK-Constraint in der Migration dasselbe auf Datenbankebene,
selbst wenn Log.Record umgangen wuerde. Mandantenuebergreifende Ereignisse
(z.B. Superadmin-Aktionen) nutzen den reservierten Wert audit.SystemTenant
statt NULL oder leerem String — es gibt keinen Weg, ganz ohne Tenant-Bezug
zu schreiben.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Automatisierter Test belegt genau einen Audit-Eintrag pro
   sicherheitsrelevantem Vorgang — TestRecord_PersistsExactlyOneEventPerSecurityIncident
   (simulierter fehlgeschlagener Login), Feldinhalte verifiziert. PASS.
2. Fehlender Tenant-Bezug durch Constraint/Test verhindert —
   TestRecord_RejectsMissingTenant (App-Ebene) UND
   TestConstraint_RejectsMissingTenantAtDatabaseLevel (direkter INSERT unter
   Umgehung von Log.Record, durch CHECK-Constraint abgewiesen). PASS.
3. Datenmodell von zweiter Person gegen Dokumentation geprueft — NICHT
   durchgefuehrt (keine zweite Person in dieser Session verfuegbar). Offen.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 19:22:50 +02:00
sysopsandClaude Sonnet 5 4a30345e07 LIC-02: feature-flag-service-je-tenant
internal/flag: Store (Verwaltung) + Service (Auswertung mit TTL-Cache,
Default 5s) — Unleash-Prinzip Flag-Verwaltung vs. Flag-Auswertung getrennt,
als Kernfunktion des Core-Dienstes selbst statt separater Infrastruktur.

evaluate() wendet drei Strategien in fester Reihenfolge an: global an/aus,
Tenant-Zielgruppe, deterministischer Prozentsatz-Rollout (FNV-Hash aus
Tenant+Key, stabil pro Tenant). IsEnabled liefert IMMER nur bool (kein
Fehlerwert) — ein nicht erreichbarer Flag-Dienst kann damit keinen
Aufrufer zum Absturz bringen: bei DB-Fehler wird der zuletzt bekannte
Cache-Stand verwendet, ohne jeglichen Stand faellt der Dienst sicher auf
false zurueck. Service.Invalidate erzwingt sofortiges Neuladen fuer den
Schreiber selbst, andere Instanzen sehen Aenderungen spaetestens nach der
TTL (Akzeptanzkriterium 3, kein Neustart noetig).

Bugfix waehrend Tests: Store.Set uebergab ein nil-TargetTenantSlugs-Slice
als SQL NULL statt leerem Array (NOT-NULL-Verletzung) — auf leeres Slice
normalisiert.

Akzeptanzkriterium 4 (Deaktivierung loescht keine Daten): dieses Paket
besitzt ausschliesslich die eigene feature_flags-Zeile, hat keinerlei
Code-Pfad, der Modul-Geschaeftsdaten anfassen koennte — Loeschung bleibt
strukturell der Archive-Retention-Engine vorbehalten.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Cache-Invalidierungszeit automatisiert gemessen —
   TestService_CacheInvalidationTiming: Aenderung wirksam nach 153ms bei
   TTL=150ms (innerhalb Ziel+Toleranz), vorher nachweislich noch alter Stand. PASS.
2. Zielgruppen-Strategie liefert erwartete Auswertung —
   TestService_TargetTenantStrategy / TestEvaluate_TargetTenantStrategy. PASS.
3. Ausfall des Flag-Dienstes fuehrt zu dokumentiertem Fallback, kein Absturz —
   TestService_FallsBackOnStoreFailure (mit recover()-Absicherung): Fallback
   auf Cache-Stand bzw. sicheres false bei komplett unerreichbarer DB, geloggt. PASS.
4. Modul-Deaktivierung/Reaktivierung ohne Datenverlust — architektonisch durch
   fehlenden Code-Pfad sichergestellt (siehe oben), zusaetzlich durch
   TestService_InvalidateForcesImmediateRefresh (Toggle aus/an bleibt
   konsistent nachvollziehbar) mitabgedeckt. PASS.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 19:19:59 +02:00
sysopsandClaude Sonnet 5 d63db93a4c LIC-01: lizenzmodell-lizenzschluessel-pruefung
internal/license: Ed25519-signierte Lizenzschluessel (stdlib crypto/ed25519,
keine neue Abhaengigkeit). Issuer haelt den privaten Schluessel (lebt beim
Lizenzgeber), Validator nur den oeffentlichen (lebt im Core-Prozess) — klare
Trennung Ausstellung/Pruefung nach Unleash-Vorbild (Flag-Verwaltung vs.
Flag-Auswertung).

Store.Install prueft NUR die Signatur und persistiert den Lizenzumfang
(Plan, Modul-Liste, Laufzeit) in tenant_licenses (Registry-DB, 1:1 zu
tenants). Eine bereits abgelaufene, aber korrekt signierte Lizenz laesst
sich trotzdem einspielen — der Ablauf wird erst bei Store.RequireActive
bewertet (liefert ErrLicenseExpired statt Panic/Absturz), waehrend
Store.Status den Umfang unabhaengig vom Ablauf weiterhin liefert.

Neu: scripts/run-checks.sh buendelt reset-test-env.sh + go build/vet/test
(-p 1) zu einem Ein-Kommando-Check fuer den Testhost.

Pruefungen (ausgefuehrt auf root@192.168.1.131, go build/vet/test PASS):
1. Manipulierter Lizenzschluessel zuverlaessig erkannt —
   TestParse_RejectsTamperedKey, TestParse_RejectsWrongKeyPair,
   TestStore_InstallRejectsInvalidSignature. PASS.
2. Ablauf loest definierten eingeschraenkten Zustand aus, kein harter
   Systemausfall — TestStore_RequireActive_DetectsExpiry (inkl. recover()-
   Absicherung im Test, dass kein Panic auftritt), ErrLicenseExpired statt
   Absturz; Status bleibt trotzdem abfragbar. PASS.
3. Signaturpruefung von zweiter Person gegen Dokumentation nachvollzogen —
   NICHT durchgefuehrt (keine zweite Person in dieser Session verfuegbar).
   Offen.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-27 19:12:13 +02:00
120 changed files with 3709 additions and 4551 deletions
-2
View File
@@ -1,4 +1,2 @@
*.log
.env
web/*/node_modules/
web/*/.next/
-96
View File
@@ -64,9 +64,6 @@ Keine Änderungen ermittelbar.
## 2026-08-27 17:28 17:29 (1m)
**Beschreibung:** Claude Code Session
**Projekt:** code
## 2026-08-28 21:44 21:44 (0m)
**Beschreibung:** Claude Code Session
**Projekt:** nexarch
### Commits
Keine Commits in dieser Session.
@@ -130,98 +127,5 @@ Keine Commits in dieser Session.
- internal/config/config.go | 29 +++++++++++++++++++++++++++++
- internal/db/db.go | 11 +++++++++++
- migrations/0001_tenant_registry.sql | 10 ++++++++++
- web/shl/README.md | 45 +++++++++++++++++++++++++++++++++++++++++++++
- web/shl/__tests__/Dialog.test.tsx | 38 ++++++++++++++++++++++++++++++++++++++
- web/shl/__tests__/tokens.test.ts | 39 +++++++++++++++++++++++++++++++++++++++
- web/shl/components/Dialog.tsx | 88 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/components/FormElements.tsx | 98 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/components/Shell.tsx | 62 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/components/Table.tsx | 83 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/components/Toast.tsx | 67 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/i18n/i18n.tsx | 75 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/index.ts | 25 +++++++++++++++++++++++++
- web/shl/package.json | 23 +++++++++++++++++++++++
- web/shl/theme/ThemeProvider.tsx | 81 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/tokens/tokens.ts | 96 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/shl/tsconfig.json | 18 ++++++++++++++++++
---
## 2026-08-28 21:51 21:57 (5m)
**Beschreibung:** Claude Code Session
**Projekt:** nexarch
### Commits
- 3c226da SHL-01: fix — vitest jsdom-environment + jest-dom-Setup (3 Dialog-Tests schlugen ohne DOM fehl)
### Geänderte Dateien
- web/shl/package.json | 2 ++
- web/shl/vitest.config.ts | 8 ++++++++
- web/shl/vitest.setup.ts | 1 +
---
## 2026-08-28 23:45 23:51 (5m)
**Beschreibung:** Claude Code Session
**Projekt:** code
### Commits
- b27a640 CFG-04: benachrichtigungs-einstellungen-oberflaeche (handler+tests fuer notifyprefs, web/notifications next.js-frontend auf shl-01)
- 81ff8c1 Merge branch 'feature/shl-01-ui-shell-design-system-zentral' into feature/cfg-04-benachrichtigungs-einstellungen-oberflaeche
- 22b3349 Merge branch 'feature/iam-02-login-session-jwt-grundgeruest' into feature/cfg-04-benachrichtigungs-einstellungen-oberflaeche
- 2df3f93 CFG-04: backend teil 1 — internal/notifyprefs (praeferenz-store + enqueueifallowed-filter vor dispatcher)
### Geänderte Dateien
- internal/notifyprefs/handler.go | 83 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- internal/notifyprefs/handler_test.go | 122 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- internal/notifyprefs/prefs_test.go | 183 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/app/layout.tsx | 30 ++++++++++++++++++++++++++++++
- web/notifications/app/overview/page.tsx | 43 +++++++++++++++++++++++++++++++++++++++++++
- web/notifications/app/page.tsx | 5 +++++
- web/notifications/app/settings/page.tsx | 97 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/lib/api.ts | 45 +++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/next.config.mjs | 5 +++++
- web/notifications/package.json | 22 ++++++++++++++++++++++
- web/notifications/tsconfig.json | 21 +++++++++++++++++++++
---
## 2026-08-28 23:51 23:51 (0m)
**Beschreibung:** Claude Code Session
**Projekt:** code
### Commits
Keine Commits in dieser Session.
### Geänderte Dateien
- internal/notifyprefs/handler.go | 83 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- internal/notifyprefs/handler_test.go | 122 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- internal/notifyprefs/prefs_test.go | 183 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/app/layout.tsx | 30 ++++++++++++++++++++++++++++++
- web/notifications/app/overview/page.tsx | 43 +++++++++++++++++++++++++++++++++++++++++++
- web/notifications/app/page.tsx | 5 +++++
- web/notifications/app/settings/page.tsx | 97 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/lib/api.ts | 45 +++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/next.config.mjs | 5 +++++
- web/notifications/package.json | 22 ++++++++++++++++++++++
- web/notifications/tsconfig.json | 21 +++++++++++++++++++++
---
## 2026-08-28 23:52 23:52 (0m)
**Beschreibung:** Claude Code Session
**Projekt:** code
### Commits
Keine Commits in dieser Session.
### Geänderte Dateien
- internal/notifyprefs/handler.go | 83 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- internal/notifyprefs/handler_test.go | 122 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- internal/notifyprefs/prefs_test.go | 183 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/app/layout.tsx | 30 ++++++++++++++++++++++++++++++
- web/notifications/app/overview/page.tsx | 43 +++++++++++++++++++++++++++++++++++++++++++
- web/notifications/app/page.tsx | 5 +++++
- web/notifications/app/settings/page.tsx | 97 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/lib/api.ts | 45 +++++++++++++++++++++++++++++++++++++++++++++
- web/notifications/next.config.mjs | 5 +++++
- web/notifications/package.json | 22 ++++++++++++++++++++++
- web/notifications/tsconfig.json | 21 +++++++++++++++++++++
---
+1 -10
View File
@@ -8,7 +8,6 @@ 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() {
@@ -35,20 +34,12 @@ 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)
})
// Vorlaeufige Pfade ohne Versionierung/Auth — werden mit API-01/IAM-02 abgeloest.
// Vorlaeufiger Pfad ohne Versionierung/Auth — wird mit API-01/IAM-01 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 {
-59
View File
@@ -1,59 +0,0 @@
// notifyprefs-api ist der Aufrufpunkt fuer CFG-06: startet den bereits
// fertigen CFG-04-Handler (internal/notifyprefs) als eigenstaendigen
// HTTP-Dienst, hinter derselben Session-Auth wie IAM-16 (account-api) —
// GLEICHER JWT-Secret. REINES WIRING — keine Aenderung an
// internal/notifyprefs/.
package main
import (
"context"
"log"
"net/http"
"os"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/auth"
"gitea.perlbach24.de/scripte/nexarch/internal/notifyprefs"
)
func requireEnv(name string) string {
v := os.Getenv(name)
if v == "" {
log.Fatalf("%s muss gesetzt sein", name)
}
return v
}
func main() {
dsn := requireEnv("NEXARCH_NOTIFYPREFS_TENANT_DSN")
// MUSS identisch mit NEXARCH_ACCOUNT_JWT_SECRET (IAM-16) sein, sonst
// verwirft dieser Dienst gueltige account-api-Sessions.
jwtSecret := requireEnv("NEXARCH_NOTIFYPREFS_JWT_SECRET")
addr := os.Getenv("NEXARCH_NOTIFYPREFS_API_LISTEN_ADDR")
if addr == "" {
addr = "127.0.0.1:8101"
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, dsn)
if err != nil {
log.Fatalf("datenbankverbindung: %v", err)
}
defer pool.Close()
tokenIssuer := auth.NewTokenIssuer(jwtSecret)
handler := notifyprefs.NewHandler(notifyprefs.NewStore(pool))
mux := http.NewServeMux()
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
mux.HandleFunc("GET /notifications/preferences", auth.RequireAuth(tokenIssuer, handler.ListMine))
mux.HandleFunc("POST /notifications/preferences", auth.RequireAuth(tokenIssuer, handler.SetPreference))
mux.HandleFunc("GET /notifications/preferences/tenant", auth.RequireAuth(tokenIssuer, handler.ListTenantOverview))
log.Printf("notifyprefs-api: listening on %s", addr)
if err := http.ListenAndServe(addr, mux); err != nil {
log.Fatalf("http server: %v", err)
}
}
+70
View File
@@ -0,0 +1,70 @@
// resync-api ist der Aufrufpunkt fuer API-11: startet den bereits
// fertigen internal/resync.Handler (API-06) als eigenstaendigen
// HTTP-Dienst. REINES WIRING — keine Aenderung an internal/resync/,
// internal/audit/, internal/usage/ oder internal/moduleregistry/.
package main
import (
"context"
"log"
"net/http"
"os"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/audit"
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
"gitea.perlbach24.de/scripte/nexarch/internal/resync"
"gitea.perlbach24.de/scripte/nexarch/internal/usage"
)
// auditAdapter erfüllt resync.AuditRecorder über den bestehenden
// audit.Log-Schreibpfad — kein neuer Audit-Code, nur Signatur-Anpassung
// (audit.Log.Record nimmt ein Event-Struct, resync.AuditRecorder einzelne
// Felder).
type auditAdapter struct{ log *audit.Log }
func (a auditAdapter) Record(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any, occurredAt time.Time) error {
return a.log.Record(ctx, audit.Event{
TenantSlug: tenantSlug, Actor: actor, Action: action, Target: target,
Metadata: metadata, OccurredAt: occurredAt,
})
}
func main() {
registryDSN := os.Getenv("NEXARCH_RESYNC_REGISTRY_DSN")
if registryDSN == "" {
log.Fatal("NEXARCH_RESYNC_REGISTRY_DSN muss gesetzt sein")
}
addr := os.Getenv("NEXARCH_RESYNC_API_LISTEN_ADDR")
if addr == "" {
addr = "127.0.0.1:8098"
}
ctx := context.Background()
pool, err := pgxpool.New(ctx, registryDSN)
if err != nil {
log.Fatalf("datenbankverbindung: %v", err)
}
defer pool.Close()
flagStore := flag.NewStore(pool)
flagService := flag.NewService(flagStore, 30*time.Second)
registry := moduleregistry.NewRegistry(pool, flagService)
auditLog := audit.NewLog(pool)
usageStore := usage.NewStore(pool)
handler := resync.NewHandler(registry, auditAdapter{log: auditLog}, usageStore)
mux := http.NewServeMux()
mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
log.Printf("resync-api: listening on %s", addr)
if err := http.ListenAndServe(addr, mux); err != nil {
log.Fatalf("http server: %v", err)
}
}
@@ -1,14 +0,0 @@
[Unit]
Description=NEXARCH Core - Benachrichtigungs-Einstellungen-Dienst (CFG-04/CFG-06)
After=network.target postgresql.service
[Service]
Type=simple
User=nexarch
EnvironmentFile=/etc/nexarch/notifyprefs-api.env
ExecStart=__INSTALL_DIR__/bin/notifyprefs-api
Restart=on-failure
StandardOutput=journal
[Install]
WantedBy=multi-user.target
@@ -0,0 +1,14 @@
[Unit]
Description=NEXARCH Core - Wiederanlauf-Nachsynchronisierung (API-06/API-11)
After=network.target postgresql.service
[Service]
Type=simple
User=nexarch
EnvironmentFile=/etc/nexarch/resync-api.env
ExecStart=__INSTALL_DIR__/bin/resync-api
Restart=on-failure
StandardOutput=journal
[Install]
WantedBy=multi-user.target
+76
View File
@@ -0,0 +1,76 @@
# API-11 Prüfprotokoll: Wiederanlauf-Nachsynchronisierungs-Endpunkt starten (API-06 als laufender Dienst)
Voraussetzung API-06 bereits Fertig, hier UNVERÄNDERT.
## Reines Wiring, keine neue Logik
`git diff --stat internal/resync/ internal/audit/ internal/usage/ internal/moduleregistry/`
liefert KEINEN Diff gegenüber den jeweiligen Ticket-Ständen. `API-11`
fügt ausschließlich `cmd/resync-api/main.go` hinzu — inklusive eines
kleinen `auditAdapter`, der `resync.AuditRecorder` (einzelne Felder)
auf `audit.Log.Record` (Event-Struct) abbildet. Das ist reine
Signatur-Anpassung, keine neue Geschäftslogik.
## Root Cause (dokumentiert)
`cmd/core/main.go` ist seit TEN-01 minimal geblieben (nur `/healthz`,
`/internal/tenants`) — kein späteres Ticket (RBAC-02, CFG-02, API-06,
...) wurde je dort zentral eingehängt. Jedes Modul entstand auf einer
eigenen, unabhängigen Feature-Branch-Kette. API-11 folgt dem in dieser
Session etablierten Muster (RBAC-06, CFG-05, RET-09): ein eigener,
kleiner HTTP-Dienst statt eines zentralen `cmd/core`-Umbaus.
## Umsetzung
- `cmd/resync-api/main.go` startet `internal/resync.Handler` mit
echten Produktions-Implementierungen: `moduleregistry.Registry`
(Auth), `audit.Log` (über `auditAdapter`), `usage.Store`.
- `deploy/systemd/nexarch-resync-api.service.tmpl`.
## Prüfungen
| # | Prüfung | Ergebnis |
|---|---|---|
| 1 | Dienst startet und bleibt stabil (systemctl status aktiv) | **bestanden** real auf 131: `nexarch-resync-api.service` aktiv, `Restart=on-failure` |
| 2 | Realer POST /internal/resync/usage von einem externen Testclient gegen den laufenden Dienst liefert die erwartete Verarbeitung | **bestanden** real per `curl`: mit echtem, über `moduleregistry.Registry.Provision` ausgestelltem Service-Credential (`X-Nexarch-Client-Id`/`X-Nexarch-Client-Secret`) liefert der Aufruf `{"applied":1}`, `usage_counters` zeigt real den erhöhten Zähler; mit falschem Credential 401. Testdaten (Modul, Credential, Zähler-Zeile) anschließend entfernt |
| 3 | Code-Review: keine Änderung an internal/resync/ selbst, nur main.go+systemd neu | **bestanden** `git diff --stat` bestätigt: `internal/resync/`, `internal/audit/`, `internal/usage/`, `internal/moduleregistry/` unverändert gegenüber ihren jeweiligen Ticket-Ständen |
## Echte Verdrahtung auf 192.168.1.131
- `resync-api` gebaut nach `/opt/nexarch-core/bin/`,
`/etc/nexarch/resync-api.env` (0600), `nexarch-resync-api.service`
installiert/aktiviert.
- Reale Rechtevergabe-Lücke gefunden und behoben (gleiches Muster wie
bei den vorherigen Wrapper-Diensten): `modules`, `module_credentials`,
`audit_events`, `feature_flags`, `usage_counters`,
`resync_audit_buffer`, `resync_usage_buffer` gehörten `postgres`,
`nexarch_core` hatte keine Rechte — `GRANT` nachgezogen und über
`information_schema.role_table_grants` verifiziert, bevor der
End-zu-Ende-Test erneut lief.
- Zusätzliche reale Erkenntnis: `usage_counters.tenant_id` ist `UUID`,
nicht der Tenant-Slug (String) — beim ersten Testversuch mit `"acme"`
scheiterte der Insert intern, `UsageHandler` meldete `applied:0` statt
eines Fehlers (stiller Fehlschlag pro Delta, so von API-06 selbst so
entworfen: "Aufrufer entfernt aus seinem Puffer nur bestätigt
übernommene Deltas" — kein API-11-Defekt, sondern korrektes,
bestehendes API-06-Verhalten). Mit echter UUID als `tenant_slug`-Wert
lieferte der Aufruf real `applied:1`.
## Build/Test-Ergebnis (192.168.1.131)
```
go build ./... -> clean
go vet ./... -> clean
golangci-lint run ./cmd/resync-api/... -> 0 issues
```
Keine neuen Go-Tests nötig (kein neuer Fachcode außer main.go/Adapter,
die eigentliche Logik ist bereits durch API-06s eigene Tests
abgedeckt).
## Gesamtergebnis
**Bestanden.** API-06 ist jetzt ein real laufender, über systemd
verwalteter Dienst. Modul-Clients wie DMS' `storage.HTTPUsageReporter`
(RET-06/DOC-16-Umfeld) können sich jetzt real gegen einen laufenden
Endpunkt verdrahten, statt gegen unverdrahteten Go-Code zu testen.
-52
View File
@@ -1,52 +0,0 @@
# CFG-06 Prüfprotokoll: Benachrichtigungs-Einstellungen-Dienst starten (CFG-04 als laufender Dienst)
Voraussetzung CFG-04, IAM-16 beide bereits Fertig, hier UNVERÄNDERT.
## Reines Wiring, keine neue Logik
`git diff --stat internal/notifyprefs/` liefert KEINEN Diff.
`cmd/notifyprefs-api/main.go` setzt ausschließlich bestehende
Konstruktoren zusammen, jeder Endpunkt hinter `auth.RequireAuth` mit
demselben JWT-Secret wie IAM-16 (account-api) — dritter Baustein der
Core-GUI mit bewiesener Cross-Service-Session-Gültigkeit nach RBAC-07.
## Umsetzung
- `cmd/notifyprefs-api/main.go` `GET/POST /notifications/preferences`,
`GET /notifications/preferences/tenant`, alle hinter `auth.RequireAuth`.
- `deploy/systemd/nexarch-notifyprefs-api.service.tmpl`.
## Prüfungen
| # | Prüfung | Ergebnis |
|---|---|---|
| 1 | Dienst startet und bleibt stabil (systemctl status aktiv) | **bestanden** real auf 131: `nexarch-notifyprefs-api.service` aktiv |
| 2 | Realer Aufruf mit gültigem IAM-16-Session-Cookie setzt/liest eine echte Präferenz, ohne Cookie 401 | **bestanden** real per `curl`: ohne Cookie → 401; mit dem bei `account-api` ausgestellten Cookie: `POST /notifications/preferences` (Kanal `email` für `invoice_ready` deaktiviert) → 200, anschließendes `GET` liefert exakt diese Zeile zurück, korrekt dem einloggten Nutzer und Tenant (`acme`) zugeordnet |
| 3 | Code-Review: keine Änderung an internal/notifyprefs/ selbst, nur main.go+systemd neu | **bestanden** `git diff --stat internal/notifyprefs/` liefert leeren Diff |
## Echte Verdrahtung auf 192.168.1.131
- `notifyprefs-api` gebaut nach `/opt/nexarch-core/bin/`,
`/etc/nexarch/notifyprefs-api.env` (0600, `NEXARCH_NOTIFYPREFS_JWT_SECRET`
identisch zu `NEXARCH_ACCOUNT_JWT_SECRET`), Dienst installiert/aktiviert.
- Nutzt dieselbe `nexarch_registry`-DB/Tabellen (`notification_preferences`)
wie CFG-05 — Grants dort bereits aus CFG-05 vorhanden, keine neue
Grant-Lücke gefunden.
- End-zu-Ende-Beweis: Testnutzer angelegt, Login gegen `account-api`,
dasselbe Cookie gegen `notifyprefs-api` verwendet → echtes Setzen und
Lesen einer Präferenz. Testdaten anschließend entfernt.
## Build/Test-Ergebnis (192.168.1.131)
```
go build ./... -> clean
go vet ./... -> clean
golangci-lint run ./cmd/notifyprefs-api/... -> 0 issues
```
## Gesamtergebnis
**Bestanden.** CFG-04 ist jetzt ein real laufender, über systemd
verwalteter Dienst, cross-service authentifiziert über dieselbe
IAM-16-Session wie RBAC-07 — dritter erfolgreich real geprüfter
Auth-geschützter Baustein der Core-GUI.
+1 -1
View File
@@ -5,13 +5,13 @@ go 1.22
require (
github.com/golang-jwt/jwt/v5 v5.3.1
github.com/jackc/pgx/v5 v5.6.0
golang.org/x/crypto v0.17.0
)
require (
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
golang.org/x/crypto v0.17.0 // indirect
golang.org/x/sync v0.1.0 // indirect
golang.org/x/text v0.14.0 // indirect
)
+96
View File
@@ -0,0 +1,96 @@
// Package audit implementiert Core AUD-01: das zentrale, vom allgemeinen
// Anwendungs-Log getrennte Audit-Datenmodell fuer sicherheits- und
// compliancerelevante Ereignisse (wer, was, wann, an welchem Tenant).
// Unveraenderlichkeit (Append-only) ist AUD-02, Export/Filter-API ist AUD-03
// — dieses Paket liefert nur das Datenmodell und den EINEN zentralen
// Schreibpfad (Akzeptanzkriterium 3).
package audit
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// SystemTenant ist der reservierte Tenant-Bezug fuer mandantenuebergreifende
// Ereignisse (z.B. Superadmin-Aktionen) — es gibt bewusst KEINEN Weg, ein
// Ereignis ganz ohne Tenant-Bezug zu schreiben (Akzeptanzkriterium 2).
const SystemTenant = "system"
var ErrMissingTenant = errors.New("audit: tenant_slug darf nicht leer sein")
var ErrMissingActor = errors.New("audit: actor darf nicht leer sein")
var ErrMissingAction = errors.New("audit: action darf nicht leer sein")
// Event ist ein strukturiertes Audit-Ereignis (Akzeptanzkriterium 1: Akteur,
// Aktion, Zielobjekt, Zeitpunkt, Tenant).
type Event struct {
TenantSlug string
Actor string
Action string
Target string
Metadata map[string]any
OccurredAt time.Time
}
// Log ist der EINE zentrale Schreibpfad fuer Audit-Ereignisse — es gibt
// bewusst keine zweite Schreibmoeglichkeit, damit kein Handler versehentlich
// direkt in audit_events schreibt und dabei die Validierung umgeht
// (Akzeptanzkriterium 3).
type Log struct {
pool *pgxpool.Pool
}
func NewLog(pool *pgxpool.Pool) *Log {
return &Log{pool: pool}
}
// Record persistiert genau einen Audit-Eintrag. Fehlender Tenant-Bezug wird
// bereits hier abgewiesen (klarer Fehler statt Constraint-Verletzung im
// Normalfall) — die Datenbank-CHECK-Constraint aus der Migration ist die
// zweite, unumgehbare Verteidigungslinie (Akzeptanzkriterium 2 / Pruefung 2).
func (l *Log) Record(ctx context.Context, e Event) error {
if e.TenantSlug == "" {
return ErrMissingTenant
}
if e.Actor == "" {
return ErrMissingActor
}
if e.Action == "" {
return ErrMissingAction
}
if e.Metadata == nil {
e.Metadata = map[string]any{}
}
metadataJSON, err := json.Marshal(e.Metadata)
if err != nil {
return fmt.Errorf("metadaten serialisieren: %w", err)
}
if e.OccurredAt.IsZero() {
e.OccurredAt = time.Now()
}
_, err = l.pool.Exec(ctx, `
INSERT INTO audit_events (occurred_at, tenant_slug, actor, action, target, metadata)
VALUES ($1, $2, $3, $4, $5, $6)
`, e.OccurredAt, e.TenantSlug, e.Actor, e.Action, e.Target, metadataJSON)
if err != nil {
return fmt.Errorf("audit-ereignis schreiben: %w", err)
}
return nil
}
// CountByTenant ist eine schlanke Lesehilfe fuer Tests/Diagnose — die
// eigentliche Filter-/Export-API ist AUD-03, hier bewusst nicht vorgezogen.
func (l *Log) CountByTenant(ctx context.Context, tenantSlug string) (int, error) {
var n int
if err := l.pool.QueryRow(ctx, `
SELECT count(*) FROM audit_events WHERE tenant_slug = $1
`, tenantSlug).Scan(&n); err != nil {
return 0, fmt.Errorf("audit-ereignisse zaehlen: %w", err)
}
return n, nil
}
+132
View File
@@ -0,0 +1,132 @@
package audit
import (
"context"
"errors"
"os"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupAuditTest(t *testing.T) (*Log, *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 audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
)`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() {
_, _ = pool.Exec(ctx, `DELETE FROM audit_events WHERE tenant_slug LIKE 'test\_%' ESCAPE '\' OR tenant_slug = $1`, SystemTenant)
pool.Close()
}
return NewLog(pool), pool, cleanup
}
// Akzeptanzkriterium 1 + Pruefung 1: ein sicherheitsrelevanter Vorgang
// (hier: fehlgeschlagener Login) erzeugt zuverlaessig genau einen Eintrag.
func TestRecord_PersistsExactlyOneEventPerSecurityIncident(t *testing.T) {
log, pool, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
err := log.Record(ctx, Event{
TenantSlug: "test_acme",
Actor: "alice@example.com",
Action: "iam.login_failed",
Target: "user:alice@example.com",
Metadata: map[string]any{"reason": "falsches passwort"},
})
if err != nil {
t.Fatalf("record: %v", err)
}
count, err := log.CountByTenant(ctx, "test_acme")
if err != nil {
t.Fatalf("count: %v", err)
}
if count != 1 {
t.Fatalf("erwartet genau 1 audit-eintrag, habe %d", count)
}
var actor, action, target string
if err := pool.QueryRow(ctx, `
SELECT actor, action, target FROM audit_events WHERE tenant_slug = 'test_acme'
`).Scan(&actor, &action, &target); err != nil {
t.Fatalf("eintrag lesen: %v", err)
}
if actor != "alice@example.com" || action != "iam.login_failed" || target != "user:alice@example.com" {
t.Fatalf("eintrag unerwartet: actor=%q action=%q target=%q", actor, action, target)
}
}
// Akzeptanzkriterium 2 + Pruefung 2 (App-Ebene): fehlender Tenant-Bezug wird
// bereits vom zentralen Schreibpfad abgewiesen.
func TestRecord_RejectsMissingTenant(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
err := log.Record(ctx, Event{TenantSlug: "", Actor: "alice", Action: "irgendwas"})
if !errors.Is(err, ErrMissingTenant) {
t.Fatalf("erwartet ErrMissingTenant, habe %v", err)
}
}
// Akzeptanzkriterium 2 + Pruefung 2 (DB-Ebene): selbst ein direkter INSERT,
// der Log.Record umgeht, wird durch die CHECK-Constraint verhindert — der
// Schutz haengt nicht allein von der Go-Validierung ab.
func TestConstraint_RejectsMissingTenantAtDatabaseLevel(t *testing.T) {
_, pool, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
_, err := pool.Exec(ctx, `
INSERT INTO audit_events (tenant_slug, actor, action, target)
VALUES ('', 'alice', 'irgendwas', 'ziel')
`)
if err == nil {
t.Fatal("erwartet fehler durch CHECK-constraint bei leerem tenant_slug, habe nil")
}
}
func TestRecord_RejectsMissingActorAndAction(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
if err := log.Record(ctx, Event{TenantSlug: "test_acme", Actor: "", Action: "x"}); !errors.Is(err, ErrMissingActor) {
t.Fatalf("erwartet ErrMissingActor, habe %v", err)
}
if err := log.Record(ctx, Event{TenantSlug: "test_acme", Actor: "alice", Action: ""}); !errors.Is(err, ErrMissingAction) {
t.Fatalf("erwartet ErrMissingAction, habe %v", err)
}
}
func TestRecord_SystemTenantForCrossTenantEvents(t *testing.T) {
log, _, cleanup := setupAuditTest(t)
defer cleanup()
ctx := context.Background()
if err := log.Record(ctx, Event{TenantSlug: SystemTenant, Actor: "superadmin", Action: "tenant.provisioned", Target: "tenant:acme"}); err != nil {
t.Fatalf("record mit SystemTenant: %v", err)
}
}
-67
View File
@@ -1,67 +0,0 @@
package auth
import (
"encoding/json"
"net/http"
"time"
)
// Handler stellt Login/Logout als HTTP-Endpunkte bereit. Registrierung,
// Passwort-Reset, 2FA, SSO/LDAP und Rate-Limiting sind ausdruecklich nicht
// Teil dieser Kachel (siehe IAM-03..07).
type Handler struct {
login *LoginService
}
func NewHandler(login *LoginService) *Handler {
return &Handler{login: login}
}
type loginRequest struct {
Email string `json:"email"`
Password string `json:"password"`
}
func (h *Handler) Login(w http.ResponseWriter, r *http.Request) {
var req loginRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "ungueltige Anfrage", http.StatusBadRequest)
return
}
token, err := h.login.Login(r.Context(), req.Email, req.Password)
if err != nil {
http.Error(w, ErrInvalidCredentials.Error(), http.StatusUnauthorized)
return
}
http.SetCookie(w, &http.Cookie{
Name: CookieName,
Value: token,
Path: "/",
HttpOnly: true,
Secure: true,
SameSite: http.SameSiteStrictMode,
MaxAge: int(AccessTokenTTL.Seconds()),
})
w.WriteHeader(http.StatusOK)
}
// Logout loescht das Session-Cookie. Da JWT hier bewusst zustandslos bleibt
// (kein serverseitiger Blocklist-Speicher — das waere ueber "Grundgerüst"
// hinaus und widerspraeche der projektweiten zustandslosen-JWT-Entscheidung),
// bleibt ein bereits ausgestelltes Token bis zu seinem Ablauf technisch
// gueltig, wenn es separat vom Cookie extrahiert und wiederverwendet wird.
func (h *Handler) Logout(w http.ResponseWriter, r *http.Request) {
http.SetCookie(w, &http.Cookie{
Name: CookieName,
Value: "",
Path: "/",
HttpOnly: true,
Secure: true,
SameSite: http.SameSiteStrictMode,
MaxAge: -1,
Expires: time.Unix(0, 0),
})
w.WriteHeader(http.StatusOK)
}
-56
View File
@@ -1,56 +0,0 @@
package auth
import (
"context"
"errors"
"gitea.perlbach24.de/scripte/nexarch/internal/user"
)
var ErrInvalidCredentials = errors.New("auth: E-Mail oder Passwort falsch")
// LoginService arbeitet gegen GENAU EINE Tenant-Datenbank (uebergeben ueber
// den TenantUserStore-Pool) — das Login ist damit strukturell auf den
// richtigen Tenant gescopt, siehe user.TenantUserStore.GetByEmailForAuth.
type LoginService struct {
users *user.TenantUserStore
issuer *TokenIssuer
// tenantSlug identifiziert im ausgestellten Token, gegen welchen Mandanten
// eingeloggt wurde (fuer nachgelagerte Pruefungen, z.B. Middleware-Logs).
tenantSlug string
}
func NewLoginService(users *user.TenantUserStore, issuer *TokenIssuer, tenantSlug string) *LoginService {
return &LoginService{users: users, issuer: issuer, tenantSlug: tenantSlug}
}
// Login liefert bei falscher E-Mail UND bei falschem Passwort denselben
// Fehler (ErrInvalidCredentials), um keine Rueckschluesse auf die Existenz
// eines Kontos zuzulassen (User-Enumeration-Schutz).
func (s *LoginService) Login(ctx context.Context, email, password string) (string, error) {
creds, err := s.users.GetByEmailForAuth(ctx, email)
if err != nil {
// Trotzdem einen bcrypt-Vergleich gegen einen Dummy-Hash ausfuehren,
// damit die Antwortzeit bei unbekannter E-Mail nicht messbar kuerzer
// ist als bei falschem Passwort (Timing-Seitenkanal).
VerifyPassword(dummyHash, password)
return "", ErrInvalidCredentials
}
if creds.User.Status != user.StatusActive {
return "", ErrInvalidCredentials
}
if !VerifyPassword(creds.PasswordHash, password) {
return "", ErrInvalidCredentials
}
return s.issuer.Issue(creds.User.ID, s.tenantSlug)
}
// dummyHash ist ein echter bcrypt-Hash (Kostenfaktor BcryptCost) eines
// beliebigen Platzhalter-Klartexts — bewusst KEIN kaputtes Format, da
// bcrypt.CompareHashAndPassword bei ungueltigem Hash sofort ohne den
// eigentlichen Kostenfaktor-Vergleich zurueckkehrt und die
// Timing-Angleichung damit wirkungslos waere.
const dummyHash = "$2a$12$cmwiETrG9DK5/uTM2fg4uetngYUspKjME5P8fNpk0QYTaO64N0r3C"
-177
View File
@@ -1,177 +0,0 @@
package auth
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/user"
)
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',
password_hash TEXT NOT NULL DEFAULT '',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
);`
func setupTenantDB(t *testing.T, dbName 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, usersSchema); 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
}
func createUserWithPassword(t *testing.T, store *user.TenantUserStore, email, password string) user.User {
t.Helper()
ctx := context.Background()
u, err := store.Create(ctx, email, "Test User")
if err != nil {
t.Fatalf("create user: %v", err)
}
hash, err := HashPassword(password)
if err != nil {
t.Fatalf("hash password: %v", err)
}
if err := store.SetPasswordHash(ctx, u.ID, hash); err != nil {
t.Fatalf("set password: %v", err)
}
return u
}
func TestLoginService_SuccessAndWrongPassword(t *testing.T) {
pool := setupTenantDB(t, "test_iam02_login")
store := user.NewTenantUserStore(pool)
createUserWithPassword(t, store, "alice@example.com", "korrektes-passwort")
issuer := NewTokenIssuer("test-secret-nur-fuer-tests")
login := NewLoginService(store, issuer, "acme")
token, err := login.Login(context.Background(), "alice@example.com", "korrektes-passwort")
if err != nil {
t.Fatalf("login: %v", err)
}
if token == "" {
t.Fatal("erwartet nicht-leeres token")
}
if _, err := login.Login(context.Background(), "alice@example.com", "falsches-passwort"); !errors.Is(err, ErrInvalidCredentials) {
t.Fatalf("erwartet ErrInvalidCredentials, habe %v", err)
}
if _, err := login.Login(context.Background(), "unbekannt@example.com", "irgendwas"); !errors.Is(err, ErrInvalidCredentials) {
t.Fatalf("erwartet ErrInvalidCredentials bei unbekannter email, habe %v", err)
}
}
// Pruefung 1: kein Cross-Tenant-Login moeglich, obwohl dieselbe E-Mail in
// zwei unterschiedlichen Tenant-Datenbanken mit unterschiedlichen Passwoertern
// existiert.
func TestLoginService_NoCrossTenantLogin(t *testing.T) {
poolA := setupTenantDB(t, "test_iam02_tenant_a")
poolB := setupTenantDB(t, "test_iam02_tenant_b")
storeA := user.NewTenantUserStore(poolA)
storeB := user.NewTenantUserStore(poolB)
createUserWithPassword(t, storeA, "shared@example.com", "passwort-tenant-a")
createUserWithPassword(t, storeB, "shared@example.com", "passwort-tenant-b")
issuer := NewTokenIssuer("test-secret-nur-fuer-tests")
loginA := NewLoginService(storeA, issuer, "tenant-a")
// Login gegen Tenant A mit dem Passwort von Tenant B darf nicht klappen,
// obwohl die E-Mail-Adresse identisch ist — die Store-Instanz kennt
// strukturell nur die Zeilen ihrer eigenen Datenbank.
if _, err := loginA.Login(context.Background(), "shared@example.com", "passwort-tenant-b"); !errors.Is(err, ErrInvalidCredentials) {
t.Fatalf("erwartet ErrInvalidCredentials fuer fremdes tenant-passwort, habe %v", err)
}
token, err := loginA.Login(context.Background(), "shared@example.com", "passwort-tenant-a")
if err != nil {
t.Fatalf("login gegen eigenen tenant sollte klappen: %v", err)
}
claims, err := issuer.Verify(token)
if err != nil {
t.Fatalf("verify: %v", err)
}
if claims.TenantSlug != "tenant-a" {
t.Fatalf("token tenant = %q, want tenant-a", claims.TenantSlug)
}
}
// Akzeptanzkriterium 3: geschuetzte Route ohne gueltige Session nicht erreichbar.
func TestRequireAuth_BlocksWithoutValidCookie(t *testing.T) {
issuer := NewTokenIssuer("test-secret-nur-fuer-tests")
protected := RequireAuth(issuer, func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
})
// Kein Cookie.
req := httptest.NewRequest(http.MethodGet, "/geschuetzt", nil)
rec := httptest.NewRecorder()
protected(rec, req)
if rec.Code != http.StatusUnauthorized {
t.Fatalf("ohne cookie: status = %d, want 401", rec.Code)
}
// Manipuliertes Cookie.
req = httptest.NewRequest(http.MethodGet, "/geschuetzt", nil)
req.AddCookie(&http.Cookie{Name: CookieName, Value: "kaputt.token.hier"})
rec = httptest.NewRecorder()
protected(rec, req)
if rec.Code != http.StatusUnauthorized {
t.Fatalf("mit kaputtem cookie: status = %d, want 401", rec.Code)
}
// Gueltiges Token.
token, err := issuer.Issue("user-1", "acme")
if err != nil {
t.Fatalf("issue: %v", err)
}
req = httptest.NewRequest(http.MethodGet, "/geschuetzt", nil)
req.AddCookie(&http.Cookie{Name: CookieName, Value: token})
rec = httptest.NewRecorder()
protected(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("mit gueltigem cookie: status = %d, want 200", rec.Code)
}
}
-41
View File
@@ -1,41 +0,0 @@
package auth
import (
"context"
"net/http"
)
const CookieName = "nexarch_session"
type contextKey int
const claimsContextKey contextKey = iota
// RequireAuth schuetzt eine Route: ohne gueltiges, nicht abgelaufenes Token
// im Session-Cookie wird 401 zurueckgegeben und der Handler nicht aufgerufen
// (IAM-02 Akzeptanzkriterium 3).
func RequireAuth(issuer *TokenIssuer, next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
cookie, err := r.Cookie(CookieName)
if err != nil {
http.Error(w, "nicht angemeldet", http.StatusUnauthorized)
return
}
claims, err := issuer.Verify(cookie.Value)
if err != nil {
http.Error(w, "nicht angemeldet", http.StatusUnauthorized)
return
}
ctx := context.WithValue(r.Context(), claimsContextKey, claims)
next(w, r.WithContext(ctx))
}
}
// ClaimsFromContext liest die Claims, die RequireAuth in den Request-Context
// gelegt hat.
func ClaimsFromContext(ctx context.Context) (*Claims, bool) {
c, ok := ctx.Value(claimsContextKey).(*Claims)
return c, ok
}
-28
View File
@@ -1,28 +0,0 @@
// Package auth implementiert Core IAM-02: Login/Logout, Passwort-Hashing und
// die Middleware zum Schutz von Routen. Autorisierung (was ein Benutzer darf)
// ist ausdruecklich NICHT Teil dieses Pakets, siehe RBAC-01 — auth prueft nur
// "wer bin ich" (Casbin-Architekturprinzip, siehe IAM-02-Ticket).
package auth
import "golang.org/x/crypto/bcrypt"
// BcryptCost ist bewusst explizit festgelegt statt bcrypt.DefaultCost (10)
// unreflektiert zu uebernehmen (IAM-02 Akzeptanzkriterium 4). Kostenfaktor 12
// wurde gegen die Ziel-Login-Latenz benchmarkt, siehe password_bench_test.go
// und den Pruefungs-Eintrag in der Commit-Nachricht.
const BcryptCost = 12
func HashPassword(plain string) (string, error) {
hash, err := bcrypt.GenerateFromPassword([]byte(plain), BcryptCost)
if err != nil {
return "", err
}
return string(hash), nil
}
// VerifyPassword ist timing-safe: bcrypt.CompareHashAndPassword vergleicht
// konstant in der Zeit bzgl. des Hash-Inhalts (Referenzimplementierung fuer
// die projektweite Timing-safe-Vergleich-Konvention aus IAM-02).
func VerifyPassword(hash, plain string) bool {
return bcrypt.CompareHashAndPassword([]byte(hash), []byte(plain)) == nil
}
-42
View File
@@ -1,42 +0,0 @@
package auth
import (
"testing"
"time"
)
// TargetLoginLatency ist der Zielwert aus IAM-02 Akzeptanzkriterium 4: der
// bcrypt-Vergleich allein darf die Login-Latenz nicht dominieren. 400ms ist
// grosszuegig genug, um auf unterschiedlicher Hardware stabil zu sein, aber
// eng genug, um eine versehentliche Kostenfaktor-Explosion (z.B. 16 statt 12)
// zuverlaessig aufzudecken.
const TargetLoginLatency = 400 * time.Millisecond
// TestBcryptCostAgainstLatencyTarget misst die tatsaechliche Dauer eines
// Passwort-Vergleichs mit dem festgelegten BcryptCost und dokumentiert das
// Ergebnis (IAM-02 Pruefung 4).
func TestBcryptCostAgainstLatencyTarget(t *testing.T) {
hash, err := HashPassword("benchmark-passwort")
if err != nil {
t.Fatalf("hash: %v", err)
}
start := time.Now()
if !VerifyPassword(hash, "benchmark-passwort") {
t.Fatal("verifikation haette erfolgreich sein muessen")
}
elapsed := time.Since(start)
t.Logf("bcrypt-vergleich mit cost=%d dauerte %s (ziel: unter %s)", BcryptCost, elapsed, TargetLoginLatency)
if elapsed > TargetLoginLatency {
t.Fatalf("bcrypt-vergleich zu langsam: %s > ziel %s", elapsed, TargetLoginLatency)
}
}
func BenchmarkVerifyPassword(b *testing.B) {
hash, _ := HashPassword("benchmark-passwort")
b.ResetTimer()
for i := 0; i < b.N; i++ {
VerifyPassword(hash, "benchmark-passwort")
}
}
-28
View File
@@ -1,28 +0,0 @@
package auth
import "testing"
func TestHashAndVerifyPassword(t *testing.T) {
hash, err := HashPassword("s3hr-geheim!")
if err != nil {
t.Fatalf("hash: %v", err)
}
if hash == "s3hr-geheim!" {
t.Fatal("passwort wurde nicht gehasht")
}
if !VerifyPassword(hash, "s3hr-geheim!") {
t.Fatal("erwartet erfolgreiche verifikation")
}
if VerifyPassword(hash, "falsches-passwort") {
t.Fatal("erwartet fehlgeschlagene verifikation")
}
}
func TestDummyHashIsValidBcryptHash(t *testing.T) {
// Stellt sicher, dass der Timing-Angleichs-Hash in login.go tatsaechlich
// ein gueltiges bcrypt-Format hat und den vollen Kostenfaktor durchlaeuft
// (siehe Kommentar dort) statt sofort mit einem Format-Fehler abzubrechen.
if VerifyPassword(dummyHash, "irgendein-text") {
t.Fatal("dummyHash sollte fuer beliebigen text nicht passen")
}
}
-63
View File
@@ -1,63 +0,0 @@
package auth
import (
"errors"
"time"
"github.com/golang-jwt/jwt/v5"
)
// AccessTokenTTL ist bewusst kurz gehalten (Session-Ablauf statt langlebiger
// Tokens), passend zur "so vertrauenswuerdig wie noetig"-Produkt-DNA.
const AccessTokenTTL = 30 * time.Minute
var ErrInvalidToken = errors.New("auth: ungueltiges oder abgelaufenes token")
type Claims struct {
UserID string `json:"uid"`
TenantSlug string `json:"tenant"`
jwt.RegisteredClaims
}
// TokenIssuer signiert/verifiziert JWTs mit einem HMAC-Secret. Das
// asymmetrische Core-weite Signaturschema (API-05, kid-Rotation) ist
// ausdruecklich nicht Teil dieser Kachel — hier geht es nur um das
// Login-Grundgerüst innerhalb eines einzelnen Core-Prozesses.
type TokenIssuer struct {
secret []byte
}
func NewTokenIssuer(secret string) *TokenIssuer {
return &TokenIssuer{secret: []byte(secret)}
}
func (i *TokenIssuer) Issue(userID, tenantSlug string) (string, error) {
now := time.Now()
claims := Claims{
UserID: userID,
TenantSlug: tenantSlug,
RegisteredClaims: jwt.RegisteredClaims{
IssuedAt: jwt.NewNumericDate(now),
ExpiresAt: jwt.NewNumericDate(now.Add(AccessTokenTTL)),
},
}
token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
return token.SignedString(i.secret)
}
// Verify prueft Signatur UND Ablauf (jwt.ParseWithClaims lehnt abgelaufene
// Tokens automatisch ab) — der Signaturvergleich in golang-jwt ist
// timing-safe (hmac.Equal).
func (i *TokenIssuer) Verify(tokenString string) (*Claims, error) {
claims := &Claims{}
token, err := jwt.ParseWithClaims(tokenString, claims, func(t *jwt.Token) (interface{}, error) {
if _, ok := t.Method.(*jwt.SigningMethodHMAC); !ok {
return nil, ErrInvalidToken
}
return i.secret, nil
})
if err != nil || !token.Valid {
return nil, ErrInvalidToken
}
return claims, nil
}
-82
View File
@@ -1,82 +0,0 @@
package auth
import (
"strings"
"testing"
"time"
"github.com/golang-jwt/jwt/v5"
)
func TestTokenIssueAndVerify(t *testing.T) {
issuer := NewTokenIssuer("test-secret-nur-fuer-tests")
token, err := issuer.Issue("user-1", "acme")
if err != nil {
t.Fatalf("issue: %v", err)
}
claims, err := issuer.Verify(token)
if err != nil {
t.Fatalf("verify: %v", err)
}
if claims.UserID != "user-1" || claims.TenantSlug != "acme" {
t.Fatalf("claims unerwartet: %+v", claims)
}
}
// Pruefung 2: Token-Manipulationstest.
func TestTokenVerify_RejectsManipulatedPayload(t *testing.T) {
issuer := NewTokenIssuer("test-secret-nur-fuer-tests")
token, err := issuer.Issue("user-1", "acme")
if err != nil {
t.Fatalf("issue: %v", err)
}
parts := strings.Split(token, ".")
if len(parts) != 3 {
t.Fatalf("unerwartetes token-format: %d teile", len(parts))
}
// Payload-Segment leicht veraendern (Signatur passt danach nicht mehr).
tampered := parts[0] + "." + parts[1] + "x" + "." + parts[2]
if _, err := issuer.Verify(tampered); err == nil {
t.Fatal("erwartet fehler bei manipuliertem token, habe nil")
}
}
func TestTokenVerify_RejectsWrongSecret(t *testing.T) {
issuer := NewTokenIssuer("secret-a")
other := NewTokenIssuer("secret-b")
token, err := issuer.Issue("user-1", "acme")
if err != nil {
t.Fatalf("issue: %v", err)
}
if _, err := other.Verify(token); err == nil {
t.Fatal("erwartet fehler bei falschem secret, habe nil")
}
}
// Pruefung 3: abgelaufenes Token erzwingt Neuanmeldung.
func TestTokenVerify_RejectsExpiredToken(t *testing.T) {
issuer := NewTokenIssuer("test-secret-nur-fuer-tests")
claims := Claims{
UserID: "user-1",
TenantSlug: "acme",
RegisteredClaims: jwt.RegisteredClaims{
IssuedAt: jwt.NewNumericDate(time.Now().Add(-2 * AccessTokenTTL)),
ExpiresAt: jwt.NewNumericDate(time.Now().Add(-time.Minute)),
},
}
tok := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
expired, err := tok.SignedString([]byte("test-secret-nur-fuer-tests"))
if err != nil {
t.Fatalf("signieren: %v", err)
}
if _, err := issuer.Verify(expired); err == nil {
t.Fatal("erwartet fehler bei abgelaufenem token, habe nil")
}
}
-88
View File
@@ -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()
}
-114
View File
@@ -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)
}
}
-129
View File
@@ -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(&currentVersion)
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()
}
-119
View File
@@ -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)
}
}
-207
View File
@@ -1,207 +0,0 @@
package channels
import (
"context"
"errors"
"fmt"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
)
func setupTest(t *testing.T) (*TemplateStore, *InAppStore, 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_templates (
tenant_slug TEXT NOT NULL, key TEXT NOT NULL, subject TEXT NOT NULL, body TEXT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (tenant_slug, key)
);
CREATE TABLE IF NOT EXISTS in_app_notifications (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), tenant_slug TEXT NOT NULL, user_id TEXT NOT NULL,
title TEXT NOT NULL, body TEXT NOT NULL, read_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
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 NewTemplateStore(pool), NewInAppStore(pool), cleanup
}
func uniqueKey(prefix string) string {
return fmt.Sprintf("%s_%d", prefix, time.Now().UnixNano())
}
// Akzeptanzkriterium 1 + Pruefung 1: Vorlagenrendering mit fehlenden
// Platzhaltern bricht kontrolliert ab.
func TestRender_MissingPlaceholderAborts(t *testing.T) {
tmpl := Template{Subject: "Hallo {{.name}}", Body: "Dein Code: {{.code}}"}
_, _, err := Render(tmpl, map[string]any{"name": "Alice"}) // "code" fehlt
if err == nil {
t.Fatal("erwartet fehler bei fehlendem platzhalter 'code'")
}
}
func TestRender_SucceedsWithAllPlaceholders(t *testing.T) {
tmpl := Template{Subject: "Hallo {{.name}}", Body: "Dein Code: {{.code}}"}
subject, body, err := Render(tmpl, map[string]any{"name": "Alice", "code": "1234"})
if err != nil {
t.Fatalf("render: %v", err)
}
if subject != "Hallo Alice" || body != "Dein Code: 1234" {
t.Fatalf("unerwartet: subject=%q body=%q", subject, body)
}
}
// Akzeptanzkriterium 3: Vorlagen pro Tenant anpassbar, Fallback auf global.
func TestTemplateStore_TenantOverrideFallsBackToGlobal(t *testing.T) {
templates, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
key := uniqueKey("welcome")
if err := templates.Set(ctx, GlobalTemplateScope, key, "Willkommen", "Standardtext"); err != nil {
t.Fatalf("set global: %v", err)
}
if err := templates.Set(ctx, "acme", key, "Willkommen bei ACME", "ACME-Text"); err != nil {
t.Fatalf("set tenant: %v", err)
}
got, err := templates.Resolve(ctx, "acme", key)
if err != nil {
t.Fatalf("resolve acme: %v", err)
}
if got.Subject != "Willkommen bei ACME" {
t.Fatalf("erwartet tenant-vorlage, habe %q", got.Subject)
}
got, err = templates.Resolve(ctx, "globex", key) // hat keine eigene vorlage
if err != nil {
t.Fatalf("resolve globex: %v", err)
}
if got.Subject != "Willkommen" {
t.Fatalf("erwartet global-fallback, habe %q", got.Subject)
}
}
// Akzeptanzkriterium 2 + Pruefung 2: als gelesen markiert wird korrekt gefuehrt.
func TestInAppStore_MarkReadIsReflectedCorrectly(t *testing.T) {
_, inApp, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
userID := uniqueKey("user")
id, err := inApp.Create(ctx, "acme", userID, "Titel", "Text")
if err != nil {
t.Fatalf("create: %v", err)
}
list, err := inApp.ListForUser(ctx, "acme", userID)
if err != nil {
t.Fatalf("list (vor markierung): %v", err)
}
if len(list) != 1 || list[0].ReadAt != nil {
t.Fatalf("erwartet 1 ungelesene benachrichtigung, habe %+v", list)
}
if err := inApp.MarkRead(ctx, id); err != nil {
t.Fatalf("mark read: %v", err)
}
list, err = inApp.ListForUser(ctx, "acme", userID)
if err != nil {
t.Fatalf("list (nach markierung): %v", err)
}
if len(list) != 1 || list[0].ReadAt == nil {
t.Fatalf("erwartet als gelesen markiert, habe %+v", list)
}
}
// EmailSender bricht ab, BEVOR SMTP ueberhaupt kontaktiert wird, wenn keine
// Vorlage aufloesbar ist — Nachweis, dass der Abbruch vor dem Netzwerkzugriff
// erfolgt (Akzeptanzkriterium 1).
func TestEmailSender_AbortsBeforeSMTPWhenTemplateMissing(t *testing.T) {
templates, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
sender := NewEmailSender(templates, "nicht-aufloesbarer-smtp-host.invalid", "25", "noreply@example.com")
err := sender.Send(ctx, notify.Notification{
Recipient: "user@example.com",
Payload: map[string]any{"template_key": uniqueKey("nie_konfiguriert"), "tenant_slug": "acme"},
})
if !errors.Is(err, ErrTemplateNotFound) {
t.Fatalf("erwartet ErrTemplateNotFound (kein smtp-versuch), habe %v", err)
}
}
// Akzeptanzkriterium 3 + Pruefung 3: E-Mail-Versand bei nicht erreichbarem
// SMTP-Server loest das dokumentierte Retry-Verhalten ueber CFG-02 aus.
func TestEmailSender_TriggersDispatcherRetryOnUnreachableSMTP(t *testing.T) {
templates, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
adminDSN := os.Getenv("TEST_ADMIN_DSN")
pool, err := pgxpool.New(ctx, adminDSN)
if err != nil {
t.Fatalf("pool: %v", err)
}
defer pool.Close()
key := uniqueKey("retry_test")
if err := templates.Set(ctx, GlobalTemplateScope, key, "Betreff", "Text ohne Platzhalter"); err != nil {
t.Fatalf("set template: %v", err)
}
sender := NewEmailSender(templates, "nicht-aufloesbarer-smtp-host.invalid", "25", "noreply@example.com")
dispatcher := notify.NewDispatcher(pool).WithRetryPolicy(2, time.Millisecond)
jobID, err := dispatcher.Enqueue(ctx, "email", "user@example.com", map[string]any{"template_key": key, "tenant_slug": "acme"})
if err != nil {
t.Fatalf("enqueue: %v", err)
}
for i := 0; i < 2; i++ {
time.Sleep(5 * time.Millisecond)
if _, _, err := dispatcher.ProcessDue(ctx, sender, 10); err != nil {
t.Fatalf("process due %d: %v", i, err)
}
}
var status string
var attempts int
if err := pool.QueryRow(ctx, `SELECT status, attempts FROM notification_jobs WHERE id = $1`, jobID).Scan(&status, &attempts); err != nil {
t.Fatalf("status lesen: %v", err)
}
if status != "failed" {
t.Fatalf("erwartet status failed nach ausgeschoepften wiederholungen bei unerreichbarem smtp, habe %q", status)
}
if attempts != 2 {
t.Fatalf("erwartet 2 versuche, habe %d", attempts)
}
}
-81
View File
@@ -1,81 +0,0 @@
package channels
import (
"context"
"fmt"
"net/smtp"
"os"
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
)
// EmailSender implementiert notify.Sender fuer den E-Mail-Kanal
// (Akzeptanzkriterium 1). SMTP-Zugangsdaten kommen ausschliesslich aus
// Umgebungsvariablen, nie aus Code/DB.
type EmailSender struct {
templates *TemplateStore
host string
port string
from string
username string
password string
}
// NewEmailSenderFromEnv liest NEXARCH_SMTP_HOST/PORT/FROM (Pflicht) sowie
// optional NEXARCH_SMTP_USER/PASSWORD.
func NewEmailSenderFromEnv(templates *TemplateStore) (*EmailSender, error) {
host := os.Getenv("NEXARCH_SMTP_HOST")
port := os.Getenv("NEXARCH_SMTP_PORT")
from := os.Getenv("NEXARCH_SMTP_FROM")
if host == "" || port == "" || from == "" {
return nil, fmt.Errorf("channels: NEXARCH_SMTP_HOST/PORT/FROM muessen gesetzt sein")
}
return &EmailSender{
templates: templates,
host: host,
port: port,
from: from,
username: os.Getenv("NEXARCH_SMTP_USER"),
password: os.Getenv("NEXARCH_SMTP_PASSWORD"),
}, nil
}
// NewEmailSender erlaubt Tests, Host/Port explizit zu setzen (z.B. einen
// absichtlich nicht erreichbaren Host fuer den Retry-Nachweis), ohne
// Umgebungsvariablen zu benoetigen.
func NewEmailSender(templates *TemplateStore, host, port, from string) *EmailSender {
return &EmailSender{templates: templates, host: host, port: port, from: from}
}
// Send rendert zuerst die Vorlage — schlaegt das fehl (z.B. fehlender
// Platzhalter), wird NIE eine SMTP-Verbindung aufgebaut
// (Akzeptanzkriterium 1 / Pruefung 1: kontrollierter Abbruch statt
// fehlerhafter Mail). Ein danach fehlschlagender SMTP-Versand liefert einen
// Fehler zurueck, den CFG-02s Dispatcher fuer die bereits getestete
// Wiederholungslogik nutzt (Akzeptanzkriterium 3 / Pruefung 3) — kein
// zweiter Retry-Mechanismus hier.
func (e *EmailSender) Send(ctx context.Context, n notify.Notification) error {
templateKey, _ := n.Payload["template_key"].(string)
tenantSlug, _ := n.Payload["tenant_slug"].(string)
tmpl, err := e.templates.Resolve(ctx, tenantSlug, templateKey)
if err != nil {
return fmt.Errorf("vorlage aufloesen: %w", err)
}
subject, body, err := Render(tmpl, n.Payload)
if err != nil {
return err
}
msg := []byte("Subject: " + subject + "\r\n\r\n" + body)
var auth smtp.Auth
if e.username != "" {
auth = smtp.PlainAuth("", e.username, e.password, e.host)
}
addr := e.host + ":" + e.port
if err := smtp.SendMail(addr, auth, e.from, []string{n.Recipient}, msg); err != nil {
return fmt.Errorf("smtp-versand fehlgeschlagen: %w", err)
}
return nil
}
-74
View File
@@ -1,74 +0,0 @@
package channels
import (
"context"
"fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
type InAppNotification struct {
ID string
TenantSlug string
UserID string
Title string
Body string
ReadAt *time.Time
CreatedAt time.Time
}
// InAppStore persistiert In-App-Benachrichtigungen (Akzeptanzkriterium 2).
type InAppStore struct {
pool *pgxpool.Pool
}
func NewInAppStore(pool *pgxpool.Pool) *InAppStore {
return &InAppStore{pool: pool}
}
func (s *InAppStore) Create(ctx context.Context, tenantSlug, userID, title, body string) (string, error) {
var id string
err := s.pool.QueryRow(ctx, `
INSERT INTO in_app_notifications (tenant_slug, user_id, title, body)
VALUES ($1, $2, $3, $4)
RETURNING id
`, tenantSlug, userID, title, body).Scan(&id)
if err != nil {
return "", fmt.Errorf("in-app-benachrichtigung speichern: %w", err)
}
return id, nil
}
// ListForUser liefert alle Benachrichtigungen eines Benutzers (ueber API
// abrufbar, Akzeptanzkriterium 2).
func (s *InAppStore) ListForUser(ctx context.Context, tenantSlug, userID string) ([]InAppNotification, error) {
rows, err := s.pool.Query(ctx, `
SELECT id, title, body, read_at, created_at FROM in_app_notifications
WHERE tenant_slug = $1 AND user_id = $2 ORDER BY created_at DESC
`, tenantSlug, userID)
if err != nil {
return nil, fmt.Errorf("benachrichtigungen auflisten: %w", err)
}
defer rows.Close()
var out []InAppNotification
for rows.Next() {
n := InAppNotification{TenantSlug: tenantSlug, UserID: userID}
if err := rows.Scan(&n.ID, &n.Title, &n.Body, &n.ReadAt, &n.CreatedAt); err != nil {
return nil, fmt.Errorf("benachrichtigung lesen: %w", err)
}
out = append(out, n)
}
return out, rows.Err()
}
// MarkRead markiert eine Benachrichtigung als gelesen (Akzeptanzkriterium 2
// / Pruefung 2).
func (s *InAppStore) MarkRead(ctx context.Context, id string) error {
_, err := s.pool.Exec(ctx, `UPDATE in_app_notifications SET read_at = now() WHERE id = $1`, id)
if err != nil {
return fmt.Errorf("als gelesen markieren: %w", err)
}
return nil
}
-41
View File
@@ -1,41 +0,0 @@
package channels
import (
"context"
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
)
// InAppSender implementiert notify.Sender fuer den In-App-Kanal
// (Akzeptanzkriterium 2). Nutzt eine Vorlage, falls payload["template_key"]
// gesetzt ist, sonst direkt payload["title"]/["body"].
type InAppSender struct {
store *InAppStore
templates *TemplateStore
}
func NewInAppSender(store *InAppStore, templates *TemplateStore) *InAppSender {
return &InAppSender{store: store, templates: templates}
}
func (s *InAppSender) Send(ctx context.Context, n notify.Notification) error {
tenantSlug, _ := n.Payload["tenant_slug"].(string)
var title, body string
if templateKey, ok := n.Payload["template_key"].(string); ok && templateKey != "" {
tmpl, err := s.templates.Resolve(ctx, tenantSlug, templateKey)
if err != nil {
return err
}
title, body, err = Render(tmpl, n.Payload)
if err != nil {
return err
}
} else {
title, _ = n.Payload["title"].(string)
body, _ = n.Payload["body"].(string)
}
_, err := s.store.Create(ctx, tenantSlug, n.Recipient, title, body)
return err
}
-32
View File
@@ -1,32 +0,0 @@
package channels
import (
"context"
"fmt"
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
)
// Router waehlt anhand von Notification.Channel den zustaendigen Kanal aus
// — ein neuer Kanal wird per Register() ergaenzt, ohne den Dispatcher
// (CFG-02) oder Router selbst umzubauen (Unleash-artiger Strategie-Gedanke,
// siehe Ticket-DNA).
type Router struct {
channels map[string]notify.Sender
}
func NewRouter() *Router {
return &Router{channels: make(map[string]notify.Sender)}
}
func (r *Router) Register(channel string, sender notify.Sender) {
r.channels[channel] = sender
}
func (r *Router) Send(ctx context.Context, n notify.Notification) error {
sender, ok := r.channels[n.Channel]
if !ok {
return fmt.Errorf("channels: unbekannter kanal %q", n.Channel)
}
return sender.Send(ctx, n)
}
-104
View File
@@ -1,104 +0,0 @@
// Package channels implementiert Core CFG-03: konkrete Zustellkanaele fuer
// den CFG-02-Dispatcher (E-Mail, In-App) inklusive Vorlagenverwaltung.
// Neue Kanaele lassen sich ergaenzen, ohne den Dispatcher selbst
// anzufassen — jeder Kanal implementiert nur notify.Sender.
package channels
import (
"bytes"
"context"
"errors"
"fmt"
"text/template"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
// GlobalTemplateScope ist der Fallback-Wert, wenn ein Tenant keine eigene
// Vorlage konfiguriert hat (Akzeptanzkriterium 3).
const GlobalTemplateScope = "global"
var ErrTemplateNotFound = errors.New("channels: keine vorlage gefunden")
type Template struct {
Subject string
Body string
}
type TemplateStore struct {
pool *pgxpool.Pool
}
func NewTemplateStore(pool *pgxpool.Pool) *TemplateStore {
return &TemplateStore{pool: pool}
}
// Set legt eine Vorlage fuer einen Tenant (oder GlobalTemplateScope) fest.
func (s *TemplateStore) Set(ctx context.Context, tenantSlug, key, subject, body string) error {
_, err := s.pool.Exec(ctx, `
INSERT INTO notification_templates (tenant_slug, key, subject, body, updated_at)
VALUES ($1, $2, $3, $4, now())
ON CONFLICT (tenant_slug, key) DO UPDATE SET subject = $3, body = $4, updated_at = now()
`, tenantSlug, key, subject, body)
if err != nil {
return fmt.Errorf("vorlage speichern: %w", err)
}
return nil
}
// Resolve liefert die Vorlage eines Tenants, faellt auf GlobalTemplateScope
// zurueck, wenn der Tenant keine eigene gesetzt hat (Akzeptanzkriterium 3).
func (s *TemplateStore) Resolve(ctx context.Context, tenantSlug, key string) (Template, error) {
if tenantSlug != "" && tenantSlug != GlobalTemplateScope {
if t, err := s.get(ctx, tenantSlug, key); err == nil {
return t, nil
} else if !errors.Is(err, ErrTemplateNotFound) {
return Template{}, err
}
}
return s.get(ctx, GlobalTemplateScope, key)
}
func (s *TemplateStore) get(ctx context.Context, tenantSlug, key string) (Template, error) {
var t Template
err := s.pool.QueryRow(ctx, `
SELECT subject, body FROM notification_templates WHERE tenant_slug = $1 AND key = $2
`, tenantSlug, key).Scan(&t.Subject, &t.Body)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return Template{}, ErrTemplateNotFound
}
return Template{}, fmt.Errorf("vorlage lesen: %w", err)
}
return t, nil
}
// Render fuellt eine Vorlage mit data. Fehlt ein referenzierter Platzhalter
// in data, bricht das Rendering kontrolliert MIT FEHLER ab, statt eine
// fehlerhafte/unvollstaendige Nachricht zu erzeugen (Akzeptanzkriterium 1 /
// Pruefung 1) — text/template mit Option("missingkey=error") liefert dafuer
// einen Fehler statt stillschweigend "<no value>" einzusetzen.
func Render(tmpl Template, data map[string]any) (subject, body string, err error) {
subject, err = renderOne("subject", tmpl.Subject, data)
if err != nil {
return "", "", err
}
body, err = renderOne("body", tmpl.Body, data)
if err != nil {
return "", "", err
}
return subject, body, nil
}
func renderOne(name, text string, data map[string]any) (string, error) {
tmpl, err := template.New(name).Option("missingkey=error").Parse(text)
if err != nil {
return "", fmt.Errorf("vorlage %q parsen: %w", name, err)
}
var buf bytes.Buffer
if err := tmpl.Execute(&buf, data); err != nil {
return "", fmt.Errorf("vorlage %q rendern (fehlender platzhalter?): %w", name, err)
}
return buf.String(), nil
}
+87
View File
@@ -0,0 +1,87 @@
// 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
}
+43
View File
@@ -0,0 +1,43 @@
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")
}
}
}
+87
View File
@@ -0,0 +1,87 @@
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()
}
+179
View File
@@ -0,0 +1,179 @@
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")
}
}()
}
+106
View File
@@ -0,0 +1,106 @@
// Package license implementiert Core LIC-01: Lizenzmodell je Tenant (Plan,
// Modul-Umfang, Laufzeit) und die kryptographische Pruefung signierter
// Lizenzschluessel. Feature-Flag-AUSWERTUNG zur Laufzeit (LIC-02) und die
// Verwaltungsoberflaeche (LIC-04) sind ausdruecklich nicht Teil dieses Pakets
// — hier geht es nur um Ausstellung/Validierung/Persistenz (Unleash-Vorbild:
// klare Trennung Flag-Verwaltung vs. Flag-Auswertung).
package license
import (
"crypto/ed25519"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"time"
)
var (
ErrInvalidSignature = errors.New("license: signatur ungueltig")
ErrMalformedKey = errors.New("license: lizenzschluessel hat ungueltiges format")
)
// Payload ist der signierte Lizenzinhalt (Akzeptanzkriterium 3: Plan,
// Modul-Liste, Laufzeit).
type Payload struct {
TenantSlug string `json:"tenant_slug"`
Plan string `json:"plan"`
Modules []string `json:"modules"`
IssuedAt time.Time `json:"issued_at"`
ValidUntil time.Time `json:"valid_until"`
}
// Issuer stellt signierte Lizenzschluessel aus. Haelt den PRIVATEN
// Ed25519-Schluessel — lebt in der Praxis beim Lizenzgeber, nicht im
// laufenden Core-Prozess (der nur den Validator mit dem oeffentlichen
// Schluessel braucht).
type Issuer struct {
priv ed25519.PrivateKey
}
func NewIssuer(priv ed25519.PrivateKey) *Issuer {
return &Issuer{priv: priv}
}
// Issue liefert den Lizenzschluessel im Format base64(payload-json) "." base64(signatur).
func (i *Issuer) Issue(payload Payload) (string, error) {
raw, err := json.Marshal(payload)
if err != nil {
return "", fmt.Errorf("payload serialisieren: %w", err)
}
sig := ed25519.Sign(i.priv, raw)
return base64.RawURLEncoding.EncodeToString(raw) + "." + base64.RawURLEncoding.EncodeToString(sig), nil
}
// Validator prueft Lizenzschluessel gegen den OEFFENTLICHEN Ed25519-Schluessel
// — das ist alles, was der laufende Core-Prozess kennen muss.
type Validator struct {
pub ed25519.PublicKey
}
func NewValidator(pub ed25519.PublicKey) *Validator {
return &Validator{pub: pub}
}
// Parse prueft die Signatur (Akzeptanzkriterium 1 / Pruefung 1) und liefert
// bei Erfolg den entschluesselten Payload. Ein manipulierter Schluessel wird
// hier zuverlaessig erkannt, unabhaengig davon, ob die Laufzeit noch gueltig
// waere — Signaturpruefung und Ablaufpruefung sind bewusst getrennt
// (Signatur bei Einspielen, Ablauf bei jeder Nutzung, siehe Store.RequireActive).
func (v *Validator) Parse(key string) (Payload, error) {
rawPart, sigPart, ok := splitOnce(key, '.')
if !ok {
return Payload{}, ErrMalformedKey
}
raw, err := base64.RawURLEncoding.DecodeString(rawPart)
if err != nil {
return Payload{}, ErrMalformedKey
}
sig, err := base64.RawURLEncoding.DecodeString(sigPart)
if err != nil {
return Payload{}, ErrMalformedKey
}
if !ed25519.Verify(v.pub, raw, sig) {
return Payload{}, ErrInvalidSignature
}
var p Payload
if err := json.Unmarshal(raw, &p); err != nil {
// Signatur war gueltig, aber Payload nicht mehr parsebar — sollte bei
// unveraenderten Schluesseln nie vorkommen, trotzdem kein Panic.
return Payload{}, fmt.Errorf("%w: payload nicht lesbar", ErrMalformedKey)
}
return p, nil
}
func splitOnce(s string, sep byte) (before, after string, ok bool) {
for i := 0; i < len(s); i++ {
if s[i] == sep {
return s[:i], s[i+1:], true
}
}
return "", "", false
}
+106
View File
@@ -0,0 +1,106 @@
package license
import (
"crypto/ed25519"
"errors"
"testing"
"time"
)
func testKeyPair(t *testing.T) (ed25519.PublicKey, ed25519.PrivateKey) {
t.Helper()
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
t.Fatalf("schluesselpaar erzeugen: %v", err)
}
return pub, priv
}
func TestIssueAndParse_RoundTrip(t *testing.T) {
pub, priv := testKeyPair(t)
issuer := NewIssuer(priv)
validator := NewValidator(pub)
payload := Payload{
TenantSlug: "acme",
Plan: "pro",
Modules: []string{"dms", "mail"},
IssuedAt: time.Now().Truncate(time.Second),
ValidUntil: time.Now().Add(365 * 24 * time.Hour).Truncate(time.Second),
}
key, err := issuer.Issue(payload)
if err != nil {
t.Fatalf("issue: %v", err)
}
got, err := validator.Parse(key)
if err != nil {
t.Fatalf("parse: %v", err)
}
if got.TenantSlug != payload.TenantSlug || got.Plan != payload.Plan || len(got.Modules) != 2 {
t.Fatalf("payload nach parse unerwartet: %+v", got)
}
}
// Akzeptanzkriterium 1 + Pruefung 1: manipulierter Schluessel wird zuverlaessig erkannt.
func TestParse_RejectsTamperedKey(t *testing.T) {
pub, priv := testKeyPair(t)
issuer := NewIssuer(priv)
validator := NewValidator(pub)
key, err := issuer.Issue(Payload{TenantSlug: "acme", Plan: "pro", ValidUntil: time.Now().Add(time.Hour)})
if err != nil {
t.Fatalf("issue: %v", err)
}
// Ein Zeichen im signierten Teil aendern.
tampered := []byte(key)
changed := false
for i, c := range tampered {
if c != '.' {
if c == 'A' {
tampered[i] = 'B'
} else {
tampered[i] = 'A'
}
changed = true
break
}
}
if !changed {
t.Fatal("testaufbau fehlerhaft: nichts zum manipulieren gefunden")
}
if _, err := validator.Parse(string(tampered)); !errors.Is(err, ErrInvalidSignature) {
t.Fatalf("erwartet ErrInvalidSignature, habe %v", err)
}
}
func TestParse_RejectsWrongKeyPair(t *testing.T) {
_, priv := testKeyPair(t)
otherPub, _ := testKeyPair(t)
issuer := NewIssuer(priv)
validator := NewValidator(otherPub) // falscher oeffentlicher Schluessel
key, err := issuer.Issue(Payload{TenantSlug: "acme", Plan: "pro"})
if err != nil {
t.Fatalf("issue: %v", err)
}
if _, err := validator.Parse(key); !errors.Is(err, ErrInvalidSignature) {
t.Fatalf("erwartet ErrInvalidSignature, habe %v", err)
}
}
func TestParse_RejectsMalformedKey(t *testing.T) {
pub, _ := testKeyPair(t)
validator := NewValidator(pub)
cases := []string{"", "keine-punkt-trennung", "!!!.!!!"}
for _, c := range cases {
if _, err := validator.Parse(c); !errors.Is(err, ErrMalformedKey) {
t.Fatalf("Parse(%q): erwartet ErrMalformedKey, habe %v", c, err)
}
}
}
+88
View File
@@ -0,0 +1,88 @@
package license
import (
"context"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var (
ErrNoLicense = errors.New("license: kein lizenzdatensatz fuer diesen tenant")
ErrLicenseExpired = errors.New("license: lizenz abgelaufen")
)
// Store persistiert den Lizenzumfang je Tenant in der Control-Plane-Registry
// (siehe internal/tenant.Registry — dieselbe Datenbank, aber ein eigener,
// unabhaengiger Store, um internal/tenant nicht um lizenzfremde Belange zu
// erweitern).
type Store struct {
pool *pgxpool.Pool
validator *Validator
}
func NewStore(pool *pgxpool.Pool, validator *Validator) *Store {
return &Store{pool: pool, validator: validator}
}
// Install prueft die Signatur des Lizenzschluessels (Akzeptanzkriterium 1)
// und ersetzt den bisherigen Lizenzdatensatz des Tenants vollstaendig. Ein
// bereits abgelaufener, aber korrekt signierter Schluessel wird trotzdem
// gespeichert — der Ablauf wird erst bei der Nutzung (RequireActive)
// bewertet, nicht beim Einspielen.
func (s *Store) Install(ctx context.Context, tenantID, licenseKey string) (Payload, error) {
payload, err := s.validator.Parse(licenseKey)
if err != nil {
return Payload{}, err
}
_, err = s.pool.Exec(ctx, `
INSERT INTO tenant_licenses (tenant_id, plan, modules, issued_at, valid_until, raw_key, installed_at)
VALUES ($1, $2, $3, $4, $5, $6, now())
ON CONFLICT (tenant_id) DO UPDATE SET
plan = $2, modules = $3, issued_at = $4, valid_until = $5, raw_key = $6, installed_at = now()
`, tenantID, payload.Plan, payload.Modules, payload.IssuedAt, payload.ValidUntil, licenseKey)
if err != nil {
return Payload{}, fmt.Errorf("lizenz speichern: %w", err)
}
return payload, nil
}
func (s *Store) get(ctx context.Context, tenantID string) (Payload, error) {
var p Payload
row := s.pool.QueryRow(ctx, `
SELECT plan, modules, issued_at, valid_until
FROM tenant_licenses WHERE tenant_id = $1
`, tenantID)
if err := row.Scan(&p.Plan, &p.Modules, &p.IssuedAt, &p.ValidUntil); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return Payload{}, ErrNoLicense
}
return Payload{}, fmt.Errorf("lizenz lesen: %w", err)
}
return p, nil
}
// Status liefert den persistierten Lizenzumfang unabhaengig vom Ablauf
// (Akzeptanzkriterium 3: Plan, Modul-Liste, Laufzeit abfragbar).
func (s *Store) Status(ctx context.Context, tenantID string) (Payload, error) {
return s.get(ctx, tenantID)
}
// RequireActive liefert den Lizenzumfang NUR, wenn die Lizenz noch nicht
// abgelaufen ist — sonst ErrLicenseExpired statt eines harten Fehlers/Panics
// (Akzeptanzkriterium 2: definierter eingeschraenkter Zustand). Aufrufende
// Module (LIC-02/03) entscheiden, was "eingeschraenkt" konkret bedeutet.
func (s *Store) RequireActive(ctx context.Context, tenantID string) (Payload, error) {
p, err := s.get(ctx, tenantID)
if err != nil {
return Payload{}, err
}
if time.Now().After(p.ValidUntil) {
return Payload{}, ErrLicenseExpired
}
return p, nil
}
+177
View File
@@ -0,0 +1,177 @@
package license
import (
"context"
"crypto/ed25519"
"errors"
"os"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupStoreTest(t *testing.T) (*Store, *Issuer, 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 tenants (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
slug TEXT NOT NULL UNIQUE,
name TEXT NOT NULL,
db_name TEXT NOT NULL UNIQUE,
db_dsn TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'active',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS tenant_licenses (
tenant_id UUID PRIMARY KEY REFERENCES tenants(id),
plan TEXT NOT NULL,
modules TEXT[] NOT NULL,
issued_at TIMESTAMPTZ NOT NULL,
valid_until TIMESTAMPTZ NOT NULL,
raw_key TEXT NOT NULL,
installed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
`); err != nil {
t.Fatalf("schema: %v", err)
}
var tenantID string
if err := pool.QueryRow(ctx, `
INSERT INTO tenants (slug, name, db_name, db_dsn)
VALUES ('lic_test_tenant', 'Lic Test', 'tenant_lic_test', 'unused')
RETURNING id
`).Scan(&tenantID); err != nil {
t.Fatalf("test-tenant anlegen: %v", err)
}
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
t.Fatalf("schluesselpaar: %v", err)
}
issuer := NewIssuer(priv)
store := NewStore(pool, NewValidator(pub))
cleanup := func() {
_, _ = pool.Exec(ctx, `DELETE FROM tenant_licenses WHERE tenant_id = $1`, tenantID)
_, _ = pool.Exec(ctx, `DELETE FROM tenants WHERE id = $1`, tenantID)
pool.Close()
}
return store, issuer, tenantID, cleanup
}
// Akzeptanzkriterium 3: Lizenzumfang persistiert und abfragbar.
func TestStore_InstallAndStatus(t *testing.T) {
store, issuer, tenantID, cleanup := setupStoreTest(t)
defer cleanup()
ctx := context.Background()
payload := Payload{
TenantSlug: "lic_test_tenant",
Plan: "enterprise",
Modules: []string{"dms", "mail", "archive"},
IssuedAt: time.Now().Truncate(time.Second),
ValidUntil: time.Now().Add(30 * 24 * time.Hour).Truncate(time.Second),
}
key, err := issuer.Issue(payload)
if err != nil {
t.Fatalf("issue: %v", err)
}
if _, err := store.Install(ctx, tenantID, key); err != nil {
t.Fatalf("install: %v", err)
}
status, err := store.Status(ctx, tenantID)
if err != nil {
t.Fatalf("status: %v", err)
}
if status.Plan != "enterprise" || len(status.Modules) != 3 {
t.Fatalf("status unerwartet: %+v", status)
}
}
func TestStore_InstallRejectsInvalidSignature(t *testing.T) {
store, _, tenantID, cleanup := setupStoreTest(t)
defer cleanup()
ctx := context.Background()
_, otherPriv, _ := ed25519.GenerateKey(nil)
foreignIssuer := NewIssuer(otherPriv) // signiert mit falschem schluessel
key, err := foreignIssuer.Issue(Payload{TenantSlug: "lic_test_tenant", Plan: "pro", ValidUntil: time.Now().Add(time.Hour)})
if err != nil {
t.Fatalf("issue: %v", err)
}
if _, err := store.Install(ctx, tenantID, key); !errors.Is(err, ErrInvalidSignature) {
t.Fatalf("erwartet ErrInvalidSignature, habe %v", err)
}
}
// Akzeptanzkriterium 2 + Pruefung 2: abgelaufene Lizenz fuehrt zu definiertem
// eingeschraenktem Zustand (ErrLicenseExpired), nicht zu einem Absturz.
func TestStore_RequireActive_DetectsExpiry(t *testing.T) {
store, issuer, tenantID, cleanup := setupStoreTest(t)
defer cleanup()
ctx := context.Background()
expired := Payload{
TenantSlug: "lic_test_tenant",
Plan: "pro",
Modules: []string{"dms"},
IssuedAt: time.Now().Add(-48 * time.Hour),
ValidUntil: time.Now().Add(-24 * time.Hour), // bereits abgelaufen
}
key, err := issuer.Issue(expired)
if err != nil {
t.Fatalf("issue: %v", err)
}
// Einspielen einer bereits abgelaufenen, aber korrekt signierten Lizenz
// muss funktionieren (Ablauf wird erst bei Nutzung bewertet).
if _, err := store.Install(ctx, tenantID, key); err != nil {
t.Fatalf("install sollte trotz ablauf funktionieren: %v", err)
}
func() {
defer func() {
if r := recover(); r != nil {
t.Fatalf("RequireActive hat gepanict statt einen fehler zu liefern: %v", r)
}
}()
if _, err := store.RequireActive(ctx, tenantID); !errors.Is(err, ErrLicenseExpired) {
t.Fatalf("erwartet ErrLicenseExpired, habe %v", err)
}
}()
// Aber der Umfang bleibt weiterhin abfragbar (Status, im Unterschied zu RequireActive).
status, err := store.Status(ctx, tenantID)
if err != nil {
t.Fatalf("status sollte trotz ablauf funktionieren: %v", err)
}
if status.Plan != "pro" {
t.Fatalf("status unerwartet: %+v", status)
}
}
func TestStore_RequireActive_NoLicense(t *testing.T) {
store, _, tenantID, cleanup := setupStoreTest(t)
defer cleanup()
ctx := context.Background()
if _, err := store.RequireActive(ctx, tenantID); !errors.Is(err, ErrNoLicense) {
t.Fatalf("erwartet ErrNoLicense, habe %v", err)
}
}
+99
View File
@@ -0,0 +1,99 @@
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
}
+46
View File
@@ -0,0 +1,46 @@
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)
}
}
+120
View File
@@ -0,0 +1,120 @@
// 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
}
+304
View File
@@ -0,0 +1,304 @@
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)
}
}
+97
View File
@@ -0,0 +1,97 @@
package moduletrust
import (
"context"
"fmt"
"sync"
"time"
)
// StaleCache ist der generische Rechte-/Feature-Flag-Cache-Kontrakt
// (Akzeptanzkriterium 2): TTL-basiert, mit explizitem, benanntem Verhalten
// bei abgelaufenem Cache waehrend Core nicht erreichbar ist.
//
// - Get: FAIL-OPEN fuer Lesevorgaenge. Schlaegt der Refresh fehl, aber es
// gibt bereits einen (wenn auch abgelaufenen) Stand, wird dieser mit
// stale=true zurueckgegeben — Begruendung: ein bereits authentifiziertes
// Modul soll mit dem letztbekannten Stand weiterarbeiten koennen statt
// hart zu blockieren (siehe "Bekannte Fehler vermeiden" im Ticket).
// Existiert noch nie ein Stand, gibt es keinen sinnvollen Fallback —
// dann liefert auch Get einen Fehler.
// - RequireFresh: FAIL-CLOSED fuer sicherheitskritische Aktionen (z.B.
// ein komplett NEUER Login). Nutzt NIEMALS einen zwischengespeicherten
// Stand, ruft immer frisch ab — Begruendung: eine neue Vertrauens-
// entscheidung darf nicht auf veralteten Daten beruhen, auch wenn das
// bedeutet, dass die Aktion bei Core-Ausfall sichtbar fehlschlaegt statt
// unsicher "irgendwie" durchgelassen zu werden.
//
// LIC-02 (internal/flag.Service) implementiert bereits denselben Kontrakt
// fuer Feature-Flags — StaleCache verallgemeinert dasselbe Muster fuer
// JWT-Signaturschluessel, damit beide Faelle derselben dokumentierten
// Policy folgen.
type StaleCache[T any] struct {
mu sync.RWMutex
value T
hasValue bool
fetchedAt time.Time
ttl time.Duration
fetch func(ctx context.Context) (T, error)
}
func NewStaleCache[T any](ttl time.Duration, fetch func(ctx context.Context) (T, error)) *StaleCache[T] {
return &StaleCache[T]{ttl: ttl, fetch: fetch}
}
// Get liefert den Cache-Wert. FAIL-OPEN: bei Refresh-Fehler wird ein
// vorhandener, ggf. abgelaufener Stand zurueckgegeben (stale=true).
func (c *StaleCache[T]) Get(ctx context.Context) (value T, stale bool, err error) {
c.mu.RLock()
fresh := c.hasValue && time.Since(c.fetchedAt) < c.ttl
if fresh {
v := c.value
c.mu.RUnlock()
return v, false, nil
}
c.mu.RUnlock()
newVal, fetchErr := c.fetch(ctx)
if fetchErr == nil {
c.mu.Lock()
c.value, c.hasValue, c.fetchedAt = newVal, true, time.Now()
c.mu.Unlock()
return newVal, false, nil
}
c.mu.RLock()
defer c.mu.RUnlock()
if c.hasValue {
return c.value, true, nil
}
var zero T
return zero, false, fmt.Errorf("cache leer und refresh fehlgeschlagen: %w", fetchErr)
}
// Invalidate erzwingt beim naechsten Get-Aufruf einen sofortigen Refresh
// statt auf den TTL-Ablauf zu warten (API-06, Akzeptanzkriterium 1: ein
// Health-Check-getriggerter Wiederanlauf soll den Cache SOFORT aktualisieren,
// nicht die reguläre TTL abwarten) — rein additiv, aendert nichts an
// Get/RequireFresh (dasselbe Muster wie internal/flag.Service.Invalidate).
func (c *StaleCache[T]) Invalidate() {
c.mu.Lock()
c.hasValue = false
c.mu.Unlock()
}
// RequireFresh ruft IMMER frisch ab (FAIL-CLOSED) — fuer sicherheitskritische
// Aktionen, die niemals auf einem zwischengespeicherten Stand basieren duerfen.
func (c *StaleCache[T]) RequireFresh(ctx context.Context) (T, error) {
v, err := c.fetch(ctx)
if err != nil {
var zero T
return zero, fmt.Errorf("core nicht erreichbar, sicherheitskritische aktion abgelehnt: %w", err)
}
c.mu.Lock()
c.value, c.hasValue, c.fetchedAt = v, true, time.Now()
c.mu.Unlock()
return v, nil
}
+79
View File
@@ -0,0 +1,79 @@
package moduletrust
import (
"encoding/base64"
"encoding/json"
"fmt"
"net/http"
"time"
"github.com/golang-jwt/jwt/v5"
)
type Claims struct {
Subject string `json:"sub"`
TenantSlug string `json:"tenant"`
jwt.RegisteredClaims
}
// Issue signiert ein Token mit dem aktuellen Signierschluessel und traegt
// dessen KID im JWT-Header ein — der Verifier auf Modulseite waehlt darueber
// den passenden oeffentlichen Schluessel aus PublicKeySet() aus.
func (m *KeyManager) Issue(subject, tenantSlug string, ttl time.Duration) (string, error) {
key, err := m.SigningKey()
if err != nil {
return "", err
}
now := time.Now()
claims := Claims{
Subject: subject,
TenantSlug: tenantSlug,
RegisteredClaims: jwt.RegisteredClaims{
IssuedAt: jwt.NewNumericDate(now),
ExpiresAt: jwt.NewNumericDate(now.Add(ttl)),
},
}
token := jwt.NewWithClaims(jwt.SigningMethodEdDSA, claims)
token.Header["kid"] = key.KID
return token.SignedString(key.Private)
}
type jwksResponse struct {
Keys []jwksKey `json:"keys"`
}
type jwksKey struct {
Kid string `json:"kid"`
PublicKey string `json:"public_key"` // base64 (raw Ed25519, 32 Byte)
}
// ServeJWKS liefert alle bekannten oeffentlichen Schluessel als JSON —
// Module fragen dies periodisch ab (nicht pro Request), siehe Verifier.
func (m *KeyManager) ServeJWKS(w http.ResponseWriter, r *http.Request) {
set := m.PublicKeySet()
resp := jwksResponse{Keys: make([]jwksKey, 0, len(set))}
for kid, pub := range set {
resp.Keys = append(resp.Keys, jwksKey{Kid: kid, PublicKey: base64.StdEncoding.EncodeToString(pub)})
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(resp)
}
// ParseJWKS dekodiert die JSON-Antwort von ServeJWKS zurueck in kid->PublicKey
// — Hilfsfunktion fuer Module, die JWKS per HTTP abrufen.
func ParseJWKS(data []byte) (map[string][]byte, error) {
var resp jwksResponse
if err := json.Unmarshal(data, &resp); err != nil {
return nil, fmt.Errorf("jwks parsen: %w", err)
}
out := make(map[string][]byte, len(resp.Keys))
for _, k := range resp.Keys {
raw, err := base64.StdEncoding.DecodeString(k.PublicKey)
if err != nil {
return nil, fmt.Errorf("oeffentlichen schluessel %q dekodieren: %w", k.Kid, err)
}
out[k.Kid] = raw
}
return out, nil
}
+83
View File
@@ -0,0 +1,83 @@
// Package moduletrust implementiert Core API-05: asymmetrische JWT-Signatur
// mit Schluesselverteilung (JWKS), damit DMS/Mail/Archive/Workflow JWTs
// LOKAL verifizieren koennen, ohne pro Aufruf einen synchronen Request an
// Core zu stellen — Core darf Fundament sein, ohne zum Flaschenhals zu
// werden (siehe Entscheidungsverlauf "Vertrauensstellung Core<->Module" in
// nexarch-state.json). IAM-02s HS256-Session-Cookie (Browser-Login) bleibt
// unangetastet — dies ist ein zusaetzlicher, getrennter Vertrauensmechanismus
// fuer Modul-zu-Modul/Modul-zu-Core-Aufrufe.
package moduletrust
import (
"crypto/ed25519"
"crypto/rand"
"encoding/hex"
"fmt"
"sync"
)
type KeyPair struct {
KID string
Private ed25519.PrivateKey
Public ed25519.PublicKey
}
// KeyManager haelt ALLE noch gueltigen Schluesselpaare — nicht nur das
// aktuell signierende. Rotate erzeugt ein neues Paar und behaelt die alten
// fuer die Verifikation bereits ausgestellter Tokens (Akzeptanzkriterium 3:
// Rotation ohne Ausfallzeit fuer andere Module).
type KeyManager struct {
mu sync.RWMutex
keys []KeyPair // aeltestes zuerst, neuestes zuletzt
}
func NewKeyManager() (*KeyManager, error) {
m := &KeyManager{}
if _, err := m.Rotate(); err != nil {
return nil, err
}
return m, nil
}
// Rotate erzeugt ein neues Ed25519-Schluesselpaar mit eigener KID und macht
// es zum aktuellen Signierschluessel. Aeltere Schluessel bleiben in
// PublicKeySet() erhalten, damit bereits ausgestellte Tokens weiterhin
// verifizierbar sind.
func (m *KeyManager) Rotate() (KeyPair, error) {
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
return KeyPair{}, fmt.Errorf("schluesselpaar erzeugen: %w", err)
}
kidBytes := make([]byte, 8)
if _, err := rand.Read(kidBytes); err != nil {
return KeyPair{}, fmt.Errorf("kid erzeugen: %w", err)
}
kp := KeyPair{KID: hex.EncodeToString(kidBytes), Private: priv, Public: pub}
m.mu.Lock()
m.keys = append(m.keys, kp)
m.mu.Unlock()
return kp, nil
}
// SigningKey liefert den aktuellen (neuesten) Schluessel zum Signieren neuer Tokens.
func (m *KeyManager) SigningKey() (KeyPair, error) {
m.mu.RLock()
defer m.mu.RUnlock()
if len(m.keys) == 0 {
return KeyPair{}, fmt.Errorf("moduletrust: kein schluessel vorhanden")
}
return m.keys[len(m.keys)-1], nil
}
// PublicKeySet liefert ALLE bekannten oeffentlichen Schluessel (kid ->
// public key) — die Grundlage fuer den JWKS-Endpunkt.
func (m *KeyManager) PublicKeySet() map[string]ed25519.PublicKey {
m.mu.RLock()
defer m.mu.RUnlock()
out := make(map[string]ed25519.PublicKey, len(m.keys))
for _, k := range m.keys {
out[k.KID] = k.Public
}
return out
}
+214
View File
@@ -0,0 +1,214 @@
package moduletrust
import (
"context"
"crypto/ed25519"
"errors"
"net/http"
"sync"
"testing"
"time"
)
func TestIssueAndVerify_RoundTrip(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
token, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return km.PublicKeySet(), nil
})
claims, err := v.Verify(context.Background(), token)
if err != nil {
t.Fatalf("verify: %v", err)
}
if claims.Subject != "user-1" || claims.TenantSlug != "acme" {
t.Fatalf("claims unerwartet: %+v", claims)
}
}
// Akzeptanzkriterium 1: Verifikation lokal, kein Request pro Aufruf.
func TestVerify_DoesNotFetchPerCall(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
token, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
var mu sync.Mutex
fetchCalls := 0
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
mu.Lock()
fetchCalls++
mu.Unlock()
return km.PublicKeySet(), nil
})
for i := 0; i < 10; i++ {
if _, err := v.Verify(context.Background(), token); err != nil {
t.Fatalf("verify %d: %v", i, err)
}
}
mu.Lock()
defer mu.Unlock()
if fetchCalls != 1 {
t.Fatalf("erwartet genau 1 fetch fuer 10 Verify-Aufrufe innerhalb der TTL, habe %d", fetchCalls)
}
}
// Akzeptanzkriterium 2 + Pruefung 1: Core simuliert nicht erreichbar,
// bereits authentifizierte Nutzer bleiben funktionsfaehig (Fail-Open mit
// letztbekanntem Schluesselstand).
func TestVerify_FailsOpenWhenCoreUnreachableButStaleKeysExist(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
token, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
coreDown := false
v := NewVerifier(30*time.Millisecond, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
if coreDown {
return nil, errors.New("core nicht erreichbar (simuliert)")
}
return km.PublicKeySet(), nil
})
// Cache vorwaermen, waehrend Core noch erreichbar ist.
if _, err := v.Verify(context.Background(), token); err != nil {
t.Fatalf("verify (warm): %v", err)
}
// "Core abschalten" und TTL ablaufen lassen.
coreDown = true
time.Sleep(50 * time.Millisecond)
if _, err := v.Verify(context.Background(), token); err != nil {
t.Fatalf("verify sollte trotz core-ausfall mit letztbekanntem stand funktionieren: %v", err)
}
}
// Akzeptanzkriterium 2 + Pruefung 2: neue sicherheitskritische Aktionen
// (z.B. neuer Login) schlagen bei Core-Ausfall klar fehl statt unsicher
// durchgelassen zu werden — auch wenn ein (aelterer) Cache-Stand existiert.
func TestRequireFreshKeys_FailsClosedWhenCoreUnreachable(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
coreDown := false
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
if coreDown {
return nil, errors.New("core nicht erreichbar (simuliert)")
}
return km.PublicKeySet(), nil
})
// Cache vorwaermen (existiert jetzt ein "veralteter" gueltiger Stand).
if _, _, err := v.cache.Get(context.Background()); err != nil {
t.Fatalf("warm cache: %v", err)
}
coreDown = true
if err := v.RequireFreshKeys(context.Background()); err == nil {
t.Fatal("erwartet fehler (fail-closed) bei core-ausfall, habe nil")
}
}
// Akzeptanzkriterium 3 + Pruefung 3: Schluesselrotation ohne Ausfallzeit —
// ein bereits ausgestelltes Token bleibt nach Rotation weiterhin
// verifizierbar, ein zweites (simuliertes) Modul bekommt beide Schluessel.
func TestRotate_NoDowntimeForAlreadyIssuedTokens(t *testing.T) {
km, err := NewKeyManager()
if err != nil {
t.Fatalf("new key manager: %v", err)
}
oldToken, err := km.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue (alt): %v", err)
}
if _, err := km.Rotate(); err != nil {
t.Fatalf("rotate: %v", err)
}
newToken, err := km.Issue("user-2", "acme", time.Hour)
if err != nil {
t.Fatalf("issue (neu): %v", err)
}
// Simuliertes zweites Modul: fragt den vollstaendigen Schluesselsatz ab.
moduleB := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return km.PublicKeySet(), nil
})
if _, err := moduleB.Verify(context.Background(), oldToken); err != nil {
t.Fatalf("altes token sollte nach rotation weiterhin gueltig sein: %v", err)
}
if _, err := moduleB.Verify(context.Background(), newToken); err != nil {
t.Fatalf("neues token sollte gueltig sein: %v", err)
}
}
func TestVerify_RejectsUnknownKid(t *testing.T) {
km1, _ := NewKeyManager()
km2, _ := NewKeyManager() // komplett anderer, unbekannter schluessel
token, err := km1.Issue("user-1", "acme", time.Hour)
if err != nil {
t.Fatalf("issue: %v", err)
}
v := NewVerifier(time.Hour, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return km2.PublicKeySet(), nil // kennt km1s schluessel nicht
})
if _, err := v.Verify(context.Background(), token); !errors.Is(err, ErrInvalidToken) {
t.Fatalf("erwartet ErrInvalidToken, habe %v", err)
}
}
func TestJWKSRoundTrip(t *testing.T) {
km, _ := NewKeyManager()
km.Rotate()
var buf []byte
rec := &captureWriter{}
km.ServeJWKS(rec, nil)
buf = rec.body
parsed, err := ParseJWKS(buf)
if err != nil {
t.Fatalf("parse jwks: %v", err)
}
if len(parsed) != len(km.PublicKeySet()) {
t.Fatalf("erwartet %d schluessel, habe %d", len(km.PublicKeySet()), len(parsed))
}
}
type captureWriter struct {
body []byte
header http.Header
}
func (w *captureWriter) Header() http.Header {
if w.header == nil {
w.header = http.Header{}
}
return w.header
}
func (w *captureWriter) Write(p []byte) (int, error) { w.body = append(w.body, p...); return len(p), nil }
func (w *captureWriter) WriteHeader(statusCode int) {}
+67
View File
@@ -0,0 +1,67 @@
package moduletrust
import (
"context"
"crypto/ed25519"
"errors"
"time"
"github.com/golang-jwt/jwt/v5"
)
var ErrInvalidToken = errors.New("moduletrust: ungueltiges token")
// KeyFetchFunc holt den aktuellen Schluesselsatz von Core (z.B. per HTTP-GET
// auf ServeJWKS + ParseJWKS). Wird vom Verifier nur bei abgelaufener TTL
// aufgerufen — NICHT bei jeder Verify()-Anfrage (Akzeptanzkriterium 1).
type KeyFetchFunc func(ctx context.Context) (map[string]ed25519.PublicKey, error)
// Verifier ist die Modulseite von API-05: verifiziert JWTs LOKAL gegen einen
// per StaleCache zwischengespeicherten Schluesselsatz, ohne pro Aufruf einen
// synchronen Request an Core zu stellen.
type Verifier struct {
cache *StaleCache[map[string]ed25519.PublicKey]
}
func NewVerifier(ttl time.Duration, fetch KeyFetchFunc) *Verifier {
return &Verifier{cache: NewStaleCache(ttl, func(ctx context.Context) (map[string]ed25519.PublicKey, error) {
return fetch(ctx)
})}
}
// Verify prueft die Signatur LOKAL gegen den (ggf. abgelaufenen, aber
// vorhandenen) Schluesselsatz — FAIL-OPEN fuer bereits ausgestellte Tokens
// (Akzeptanzkriterium 2): ist Core nicht erreichbar, aber ein alter
// Schluesselsatz bekannt, wird damit weiter verifiziert.
func (v *Verifier) Verify(ctx context.Context, tokenString string) (*Claims, error) {
keys, _, err := v.cache.Get(ctx)
if err != nil {
return nil, err
}
claims := &Claims{}
token, err := jwt.ParseWithClaims(tokenString, claims, func(t *jwt.Token) (interface{}, error) {
if _, ok := t.Method.(*jwt.SigningMethodEd25519); !ok {
return nil, ErrInvalidToken
}
kid, _ := t.Header["kid"].(string)
pub, ok := keys[kid]
if !ok {
return nil, ErrInvalidToken
}
return pub, nil
})
if err != nil || !token.Valid {
return nil, ErrInvalidToken
}
return claims, nil
}
// RequireFreshKeys ruft IMMER frisch von Core ab (FAIL-CLOSED) — fuer
// sicherheitskritische Aktionen wie einen komplett neuen Login
// (Akzeptanzkriterium 2): schlaegt klar fehl, wenn Core nicht erreichbar
// ist, statt auf einem veralteten Schluesselsatz zu vertrauen.
func (v *Verifier) RequireFreshKeys(ctx context.Context) error {
_, err := v.cache.RequireFresh(ctx)
return err
}
-81
View File
@@ -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
}
-219
View File
@@ -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)
}
}
-102
View File
@@ -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
}
-83
View File
@@ -1,83 +0,0 @@
package notifyprefs
import (
"encoding/json"
"net/http"
"gitea.perlbach24.de/scripte/nexarch/internal/auth"
)
// Handler stellt Benachrichtigungspraeferenzen als HTTP-Endpunkte bereit
// (CFG-04). Duenner Client von Store — keine eigene Zustelllogik.
type Handler struct {
prefs *Store
}
func NewHandler(prefs *Store) *Handler {
return &Handler{prefs: prefs}
}
type setPreferenceRequest struct {
EventType string `json:"event_type"`
Channel string `json:"channel"`
Enabled bool `json:"enabled"`
}
// SetPreference setzt die eigene Praeferenz des angemeldeten Benutzers
// (Akzeptanzkriterium 1). Muss hinter auth.RequireAuth haengen.
func (h *Handler) SetPreference(w http.ResponseWriter, r *http.Request) {
claims, ok := auth.ClaimsFromContext(r.Context())
if !ok {
http.Error(w, "nicht angemeldet", http.StatusUnauthorized)
return
}
var req setPreferenceRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "ungueltige Anfrage", http.StatusBadRequest)
return
}
if err := h.prefs.Set(r.Context(), claims.TenantSlug, claims.UserID, req.EventType, req.Channel, req.Enabled); err != nil {
http.Error(w, "Einstellung konnte nicht gespeichert werden.", http.StatusBadRequest)
return
}
w.WriteHeader(http.StatusOK)
}
// ListMine liefert die eigenen Praeferenzen (Akzeptanzkriterium 1) — nur die
// EIGENEN, niemals die eines anderen Benutzers (Pruefung 2). claims.UserID
// kommt ausschliesslich aus dem verifizierten Session-Token, ist also nicht
// vom Client manipulierbar.
func (h *Handler) ListMine(w http.ResponseWriter, r *http.Request) {
claims, ok := auth.ClaimsFromContext(r.Context())
if !ok {
http.Error(w, "nicht angemeldet", http.StatusUnauthorized)
return
}
prefs, err := h.prefs.ListForUser(r.Context(), claims.TenantSlug, claims.UserID)
if err != nil {
http.Error(w, "Einstellungen konnten nicht geladen werden.", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(prefs)
}
// ListTenantOverview liefert die Praeferenzen ALLER Benutzer des eigenen
// Mandanten (Akzeptanzkriterium 3: Tenant-Admin-Übersicht). Die Zugriffs-
// beschraenkung "nur Tenant-Admin" ist Aufgabe von RBAC-02 (Policy-
// Enforcement-Schicht) vor diesem Handler — hier wird bewusst keine eigene
// Rechtelogik dupliziert (Ticket-Produkt-DNA: duenner Client).
func (h *Handler) ListTenantOverview(w http.ResponseWriter, r *http.Request) {
claims, ok := auth.ClaimsFromContext(r.Context())
if !ok {
http.Error(w, "nicht angemeldet", http.StatusUnauthorized)
return
}
prefs, err := h.prefs.ListForTenant(r.Context(), claims.TenantSlug)
if err != nil {
http.Error(w, "Übersicht konnte nicht geladen werden.", http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(prefs)
}
-122
View File
@@ -1,122 +0,0 @@
package notifyprefs
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"gitea.perlbach24.de/scripte/nexarch/internal/auth"
)
func sessionCookie(t *testing.T, issuer *auth.TokenIssuer, userID, tenantSlug string) *http.Cookie {
t.Helper()
token, err := issuer.Issue(userID, tenantSlug)
if err != nil {
t.Fatalf("session-token ausstellen: %v", err)
}
return &http.Cookie{Name: auth.CookieName, Value: token}
}
// Akzeptanzkriterium 1: Benutzer setzt eigene Praeferenz per HTTP.
func TestHandlerSetAndListMine(t *testing.T) {
prefs, _, _, cleanup := setupTest(t)
defer cleanup()
h := NewHandler(prefs)
issuer := auth.NewTokenIssuer("test-secret")
cookie := sessionCookie(t, issuer, "user-1", "acme")
setBody, _ := json.Marshal(setPreferenceRequest{EventType: "welcome", Channel: "email", Enabled: false})
setReq := httptest.NewRequest(http.MethodPost, "/notifications/preferences", bytes.NewReader(setBody))
setReq.AddCookie(cookie)
setRec := httptest.NewRecorder()
auth.RequireAuth(issuer, h.SetPreference)(setRec, setReq)
if setRec.Code != http.StatusOK {
t.Fatalf("set: status = %d, body: %s", setRec.Code, setRec.Body.String())
}
listReq := httptest.NewRequest(http.MethodGet, "/notifications/preferences", nil)
listReq.AddCookie(cookie)
listRec := httptest.NewRecorder()
auth.RequireAuth(issuer, h.ListMine)(listRec, listReq)
if listRec.Code != http.StatusOK {
t.Fatalf("list: status = %d, body: %s", listRec.Code, listRec.Body.String())
}
var out []Preference
if err := json.Unmarshal(listRec.Body.Bytes(), &out); err != nil {
t.Fatalf("response dekodieren: %v", err)
}
if len(out) != 1 || out[0].Enabled {
t.Fatalf("erwartet genau 1 deaktivierte praeferenz, habe %+v", out)
}
}
// Akzeptanzkriterium 1 / Pruefung 2: ListMine zeigt AUSSCHLIESSLICH die
// eigenen Praeferenzen — die UserID kommt aus dem Session-Token, nicht aus
// einem vom Client kontrollierbaren Parameter.
func TestHandlerListMine_NeverShowsOtherUsersPreferences(t *testing.T) {
prefs, _, _, cleanup := setupTest(t)
defer cleanup()
h := NewHandler(prefs)
issuer := auth.NewTokenIssuer("test-secret")
setBody, _ := json.Marshal(setPreferenceRequest{EventType: "welcome", Channel: "email", Enabled: false})
setReq := httptest.NewRequest(http.MethodPost, "/notifications/preferences", bytes.NewReader(setBody))
setReq.AddCookie(sessionCookie(t, issuer, "user-1", "acme"))
setRec := httptest.NewRecorder()
auth.RequireAuth(issuer, h.SetPreference)(setRec, setReq)
if setRec.Code != http.StatusOK {
t.Fatalf("set fuer user-1: status = %d", setRec.Code)
}
// user-2 fragt seine eigenen Praeferenzen ab — muss leer sein.
listReq := httptest.NewRequest(http.MethodGet, "/notifications/preferences", nil)
listReq.AddCookie(sessionCookie(t, issuer, "user-2", "acme"))
listRec := httptest.NewRecorder()
auth.RequireAuth(issuer, h.ListMine)(listRec, listReq)
if listRec.Code != http.StatusOK {
t.Fatalf("list fuer user-2: status = %d", listRec.Code)
}
var out []Preference
if err := json.Unmarshal(listRec.Body.Bytes(), &out); err != nil {
t.Fatalf("response dekodieren: %v", err)
}
if len(out) != 0 {
t.Fatalf("user-2 haette user-1s praeferenzen NICHT sehen duerfen, habe %+v", out)
}
}
// Akzeptanzkriterium 3: Tenant-Übersicht liefert alle Benutzer des Mandanten.
func TestHandlerListTenantOverview(t *testing.T) {
prefs, _, _, cleanup := setupTest(t)
defer cleanup()
h := NewHandler(prefs)
issuer := auth.NewTokenIssuer("test-secret")
for _, u := range []string{"user-1", "user-2"} {
body, _ := json.Marshal(setPreferenceRequest{EventType: "welcome", Channel: "email", Enabled: false})
req := httptest.NewRequest(http.MethodPost, "/notifications/preferences", bytes.NewReader(body))
req.AddCookie(sessionCookie(t, issuer, u, "acme"))
rec := httptest.NewRecorder()
auth.RequireAuth(issuer, h.SetPreference)(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("set fuer %s: status = %d", u, rec.Code)
}
}
req := httptest.NewRequest(http.MethodGet, "/notifications/preferences/tenant", nil)
req.AddCookie(sessionCookie(t, issuer, "admin-1", "acme"))
rec := httptest.NewRecorder()
auth.RequireAuth(issuer, h.ListTenantOverview)(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("overview: status = %d, body: %s", rec.Code, rec.Body.String())
}
var out []Preference
if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil {
t.Fatalf("response dekodieren: %v", err)
}
if len(out) != 2 {
t.Fatalf("erwartet 2 eintraege (beide Benutzer), habe %d", len(out))
}
}
-144
View File
@@ -1,144 +0,0 @@
// Package notifyprefs implementiert Core CFG-04: Benachrichtigungspräferenzen
// je Benutzer, Ereignistyp und Kanal. Duenner Client von CFG-02/03 — keine
// eigene Zustelllogik, nur ein Filter DAVOR, ob eine Benachrichtigung
// überhaupt in die Warteschlange (internal/notify.Dispatcher) eingereiht wird.
package notifyprefs
import (
"context"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
)
type Preference struct {
TenantSlug string
UserID string
EventType string
Channel string
Enabled bool
}
// Store verwaltet Benachrichtigungspraeferenzen in der zentralen Registry-DB
// (gleiche Ablageebene wie CFG-03s notification_templates/in_app_notifications).
type Store struct {
pool *pgxpool.Pool
}
func NewStore(pool *pgxpool.Pool) *Store {
return &Store{pool: pool}
}
// Set setzt die Praeferenz eines Benutzers fuer einen Ereignistyp+Kanal
// (Akzeptanzkriterium 1). Wirkt sofort — es gibt keinen Cache dazwischen,
// jede Pruefung (IsEnabled) liest direkt aus der DB (Akzeptanzkriterium 2).
func (s *Store) Set(ctx context.Context, tenantSlug, userID, eventType, channel string, enabled bool) error {
if tenantSlug == "" || userID == "" || eventType == "" || channel == "" {
return errors.New("notifyprefs: tenantSlug, userID, eventType und channel duerfen nicht leer sein")
}
_, err := s.pool.Exec(ctx, `
INSERT INTO notification_preferences (tenant_slug, user_id, event_type, channel, enabled)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (tenant_slug, user_id, event_type, channel)
DO UPDATE SET enabled = $5, updated_at = now()
`, tenantSlug, userID, eventType, channel, enabled)
if err != nil {
return fmt.Errorf("praeferenz speichern: %w", err)
}
return nil
}
// IsEnabled prueft, ob ein Kanal fuer einen Ereignistyp aktiv ist. Ohne
// explizite Praeferenz gilt der Kanal als AKTIVIERT (Opt-out-Default,
// siehe Migration).
func (s *Store) IsEnabled(ctx context.Context, tenantSlug, userID, eventType, channel string) (bool, error) {
var enabled bool
err := s.pool.QueryRow(ctx, `
SELECT enabled FROM notification_preferences
WHERE tenant_slug = $1 AND user_id = $2 AND event_type = $3 AND channel = $4
`, tenantSlug, userID, eventType, channel).Scan(&enabled)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return true, nil
}
return false, fmt.Errorf("praeferenz lesen: %w", err)
}
return enabled, nil
}
// ListForUser liefert alle expliziten Praeferenzen eines Benutzers
// (Akzeptanzkriterium 1: eigene Einstellungen einsehen/aendern).
func (s *Store) ListForUser(ctx context.Context, tenantSlug, userID string) ([]Preference, error) {
rows, err := s.pool.Query(ctx, `
SELECT event_type, channel, enabled FROM notification_preferences
WHERE tenant_slug = $1 AND user_id = $2
ORDER BY event_type, channel
`, tenantSlug, userID)
if err != nil {
return nil, fmt.Errorf("praeferenzen auflisten: %w", err)
}
defer rows.Close()
var out []Preference
for rows.Next() {
p := Preference{TenantSlug: tenantSlug, UserID: userID}
if err := rows.Scan(&p.EventType, &p.Channel, &p.Enabled); err != nil {
return nil, fmt.Errorf("praeferenz lesen: %w", err)
}
out = append(out, p)
}
return out, rows.Err()
}
// ListForTenant liefert ALLE expliziten Praeferenzen aller Benutzer eines
// Mandanten (Akzeptanzkriterium 3: Tenant-Admin-Übersicht).
func (s *Store) ListForTenant(ctx context.Context, tenantSlug string) ([]Preference, error) {
rows, err := s.pool.Query(ctx, `
SELECT user_id, event_type, channel, enabled FROM notification_preferences
WHERE tenant_slug = $1
ORDER BY user_id, event_type, channel
`, tenantSlug)
if err != nil {
return nil, fmt.Errorf("mandanten-praeferenzen auflisten: %w", err)
}
defer rows.Close()
var out []Preference
for rows.Next() {
p := Preference{TenantSlug: tenantSlug}
if err := rows.Scan(&p.UserID, &p.EventType, &p.Channel, &p.Enabled); err != nil {
return nil, fmt.Errorf("praeferenz lesen: %w", err)
}
out = append(out, p)
}
return out, rows.Err()
}
// EnqueueIfAllowed ist der einzige vorgesehene Weg, wie ein Modul eine
// Benachrichtigung fuer einen konkreten Benutzer+Ereignistyp auslöst: prueft
// zuerst die Praeferenz, reiht NUR bei Aktivierung tatsaechlich in
// internal/notify.Dispatcher ein (Akzeptanzkriterium 1 / Pruefung 1 — ein
// deaktivierter Kanal erzeugt nachweislich KEINE notification_jobs-Zeile,
// nicht nur eine ignorierte). skipped=true bedeutet: bewusst nicht zugestellt,
// kein Fehler.
func EnqueueIfAllowed(
ctx context.Context,
prefs *Store,
dispatcher *notify.Dispatcher,
tenantSlug, userID, eventType, channel, recipient string,
payload map[string]any,
) (jobID string, skipped bool, err error) {
enabled, err := prefs.IsEnabled(ctx, tenantSlug, userID, eventType, channel)
if err != nil {
return "", false, err
}
if !enabled {
return "", true, nil
}
jobID, err = dispatcher.Enqueue(ctx, channel, recipient, payload)
return jobID, false, err
}
-183
View File
@@ -1,183 +0,0 @@
package notifyprefs
import (
"context"
"os"
"testing"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/notify"
)
func setupTest(t *testing.T) (*Store, *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 notification_preferences (
tenant_slug TEXT NOT NULL, user_id TEXT NOT NULL, event_type TEXT NOT NULL, channel TEXT NOT NULL,
enabled BOOLEAN NOT NULL, updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_slug, user_id, event_type, channel)
);
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 notification_preferences`)
_, _ = pool.Exec(ctx, `DELETE FROM notification_jobs`)
pool.Close()
}
return NewStore(pool), notify.NewDispatcher(pool), pool, cleanup
}
// Ohne explizite Praeferenz gilt ein Kanal als aktiviert (Opt-out-Default).
func TestIsEnabled_DefaultsToTrueWithoutPreference(t *testing.T) {
prefs, _, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
enabled, err := prefs.IsEnabled(ctx, "acme", "user-1", "welcome", "email")
if err != nil {
t.Fatalf("is enabled: %v", err)
}
if !enabled {
t.Fatal("erwartet aktiviert als default ohne gesetzte praeferenz")
}
}
// Akzeptanzkriterium 1 + 2: gesetzte Praeferenz wirkt sofort.
func TestSetAndIsEnabled(t *testing.T) {
prefs, _, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
if err := prefs.Set(ctx, "acme", "user-1", "welcome", "email", false); err != nil {
t.Fatalf("set: %v", err)
}
enabled, err := prefs.IsEnabled(ctx, "acme", "user-1", "welcome", "email")
if err != nil {
t.Fatalf("is enabled: %v", err)
}
if enabled {
t.Fatal("erwartet deaktiviert nach Set(..., false)")
}
// Andere Kanaele/Ereignistypen desselben Benutzers bleiben unberuehrt.
stillOn, err := prefs.IsEnabled(ctx, "acme", "user-1", "welcome", "in_app")
if err != nil {
t.Fatalf("is enabled (anderer kanal): %v", err)
}
if !stillOn {
t.Fatal("erwartet: anderer kanal bleibt aktiviert")
}
}
// Akzeptanzkriterium 1 / Pruefung 2: Einstellungen eines Benutzers wirken
// sich NICHT auf einen anderen Benutzer aus.
func TestSet_DoesNotAffectOtherUser(t *testing.T) {
prefs, _, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
if err := prefs.Set(ctx, "acme", "user-1", "welcome", "email", false); err != nil {
t.Fatalf("set fuer user-1: %v", err)
}
enabledForOther, err := prefs.IsEnabled(ctx, "acme", "user-2", "welcome", "email")
if err != nil {
t.Fatalf("is enabled fuer user-2: %v", err)
}
if !enabledForOther {
t.Fatal("praeferenz von user-1 haette user-2 nicht beeinflussen duerfen")
}
}
// Akzeptanzkriterium 1 / Pruefung 1 (Kernbeweis): ein deaktivierter Kanal
// erhaelt NACHWEISLICH keine weitere Zustellung — geprueft ueber die
// tatsaechliche notification_jobs-Tabelle, nicht nur ueber den Rueckgabewert.
func TestEnqueueIfAllowed_SkipsWhenDisabled(t *testing.T) {
prefs, dispatcher, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
if err := prefs.Set(ctx, "acme", "user-1", "invoice_ready", "email", false); err != nil {
t.Fatalf("set: %v", err)
}
jobID, skipped, err := EnqueueIfAllowed(ctx, prefs, dispatcher, "acme", "user-1", "invoice_ready", "email", "user1@acme.example", nil)
if err != nil {
t.Fatalf("enqueue if allowed: %v", err)
}
if !skipped {
t.Fatal("erwartet skipped=true bei deaktiviertem kanal")
}
if jobID != "" {
t.Fatalf("erwartet keine job-id, habe %q", jobID)
}
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 != 0 {
t.Fatalf("erwartet 0 eingereihte jobs bei deaktiviertem kanal, habe %d", count)
}
}
func TestEnqueueIfAllowed_EnqueuesWhenEnabled(t *testing.T) {
prefs, dispatcher, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
jobID, skipped, err := EnqueueIfAllowed(ctx, prefs, dispatcher, "acme", "user-1", "invoice_ready", "email", "user1@acme.example", nil)
if err != nil {
t.Fatalf("enqueue if allowed: %v", err)
}
if skipped {
t.Fatal("erwartet skipped=false ohne deaktivierte praeferenz")
}
if jobID == "" {
t.Fatal("erwartet gesetzte job-id")
}
}
// Akzeptanzkriterium 3: Tenant-Admin-Übersicht zeigt alle Benutzer.
func TestListForTenant_ShowsAllUsers(t *testing.T) {
prefs, _, _, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
if err := prefs.Set(ctx, "acme", "user-1", "welcome", "email", false); err != nil {
t.Fatalf("set user-1: %v", err)
}
if err := prefs.Set(ctx, "acme", "user-2", "welcome", "in_app", false); err != nil {
t.Fatalf("set user-2: %v", err)
}
if err := prefs.Set(ctx, "other-tenant", "user-3", "welcome", "email", false); err != nil {
t.Fatalf("set user-3 (anderer tenant): %v", err)
}
overview, err := prefs.ListForTenant(ctx, "acme")
if err != nil {
t.Fatalf("list for tenant: %v", err)
}
if len(overview) != 2 {
t.Fatalf("erwartet 2 eintraege fuer tenant acme (nicht den anderen tenant), habe %d", len(overview))
}
}
+155
View File
@@ -0,0 +1,155 @@
// Package resync implementiert Core API-06: Wiederanlauf & Nachsynchro-
// nisierung nach einem Core-Ausfall.
//
// - Ein Modul puffert Audit-Events und Nutzungszaehler-Deltas LOKAL in
// Postgres (NICHT im Speicher — siehe "Bekannte Fehler vermeiden" im
// Ticket: ein erneuter Ausfall waehrend der Nachlieferung darf keine
// Daten verlieren, eine In-Memory-Queue wuerde das riskieren).
// - Ein Health-Check-getriggerter Worker erkennt die Core-Wiedererreich-
// barkeit SOFORT (nicht erst nach TTL-Ablauf, siehe StaleCache.Invalidate)
// und liefert die gepufferten Daten in ORIGINALER Reihenfolge, authenti-
// fiziert ueber das Service-Credential aus API-02
// (internal/moduleregistry.Registry.Authenticate).
//
// Dieses Paket dupliziert weder internal/audit (AUD-01) noch internal/usage
// (LIC-03) — es liefert nur den Puffer- und Nachlieferungs-Mechanismus
// DAVOR bzw. DANACH.
package resync
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
// BufferedAuditEvent ist ein lokal gepuffertes Audit-Ereignis. Seq
// garantiert die Wiederherstellung der urspruenglichen Reihenfolge
// (Akzeptanzkriterium 2) unabhaengig von eventuellen Uhrzeit-Ungenauigkeiten.
type BufferedAuditEvent struct {
ID int64
Seq int64
TenantSlug string
Actor string
Action string
Target string
Metadata map[string]any
CreatedAt time.Time
}
// BufferedUsageDelta ist ein lokal gepuffertes Nutzungszaehler-Inkrement.
type BufferedUsageDelta struct {
ID int64
Seq int64
TenantSlug string
Metric string
Delta int64
CreatedAt time.Time
}
// Buffer ist die lokale, persistente Pufferqueue eines Moduls.
type Buffer struct {
pool *pgxpool.Pool
}
func NewBuffer(pool *pgxpool.Pool) *Buffer {
return &Buffer{pool: pool}
}
// EnqueueAuditEvent puffert EIN Audit-Ereignis lokal — wird von einem
// Fachmodul aufgerufen, wenn Core gerade nicht erreichbar ist (die
// Erkennung "Core erreichbar oder nicht" ist NICHT Teil dieses Aufrufs,
// siehe Worker).
func (b *Buffer) EnqueueAuditEvent(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any) error {
if metadata == nil {
metadata = map[string]any{}
}
metadataJSON, err := json.Marshal(metadata)
if err != nil {
return fmt.Errorf("metadata serialisieren: %w", err)
}
_, err = b.pool.Exec(ctx, `
INSERT INTO resync_audit_buffer (tenant_slug, actor, action, target, metadata)
VALUES ($1, $2, $3, $4, $5)
`, tenantSlug, actor, action, target, metadataJSON)
if err != nil {
return fmt.Errorf("audit-ereignis puffern: %w", err)
}
return nil
}
// EnqueueUsageDelta puffert EIN Nutzungszaehler-Inkrement lokal.
func (b *Buffer) EnqueueUsageDelta(ctx context.Context, tenantSlug, metric string, delta int64) error {
_, err := b.pool.Exec(ctx, `
INSERT INTO resync_usage_buffer (tenant_slug, metric, delta)
VALUES ($1, $2, $3)
`, tenantSlug, metric, delta)
if err != nil {
return fmt.Errorf("nutzungsdelta puffern: %w", err)
}
return nil
}
// PendingAuditEvents liefert ALLE noch nicht zugestellten Audit-Events in
// ORIGINALER Reihenfolge (Akzeptanzkriterium 2 / Pruefung 1).
func (b *Buffer) PendingAuditEvents(ctx context.Context) ([]BufferedAuditEvent, error) {
rows, err := b.pool.Query(ctx, `
SELECT id, seq, tenant_slug, actor, action, target, metadata, created_at
FROM resync_audit_buffer ORDER BY seq
`)
if err != nil {
return nil, fmt.Errorf("gepufferte audit-events abfragen: %w", err)
}
defer rows.Close()
var out []BufferedAuditEvent
for rows.Next() {
var e BufferedAuditEvent
var metadataJSON []byte
if err := rows.Scan(&e.ID, &e.Seq, &e.TenantSlug, &e.Actor, &e.Action, &e.Target, &metadataJSON, &e.CreatedAt); err != nil {
return nil, fmt.Errorf("gepuffertes audit-event lesen: %w", err)
}
_ = json.Unmarshal(metadataJSON, &e.Metadata)
out = append(out, e)
}
return out, rows.Err()
}
// PendingUsageDeltas liefert ALLE noch nicht zugestellten Nutzungsdeltas.
func (b *Buffer) PendingUsageDeltas(ctx context.Context) ([]BufferedUsageDelta, error) {
rows, err := b.pool.Query(ctx, `
SELECT id, seq, tenant_slug, metric, delta, created_at
FROM resync_usage_buffer ORDER BY seq
`)
if err != nil {
return nil, fmt.Errorf("gepufferte nutzungsdeltas abfragen: %w", err)
}
defer rows.Close()
var out []BufferedUsageDelta
for rows.Next() {
var d BufferedUsageDelta
if err := rows.Scan(&d.ID, &d.Seq, &d.TenantSlug, &d.Metric, &d.Delta, &d.CreatedAt); err != nil {
return nil, fmt.Errorf("gepuffertes nutzungsdelta lesen: %w", err)
}
out = append(out, d)
}
return out, rows.Err()
}
// RemoveAuditEvent entfernt EIN Audit-Event aus dem Puffer — wird NUR nach
// von Core BESTAETIGTER Zustellung aufgerufen (Akzeptanzkriterium 2/3: erst
// entfernen, wenn sicher zugestellt, sonst bleibt es fuer den naechsten
// Versuch erhalten — kein Datenverlust bei erneutem Ausfall waehrend der
// Nachlieferung).
func (b *Buffer) RemoveAuditEvent(ctx context.Context, id int64) error {
_, err := b.pool.Exec(ctx, `DELETE FROM resync_audit_buffer WHERE id = $1`, id)
return err
}
func (b *Buffer) RemoveUsageDelta(ctx context.Context, id int64) error {
_, err := b.pool.Exec(ctx, `DELETE FROM resync_usage_buffer WHERE id = $1`, id)
return err
}
+132
View File
@@ -0,0 +1,132 @@
package resync
import (
"context"
"encoding/json"
"net/http"
"time"
)
// AuditRecorder ist die schmale Schnittstelle, ueber die Core empfangene
// Audit-Events tatsaechlich persistiert. In Produktion durch internal/audit
// (AUD-01, nicht Abhaengigkeit dieser Kachel) implementiert — dieses Paket
// dupliziert dessen Validierungs-/Speicherlogik NICHT, sondern ruft sie nur
// auf. Die Events werden vom Aufrufer sequenziell in PendingAuditEvents-
// Reihenfolge uebergeben (Akzeptanzkriterium 2).
type AuditRecorder interface {
Record(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any, occurredAt time.Time) error
}
// UsageIncrementer ist die schmale Schnittstelle zu Core's Nutzungszaehler
// (in Produktion internal/usage, LIC-03 — nicht Abhaengigkeit dieser
// Kachel). Jedes gepufferte Delta wird GENAU EINMAL angewendet.
type UsageIncrementer interface {
Increment(ctx context.Context, tenantSlug, metric string, delta int64) error
}
// CredentialAuthenticator ist die schmale Schnittstelle zu API-02s
// Service-Credential-Pruefung (internal/moduleregistry.Registry.Authenticate).
type CredentialAuthenticator interface {
Authenticate(ctx context.Context, clientID, secret string) (moduleName string, ok bool, err error)
}
// Handler nimmt nachgelieferte Audit-Events/Nutzungsdeltas auf der
// Core-Seite entgegen — authentifiziert ueber dasselbe Service-Credential
// wie jeder andere Modul-Core-Aufruf (Akzeptanzkriterium 2/3, "authentifiziert
// ueber das in API-02 definierte Service-Credential").
type Handler struct {
auth CredentialAuthenticator
audit AuditRecorder
usage UsageIncrementer
}
func NewHandler(auth CredentialAuthenticator, audit AuditRecorder, usage UsageIncrementer) *Handler {
return &Handler{auth: auth, audit: audit, usage: usage}
}
type credentialHeader struct {
ClientID string `json:"client_id"`
Secret string `json:"secret"`
}
func (h *Handler) authenticate(w http.ResponseWriter, r *http.Request) bool {
clientID := r.Header.Get("X-Nexarch-Client-Id")
secret := r.Header.Get("X-Nexarch-Client-Secret")
_, ok, err := h.auth.Authenticate(r.Context(), clientID, secret)
if err != nil || !ok {
http.Error(w, "ungueltiges service-credential", http.StatusUnauthorized)
return false
}
return true
}
type auditEventDTO struct {
TenantSlug string `json:"tenant_slug"`
Actor string `json:"actor"`
Action string `json:"action"`
Target string `json:"target"`
Metadata map[string]any `json:"metadata"`
OccurredAt time.Time `json:"occurred_at"`
}
// AuditHandler nimmt EINE Liste gepufferter Audit-Events entgegen und
// schreibt sie SEQUENZIELL in der gegebenen Reihenfolge fort
// (Akzeptanzkriterium 2 / Pruefung 1: Vollstaendigkeit + Reihenfolge).
// Bricht die Verarbeitung bei einem Fehler ab und meldet, wie viele Events
// bereits sicher geschrieben wurden — der Aufrufer (Worker) entfernt aus
// seinem lokalen Puffer NUR die bestaetigt geschriebenen Events.
func (h *Handler) AuditHandler(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(w, r) {
return
}
var events []auditEventDTO
if err := json.NewDecoder(r.Body).Decode(&events); err != nil {
http.Error(w, "ungueltiger anfrage-koerper", http.StatusBadRequest)
return
}
written := 0
for _, e := range events {
if err := h.audit.Record(r.Context(), e.TenantSlug, e.Actor, e.Action, e.Target, e.Metadata, e.OccurredAt); err != nil {
break
}
written++
}
writeJSON(w, map[string]int{"written": written})
}
type usageDeltaDTO struct {
TenantSlug string `json:"tenant_slug"`
Metric string `json:"metric"`
Delta int64 `json:"delta"`
}
// UsageHandler wendet JEDES gepufferte Delta GENAU EINMAL an
// (Akzeptanzkriterium 3 / Pruefung 2: keine Doppelzaehlung) — der Aufrufer
// entfernt aus seinem lokalen Puffer nur die bestaetigt uebernommenen Deltas.
func (h *Handler) UsageHandler(w http.ResponseWriter, r *http.Request) {
if !h.authenticate(w, r) {
return
}
var deltas []usageDeltaDTO
if err := json.NewDecoder(r.Body).Decode(&deltas); err != nil {
http.Error(w, "ungueltiger anfrage-koerper", http.StatusBadRequest)
return
}
applied := 0
for _, d := range deltas {
if err := h.usage.Increment(r.Context(), d.TenantSlug, d.Metric, d.Delta); err != nil {
break
}
applied++
}
writeJSON(w, map[string]int{"applied": applied})
}
func writeJSON(w http.ResponseWriter, body any) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(body)
}
+320
View File
@@ -0,0 +1,320 @@
package resync
import (
"context"
"fmt"
"net/http"
"net/http/httptest"
"os"
"sync"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"gitea.perlbach24.de/scripte/nexarch/internal/flag"
"gitea.perlbach24.de/scripte/nexarch/internal/moduleregistry"
)
// fakeAudit steht fuer internal/audit.Log (AUD-01, nicht Abhaengigkeit
// dieser Kachel) — zeichnet Aufrufe in Empfangsreihenfolge auf, damit
// Vollstaendigkeit UND Reihenfolge geprueft werden koennen.
type fakeAudit struct {
mu sync.Mutex
events []auditEventDTO
failAt int // -1 = nie fehlschlagen; sonst: ab diesem Index (0-basiert) schlaegt Record fehl
}
func (f *fakeAudit) Record(ctx context.Context, tenantSlug, actor, action, target string, metadata map[string]any, occurredAt time.Time) error {
f.mu.Lock()
defer f.mu.Unlock()
if f.failAt >= 0 && len(f.events) == f.failAt {
return fmt.Errorf("simulierter core-ausfall waehrend der nachlieferung")
}
f.events = append(f.events, auditEventDTO{TenantSlug: tenantSlug, Actor: actor, Action: action, Target: target, Metadata: metadata, OccurredAt: occurredAt})
return nil
}
// fakeUsage steht fuer internal/usage.Store (LIC-03, nicht Abhaengigkeit
// dieser Kachel) — summiert Deltas wie der echte Store.
type fakeUsage struct {
mu sync.Mutex
totals map[string]int64
}
func newFakeUsage() *fakeUsage { return &fakeUsage{totals: map[string]int64{}} }
func (f *fakeUsage) Increment(ctx context.Context, tenantSlug, metric string, delta int64) error {
f.mu.Lock()
defer f.mu.Unlock()
f.totals[tenantSlug+"|"+metric] += delta
return nil
}
func setupTest(t *testing.T) (*Buffer, *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 resync_audit_buffer (
id BIGSERIAL PRIMARY KEY, seq BIGSERIAL, tenant_slug TEXT NOT NULL, actor TEXT NOT NULL,
action TEXT NOT NULL, target TEXT NOT NULL DEFAULT '', metadata JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE IF NOT EXISTS resync_usage_buffer (
id BIGSERIAL PRIMARY KEY, seq BIGSERIAL, tenant_slug TEXT NOT NULL, metric TEXT NOT NULL,
delta BIGINT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
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)
}
cleanup := func() { pool.Close() }
return NewBuffer(pool), pool, cleanup
}
func uniqueModuleName() string {
return fmt.Sprintf("resync-test-%d", time.Now().UnixNano())
}
// setupModuleCredential registriert ein echtes Modul + Service-Credential
// ueber internal/moduleregistry (API-02) — dieselbe Authentifizierung wird
// vom Handler tatsaechlich geprueft, kein Mock.
func setupModuleCredential(t *testing.T, pool *pgxpool.Pool) (registry *moduleregistry.Registry, clientID, secret string) {
t.Helper()
ctx := context.Background()
flagService := flag.NewService(flag.NewStore(pool), 10*time.Millisecond)
registry = moduleregistry.NewRegistry(pool, flagService)
name := uniqueModuleName()
if _, err := registry.Register(ctx, name, "1.0.0", nil); err != nil {
t.Fatalf("modul registrieren: %v", err)
}
clientID, secret, err := registry.Provision(ctx, name)
if err != nil {
t.Fatalf("credential provisionieren: %v", err)
}
return registry, clientID, secret
}
// Akzeptanzkriterium 2 + Pruefung 1: waehrend eines simulierten Ausfalls
// lokal gepufferte Audit-Events sind nach Wiederanlauf vollstaendig und in
// korrekter Reihenfolge in Core's Audit-Log vorhanden.
func TestFlushAll_DeliversBufferedAuditEventsCompleteAndInOrder(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
registry, clientID, secret := setupModuleCredential(t, pool)
tenantSlug := "acme"
// Ereignisse "waehrend core down" lokal puffern.
actions := []string{"login", "upload", "delete", "logout"}
for _, action := range actions {
if err := buffer.EnqueueAuditEvent(ctx, tenantSlug, "user-1", action, "res-1", nil); err != nil {
t.Fatalf("enqueue %s: %v", action, err)
}
}
audit := &fakeAudit{failAt: -1}
usage := newFakeUsage()
handler := NewHandler(registry, audit, usage)
mux := http.NewServeMux()
mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
if err := worker.FlushAll(ctx); err != nil {
t.Fatalf("flushall: %v", err)
}
audit.mu.Lock()
defer audit.mu.Unlock()
if len(audit.events) != len(actions) {
t.Fatalf("erwartet %d zugestellte events, habe %d", len(actions), len(audit.events))
}
for i, e := range audit.events {
if e.Action != actions[i] {
t.Fatalf("reihenfolge falsch: position %d = %q, want %q", i, e.Action, actions[i])
}
}
// Puffer muss nach bestaetigter Zustellung leer sein.
remaining, err := buffer.PendingAuditEvents(ctx)
if err != nil {
t.Fatalf("pending: %v", err)
}
if len(remaining) != 0 {
t.Fatalf("erwartet leeren puffer nach bestaetigter zustellung, habe %d verbleibende", len(remaining))
}
}
// Bekannter-Fehler-Praevention: bricht die Zustellung waehrend der
// Nachlieferung erneut ab (Core faellt wieder aus), bleiben die NICHT
// bestaetigten Events sicher im Puffer erhalten statt verloren zu gehen.
func TestFlushAll_KeepsUnconfirmedEventsInBufferOnPartialFailure(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
registry, clientID, secret := setupModuleCredential(t, pool)
for i := 0; i < 5; i++ {
if err := buffer.EnqueueAuditEvent(ctx, "acme", "user-1", fmt.Sprintf("action-%d", i), "", nil); err != nil {
t.Fatalf("enqueue %d: %v", i, err)
}
}
// Core-Fake schlaegt AB dem 3. Event fehl -> simuliert erneuten Ausfall
// mitten in der Nachlieferung.
audit := &fakeAudit{failAt: 3}
handler := NewHandler(registry, audit, newFakeUsage())
mux := http.NewServeMux()
mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
if err := worker.FlushAll(ctx); err == nil {
t.Fatal("erwartet fehler, da core nur teilweise bestaetigt hat")
}
remaining, err := buffer.PendingAuditEvents(ctx)
if err != nil {
t.Fatalf("pending: %v", err)
}
if len(remaining) != 2 {
t.Fatalf("erwartet 2 verbleibende (nicht bestaetigte) events im puffer, habe %d", len(remaining))
}
}
// Akzeptanzkriterium 3 + Pruefung 2: Nutzungszaehler-Differenz aus der
// Ausfallzeit wird korrekt nachgebucht, ein wiederholter (fehlerhafter)
// Flush-Versuch fuehrt NICHT zu Doppelzaehlung, weil bereits bestaetigte
// Deltas aus dem Puffer entfernt sind.
func TestFlushAll_AppliesUsageDeltasWithoutDoubleCounting(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
registry, clientID, secret := setupModuleCredential(t, pool)
deltas := []int64{3, 5, 2}
var want int64
for _, d := range deltas {
want += d
if err := buffer.EnqueueUsageDelta(ctx, "acme", "api_calls", d); err != nil {
t.Fatalf("enqueue delta %d: %v", d, err)
}
}
usage := newFakeUsage()
handler := NewHandler(registry, &fakeAudit{failAt: -1}, usage)
mux := http.NewServeMux()
mux.HandleFunc("/internal/resync/audit", handler.AuditHandler)
mux.HandleFunc("/internal/resync/usage", handler.UsageHandler)
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
if err := worker.FlushAll(ctx); err != nil {
t.Fatalf("flushall 1: %v", err)
}
usage.mu.Lock()
got := usage.totals["acme|api_calls"]
usage.mu.Unlock()
if got != want {
t.Fatalf("nutzungsstand nach nachbuchung = %d, want %d", got, want)
}
// Ein zweiter Flush-Versuch (z.B. redundanter Retry) darf NICHTS mehr
// nachbuchen, da der Puffer bereits geleert wurde.
if err := worker.FlushAll(ctx); err != nil {
t.Fatalf("flushall 2: %v", err)
}
usage.mu.Lock()
got2 := usage.totals["acme|api_calls"]
usage.mu.Unlock()
if got2 != want {
t.Fatalf("nutzungsstand nach redundantem zweiten flush = %d, want unveraendert %d (keine doppelzaehlung)", got2, want)
}
}
// Akzeptanzkriterium 1 + Pruefung 3: bei erkannter Core-Wiedererreichbarkeit
// wird der Cache SOFORT invalidiert (naechster Zugriff refetcht), nicht
// erst nach TTL-Ablauf — Latenz wird gemessen und liegt weit unter einer
// langen TTL.
func TestCheckAndSync_InvalidatesCacheImmediatelyOnRecovery(t *testing.T) {
buffer, pool, cleanup := setupTest(t)
defer cleanup()
registry, clientID, secret := setupModuleCredential(t, pool)
_ = registry
mux := http.NewServeMux()
mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })
server := httptest.NewServer(mux)
defer server.Close()
client := NewCoreClient(server.URL, clientID, secret)
worker := NewWorker(buffer, client)
invalidated := false
var mu sync.Mutex
worker.OnReachable(func() {
mu.Lock()
invalidated = true
mu.Unlock()
})
start := time.Now()
becameReachable, err := worker.CheckAndSync(context.Background())
elapsed := time.Since(start)
if err != nil {
t.Fatalf("checkandsync: %v", err)
}
if !becameReachable {
t.Fatal("erwartet erkannten uebergang zu 'erreichbar' beim ersten erfolgreichen check")
}
mu.Lock()
defer mu.Unlock()
if !invalidated {
t.Fatal("erwartet sofortigen cache-invalidierungs-callback bei core-wiedererreichbarkeit")
}
// Zielwert: deutlich unter einer typischen TTL (z.B. 5s beim
// Feature-Flag-Cache, LIC-02) — hier im Millisekundenbereich, da rein
// lokal ohne Netzwerk-Overhead.
if elapsed > time.Second {
t.Fatalf("cache-invalidierung brauchte %s, erwartet deutlich unter 1s", elapsed)
}
}
+220
View File
@@ -0,0 +1,220 @@
package resync
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"time"
)
// CoreClient ist der modul-seitige HTTP-Client fuer die Nachlieferung,
// authentifiziert ueber dasselbe Service-Credential wie jeder andere
// Modul-Core-Aufruf (API-02).
type CoreClient struct {
BaseURL string
ClientID string
Secret string
HTTP *http.Client
}
func NewCoreClient(baseURL, clientID, secret string) *CoreClient {
return &CoreClient{BaseURL: baseURL, ClientID: clientID, Secret: secret, HTTP: &http.Client{Timeout: 5 * time.Second}}
}
func (c *CoreClient) post(ctx context.Context, path string, body any) (*http.Response, error) {
payload, err := json.Marshal(body)
if err != nil {
return nil, fmt.Errorf("payload serialisieren: %w", err)
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+path, bytes.NewReader(payload))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("X-Nexarch-Client-Id", c.ClientID)
req.Header.Set("X-Nexarch-Client-Secret", c.Secret)
return c.HTTP.Do(req)
}
// HealthCheck prueft, ob Core erreichbar ist — dieselbe Konvention wie
// internal/health (OPS-01): HTTP 200 auf einem Health-Endpunkt.
func (c *CoreClient) HealthCheck(ctx context.Context) bool {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.BaseURL+"/healthz", nil)
if err != nil {
return false
}
resp, err := c.HTTP.Do(req)
if err != nil {
return false
}
defer resp.Body.Close()
return resp.StatusCode == http.StatusOK
}
// Worker erkennt Core-Wiedererreichbarkeit und stoesst DANN sofort
// (Akzeptanzkriterium 1) sowohl registrierte Cache-Invalidierungen als auch
// das Nachliefern des lokalen Puffers an.
type Worker struct {
buffer *Buffer
client *CoreClient
onReachable []func()
wasDown bool
}
func NewWorker(buffer *Buffer, client *CoreClient) *Worker {
return &Worker{buffer: buffer, client: client, wasDown: true} // Start pessimistisch: erster erfolgreicher Check zaehlt als "Wiedererreichbarkeit".
}
// OnReachable registriert einen Callback, der bei jeder erkannten
// Core-Wiedererreichbarkeit sofort ausgefuehrt wird — z.B.
// moduletrust.StaleCache[T].Invalidate, damit der naechste Zugriff sofort
// neu abruft statt auf TTL-Ablauf zu warten (Akzeptanzkriterium 1).
func (w *Worker) OnReachable(fn func()) {
w.onReachable = append(w.onReachable, fn)
}
// CheckAndSync fuehrt EINEN Zyklus aus: Erreichbarkeit pruefen, bei
// erkanntem UEBERGANG "nicht erreichbar -> erreichbar" sofort die
// registrierten Callbacks ausloesen und den Puffer nachliefern. Gibt
// zurueck, ob ein Wiederanlauf in diesem Aufruf erkannt wurde (fuer
// Latenzmessung in Tests, Pruefung 3).
func (w *Worker) CheckAndSync(ctx context.Context) (becameReachable bool, err error) {
reachable := w.client.HealthCheck(ctx)
if !reachable {
w.wasDown = true
return false, nil
}
justRecovered := w.wasDown
w.wasDown = false
if !justRecovered {
return false, nil
}
for _, fn := range w.onReachable {
fn()
}
if err := w.FlushAll(ctx); err != nil {
return true, err
}
return true, nil
}
// FlushAll liefert ZUERST alle gepufferten Audit-Events (in Reihenfolge),
// DANN alle gepufferten Nutzungsdeltas nach. Jedes Element wird aus dem
// lokalen Puffer NUR entfernt, wenn Core es bestaetigt hat — bricht die
// Uebertragung vorzeitig ab (Core faellt waehrend der Nachlieferung erneut
// aus), bleibt der Rest sicher im Postgres-Puffer erhalten
// (Akzeptanzkriterium 2/3, "Bekannte Fehler vermeiden").
func (w *Worker) FlushAll(ctx context.Context) error {
if err := w.flushAuditEvents(ctx); err != nil {
return err
}
return w.flushUsageDeltas(ctx)
}
func (w *Worker) flushAuditEvents(ctx context.Context) error {
events, err := w.buffer.PendingAuditEvents(ctx)
if err != nil {
return err
}
if len(events) == 0 {
return nil
}
dtos := make([]auditEventDTO, len(events))
for i, e := range events {
dtos[i] = auditEventDTO{
TenantSlug: e.TenantSlug, Actor: e.Actor, Action: e.Action, Target: e.Target,
Metadata: e.Metadata, OccurredAt: e.CreatedAt,
}
}
resp, err := w.client.post(ctx, "/internal/resync/audit", dtos)
if err != nil {
return fmt.Errorf("audit-nachlieferung: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("audit-nachlieferung: unerwarteter status %d", resp.StatusCode)
}
var ack struct {
Written int `json:"written"`
}
if err := json.NewDecoder(resp.Body).Decode(&ack); err != nil {
return fmt.Errorf("audit-bestaetigung lesen: %w", err)
}
// NUR die von Core bestaetigt geschriebenen Events entfernen — sie sind
// nach PendingAuditEvents-Reihenfolge sortiert, die ersten `Written`
// Eintraege entsprechen also genau den bestaetigten.
for i := 0; i < ack.Written; i++ {
if err := w.buffer.RemoveAuditEvent(ctx, events[i].ID); err != nil {
return fmt.Errorf("bestaetigtes audit-event aus puffer entfernen: %w", err)
}
}
if ack.Written < len(events) {
return fmt.Errorf("core hat nur %d von %d audit-events bestaetigt", ack.Written, len(events))
}
return nil
}
func (w *Worker) flushUsageDeltas(ctx context.Context) error {
deltas, err := w.buffer.PendingUsageDeltas(ctx)
if err != nil {
return err
}
if len(deltas) == 0 {
return nil
}
dtos := make([]usageDeltaDTO, len(deltas))
for i, d := range deltas {
dtos[i] = usageDeltaDTO{TenantSlug: d.TenantSlug, Metric: d.Metric, Delta: d.Delta}
}
resp, err := w.client.post(ctx, "/internal/resync/usage", dtos)
if err != nil {
return fmt.Errorf("nutzungs-nachlieferung: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("nutzungs-nachlieferung: unerwarteter status %d", resp.StatusCode)
}
var ack struct {
Applied int `json:"applied"`
}
if err := json.NewDecoder(resp.Body).Decode(&ack); err != nil {
return fmt.Errorf("nutzungs-bestaetigung lesen: %w", err)
}
for i := 0; i < ack.Applied; i++ {
if err := w.buffer.RemoveUsageDelta(ctx, deltas[i].ID); err != nil {
return fmt.Errorf("bestaetigtes nutzungsdelta aus puffer entfernen: %w", err)
}
}
if ack.Applied < len(deltas) {
return fmt.Errorf("core hat nur %d von %d nutzungsdeltas bestaetigt", ack.Applied, len(deltas))
}
return nil
}
// Run fuehrt CheckAndSync in festen Abstaenden aus — die "kurze, definierte
// Zeitspanne" aus Akzeptanzkriterium 1 ist dieses Poll-Intervall.
func (w *Worker) Run(ctx context.Context, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
_, _ = w.CheckAndSync(ctx)
}
}
}
+32
View File
@@ -0,0 +1,32 @@
package usage
import (
"context"
"log/slog"
"time"
)
// AggregateFunc berechnet/aktualisiert Zaehlerstaende aus einer autoritativen
// Quelle (z.B. "zaehle Zeilen in einer Modul-Tabelle") — die konkrete Quelle
// haengt vom jeweiligen Modul ab und ist nicht Teil dieser Kachel. Das
// Aggregations-Grundgerüst selbst (periodischer Trigger) ist es.
type AggregateFunc func(ctx context.Context) error
// RunPeriodicAggregation ruft aggregate in festen Abstaenden auf, bis ctx
// beendet wird — dieselbe In-Prozess-Worker-Goroutine-Konvention wie
// internal/tenant.Lifecycle.RunSweeper (Akzeptanzkriterium 1: "periodisch
// aggregiert").
func RunPeriodicAggregation(ctx context.Context, interval time.Duration, aggregate AggregateFunc) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
if err := aggregate(ctx); err != nil {
slog.Error("nutzungszaehler-aggregation fehlgeschlagen", "error", err)
}
}
}
}
+144
View File
@@ -0,0 +1,144 @@
// Package usage implementiert Core LIC-03: Nutzungszaehler je Tenant
// (Benutzeranzahl, Speicherverbrauch, API-Aufrufe, ...) und die Pruefung
// gegen konfigurierte Quotas. Quotas sind Konfiguration (Tabellenzeile), kein
// Hardcode — Zitadel/Unleash-Vorbild.
package usage
import (
"context"
"errors"
"fmt"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
)
var ErrNoQuota = errors.New("usage: keine quota fuer diese metrik konfiguriert")
// Status ist die definierte Reaktion einer Quota-Pruefung (Akzeptanzkriterium 2).
type Status string
const (
StatusOK Status = "ok"
StatusWarning Status = "warning" // Schwelle (80%) erreicht, aber noch nicht ueberschritten
StatusExceeded Status = "exceeded" // Quota ueberschritten — neue Ressourcen sollten gesperrt werden
)
// warningThreshold liegt bei 80% der Quota.
const warningThreshold = 0.8
type Store struct {
pool *pgxpool.Pool
}
func NewStore(pool *pgxpool.Pool) *Store {
return &Store{pool: pool}
}
// Increment erhoeht einen Zaehler ATOMAR ueber ein einziges SQL-Statement
// (UPSERT mit value = value + delta) statt Read-Modify-Write in Go — das
// haelt Zaehlerstaende bei parallelen Schreibzugriffen konsistent
// (Akzeptanzkriterium 1 / Pruefung 2), ohne eine Anwendungs-Transaktion mit
// Lock zu brauchen.
func (s *Store) Increment(ctx context.Context, tenantID, metric string, delta int64) error {
_, err := s.pool.Exec(ctx, `
INSERT INTO usage_counters (tenant_id, metric, value, updated_at)
VALUES ($1, $2, $3, now())
ON CONFLICT (tenant_id, metric) DO UPDATE
SET value = usage_counters.value + $3, updated_at = now()
`, tenantID, metric, delta)
if err != nil {
return fmt.Errorf("zaehler erhoehen: %w", err)
}
return nil
}
// Get liefert den aktuellen Zaehlerstand — 0, wenn noch nie erhoeht wurde.
// Der Wert ist strikt tenant-gescoped (Akzeptanzkriterium 3 / Pruefung 3).
func (s *Store) Get(ctx context.Context, tenantID, metric string) (int64, error) {
var value int64
err := s.pool.QueryRow(ctx, `
SELECT value FROM usage_counters WHERE tenant_id = $1 AND metric = $2
`, tenantID, metric).Scan(&value)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return 0, nil
}
return 0, fmt.Errorf("zaehler lesen: %w", err)
}
return value, nil
}
// SetQuota legt die Obergrenze fuer (tenantID, metric) fest — Konfiguration,
// kein Hardcode.
func (s *Store) SetQuota(ctx context.Context, tenantID, metric string, limit int64) error {
_, err := s.pool.Exec(ctx, `
INSERT INTO usage_quotas (tenant_id, metric, limit_value)
VALUES ($1, $2, $3)
ON CONFLICT (tenant_id, metric) DO UPDATE SET limit_value = $3
`, tenantID, metric, limit)
if err != nil {
return fmt.Errorf("quota setzen: %w", err)
}
return nil
}
func (s *Store) GetQuota(ctx context.Context, tenantID, metric string) (int64, error) {
var limit int64
err := s.pool.QueryRow(ctx, `
SELECT limit_value FROM usage_quotas WHERE tenant_id = $1 AND metric = $2
`, tenantID, metric).Scan(&limit)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return 0, ErrNoQuota
}
return 0, fmt.Errorf("quota lesen: %w", err)
}
return limit, nil
}
// Check liefert Zaehlerstand, konfigurierte Quota und die daraus abgeleitete
// Reaktion (Akzeptanzkriterium 2 / Pruefung 1). Ist keine Quota konfiguriert,
// gilt die Metrik als unbegrenzt (StatusOK).
func (s *Store) Check(ctx context.Context, tenantID, metric string) (value, limit int64, status Status, err error) {
value, err = s.Get(ctx, tenantID, metric)
if err != nil {
return 0, 0, "", err
}
limit, err = s.GetQuota(ctx, tenantID, metric)
if errors.Is(err, ErrNoQuota) {
return value, 0, StatusOK, nil
}
if err != nil {
return 0, 0, "", err
}
switch {
case value > limit:
return value, limit, StatusExceeded, nil
case limit > 0 && float64(value) >= warningThreshold*float64(limit):
return value, limit, StatusWarning, nil
default:
return value, limit, StatusOK, nil
}
}
// Reaction wird aufgerufen, wenn Check einen Nicht-OK-Status liefert
// (Akzeptanzkriterium 2: "definierte Reaktion").
type Reaction func(ctx context.Context, tenantID, metric string, value, limit int64, status Status)
// Enforce fuehrt Check aus und ruft react auf, wenn der Status nicht OK ist —
// die konkrete "Sperre neuer Ressourcen"/Benachrichtigung liegt beim
// Aufrufer (z.B. TEN-02 vor dem Anlegen eines neuen Benutzers), Enforce
// garantiert nur, dass die Reaktion zuverlaessig ausgeloest wird.
func (s *Store) Enforce(ctx context.Context, tenantID, metric string, react Reaction) (Status, error) {
value, limit, status, err := s.Check(ctx, tenantID, metric)
if err != nil {
return "", err
}
if status != StatusOK && react != nil {
react(ctx, tenantID, metric, value, limit, status)
}
return status, nil
}
+206
View File
@@ -0,0 +1,206 @@
package usage
import (
"context"
"fmt"
"os"
"sync"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
)
func setupTest(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 usage_counters (
tenant_id UUID NOT NULL, metric TEXT NOT NULL, value BIGINT NOT NULL DEFAULT 0,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), PRIMARY KEY (tenant_id, metric)
);
CREATE TABLE IF NOT EXISTS usage_quotas (
tenant_id UUID NOT NULL, metric TEXT NOT NULL, limit_value BIGINT NOT NULL,
PRIMARY KEY (tenant_id, metric)
);
`); err != nil {
t.Fatalf("schema: %v", err)
}
cleanup := func() { pool.Close() }
return NewStore(pool), cleanup
}
func newTenantID() string {
return fmt.Sprintf("00000000-0000-0000-0000-%012d", time.Now().UnixNano()%1e12)
}
// Akzeptanzkriterium 1 + Pruefung 2: Aggregationsjob liefert bei parallelen
// Schreibzugriffen konsistente Zaehlerstaende.
func TestIncrement_ConsistentUnderConcurrentWrites(t *testing.T) {
store, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
tenant := newTenantID()
const goroutines = 50
var wg sync.WaitGroup
for i := 0; i < goroutines; i++ {
wg.Add(1)
go func() {
defer wg.Done()
if err := store.Increment(ctx, tenant, "api_calls", 1); err != nil {
t.Errorf("increment: %v", err)
}
}()
}
wg.Wait()
value, err := store.Get(ctx, tenant, "api_calls")
if err != nil {
t.Fatalf("get: %v", err)
}
if value != goroutines {
t.Fatalf("erwartet %d, habe %d (hinweis auf lost update unter nebenlaeufigkeit)", goroutines, value)
}
}
// Akzeptanzkriterium 3 + Pruefung 3: Zaehlerstand eines Tenants beeinflusst
// nicht den eines anderen.
func TestIncrement_IsolatedBetweenTenants(t *testing.T) {
store, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
tenantA, tenantB := newTenantID(), newTenantID()
if err := store.Increment(ctx, tenantA, "users", 5); err != nil {
t.Fatalf("increment a: %v", err)
}
if err := store.Increment(ctx, tenantB, "users", 1); err != nil {
t.Fatalf("increment b: %v", err)
}
valA, err := store.Get(ctx, tenantA, "users")
if err != nil {
t.Fatalf("get a: %v", err)
}
valB, err := store.Get(ctx, tenantB, "users")
if err != nil {
t.Fatalf("get b: %v", err)
}
if valA != 5 || valB != 1 {
t.Fatalf("erwartet a=5 b=1, habe a=%d b=%d", valA, valB)
}
}
// Akzeptanzkriterium 2 + Pruefung 1: Quota-Ueberschreitung wird automatisiert
// erkannt und die definierte Reaktion ausgeloest.
func TestEnforce_TriggersReactionOnExceeded(t *testing.T) {
store, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
tenant := newTenantID()
if err := store.SetQuota(ctx, tenant, "users", 10); err != nil {
t.Fatalf("set quota: %v", err)
}
if err := store.Increment(ctx, tenant, "users", 11); err != nil {
t.Fatalf("increment: %v", err)
}
var reacted bool
var gotStatus Status
status, err := store.Enforce(ctx, tenant, "users", func(ctx context.Context, tenantID, metric string, value, limit int64, status Status) {
reacted = true
gotStatus = status
})
if err != nil {
t.Fatalf("enforce: %v", err)
}
if status != StatusExceeded {
t.Fatalf("erwartet StatusExceeded, habe %q", status)
}
if !reacted || gotStatus != StatusExceeded {
t.Fatal("erwartet ausgeloeste reaktion mit StatusExceeded")
}
}
func TestCheck_WarningThresholdAndOK(t *testing.T) {
store, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
tenant := newTenantID()
if err := store.SetQuota(ctx, tenant, "storage_mb", 100); err != nil {
t.Fatalf("set quota: %v", err)
}
if err := store.Increment(ctx, tenant, "storage_mb", 50); err != nil {
t.Fatalf("increment: %v", err)
}
_, _, status, err := store.Check(ctx, tenant, "storage_mb")
if err != nil {
t.Fatalf("check: %v", err)
}
if status != StatusOK {
t.Fatalf("bei 50%% erwartet StatusOK, habe %q", status)
}
if err := store.Increment(ctx, tenant, "storage_mb", 35); err != nil { // insgesamt 85%
t.Fatalf("increment: %v", err)
}
_, _, status, err = store.Check(ctx, tenant, "storage_mb")
if err != nil {
t.Fatalf("check: %v", err)
}
if status != StatusWarning {
t.Fatalf("bei 85%% erwartet StatusWarning, habe %q", status)
}
}
func TestCheck_NoQuotaMeansUnlimited(t *testing.T) {
store, cleanup := setupTest(t)
defer cleanup()
ctx := context.Background()
tenant := newTenantID()
if err := store.Increment(ctx, tenant, "api_calls", 1_000_000); err != nil {
t.Fatalf("increment: %v", err)
}
_, _, status, err := store.Check(ctx, tenant, "api_calls")
if err != nil {
t.Fatalf("check: %v", err)
}
if status != StatusOK {
t.Fatalf("ohne konfigurierte quota erwartet StatusOK, habe %q", status)
}
}
func TestRunPeriodicAggregation_CallsRepeatedly(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond)
defer cancel()
var mu sync.Mutex
calls := 0
RunPeriodicAggregation(ctx, 20*time.Millisecond, func(ctx context.Context) error {
mu.Lock()
calls++
mu.Unlock()
return nil
})
mu.Lock()
defer mu.Unlock()
if calls < 3 {
t.Fatalf("erwartet mehrfache aufrufe innerhalb von 120ms bei 20ms interval, habe %d", calls)
}
}
-66
View File
@@ -1,66 +0,0 @@
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)
}
-173
View File
@@ -1,173 +0,0 @@
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)
}
}
-77
View File
@@ -1,77 +0,0 @@
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))
}
-173
View File
@@ -1,173 +0,0 @@
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))
}
// SetPasswordHash schreibt einen bereits berechneten bcrypt-Hash (siehe
// internal/auth, IAM-02). Der Store selbst kennt kein Klartext-Passwort.
func (s *TenantUserStore) SetPasswordHash(ctx context.Context, id, hash string) error {
tag, err := s.pool.Exec(ctx, `
UPDATE users SET password_hash = $2, updated_at = now() WHERE id = $1
`, id, hash)
if err != nil {
return fmt.Errorf("passwort setzen: %w", err)
}
if tag.RowsAffected() == 0 {
return ErrNotFound
}
return nil
}
// AuthCredentials wird ausschliesslich fuer den Login-Pfad (internal/auth)
// verwendet und traegt bewusst den password_hash, damit er nicht ueber den
// regulaeren User-Typ/JSON-Serialisierungspfad nach aussen dringen kann.
type AuthCredentials struct {
User User
PasswordHash string
}
// GetByEmailForAuth liefert Benutzer + Passwort-Hash zu einer E-Mail-Adresse
// aus GENAU DIESER Tenant-Datenbank — der Tenant-Scope ergibt sich damit
// zwingend aus dem verwendeten Pool, es gibt keine Moeglichkeit, versehentlich
// ueber Tenant-Grenzen hinweg zu suchen (bekannter archivmail-Fehler, siehe
// IAM-02 "Bekannte Fehler vermeiden").
func (s *TenantUserStore) GetByEmailForAuth(ctx context.Context, email string) (AuthCredentials, error) {
var c AuthCredentials
row := s.pool.QueryRow(ctx, `
SELECT id, email, name, status, created_at, updated_at, password_hash
FROM users WHERE email = $1
`, email)
if err := row.Scan(&c.User.ID, &c.User.Email, &c.User.Name, &c.User.Status,
&c.User.CreatedAt, &c.User.UpdatedAt, &c.PasswordHash); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return AuthCredentials{}, ErrNotFound
}
return AuthCredentials{}, fmt.Errorf("anmeldedaten lesen: %w", err)
}
return c, nil
}
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)
}
-42
View File
@@ -1,42 +0,0 @@
// 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
}
-24
View File
@@ -1,24 +0,0 @@
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)
}
}
}
-1
View File
@@ -1 +0,0 @@
DROP TABLE IF EXISTS superadmins;
-14
View File
@@ -1,14 +0,0 @@
-- 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()
);
+1
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS audit_events;
+17
View File
@@ -0,0 +1,17 @@
-- Zentrales Audit-Log-Modell (AUD-01, siehe core-kanban/tickets/AUD-01.md).
-- Getrennt vom allgemeinen Anwendungs-Log (Akzeptanzkriterium 2): eigene
-- Tabelle, eigenes Paket (internal/audit), kein Log-Framework.
-- tenant_slug ist NOT NULL + darf nicht leer sein (Akzeptanzkriterium 2 /
-- Pruefung 2) — mandantenuebergreifende Ereignisse nutzen den reservierten
-- Wert 'system', niemals NULL oder leeren String.
CREATE TABLE audit_events (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
occurred_at TIMESTAMPTZ NOT NULL DEFAULT now(),
tenant_slug TEXT NOT NULL CHECK (tenant_slug <> ''),
actor TEXT NOT NULL CHECK (actor <> ''),
action TEXT NOT NULL CHECK (action <> ''),
target TEXT NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}'::jsonb
);
CREATE INDEX audit_events_tenant_slug_idx ON audit_events (tenant_slug, occurred_at);
-2
View File
@@ -1,2 +0,0 @@
DROP TABLE IF EXISTS config_value_history;
DROP TABLE IF EXISTS config_values;
-23
View File
@@ -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
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS feature_flags;
+10
View File
@@ -0,0 +1,10 @@
-- 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
View File
@@ -0,0 +1 @@
DROP TABLE IF EXISTS tenant_licenses;
+12
View File
@@ -0,0 +1,12 @@
-- Lizenzumfang pro Mandant (LIC-01, siehe core-kanban/tickets/LIC-01.md).
-- Genau ein Lizenzdatensatz pro Tenant (tenant_id PK) — ein neues Einspielen
-- ersetzt den vorherigen Datensatz vollstaendig statt eine Historie zu fuehren.
CREATE TABLE tenant_licenses (
tenant_id UUID PRIMARY KEY REFERENCES tenants(id),
plan TEXT NOT NULL,
modules TEXT[] NOT NULL,
issued_at TIMESTAMPTZ NOT NULL,
valid_until TIMESTAMPTZ NOT NULL,
raw_key TEXT NOT NULL,
installed_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE IF EXISTS module_credentials;
DROP TABLE IF EXISTS modules;
+18
View File
@@ -0,0 +1,18 @@
-- 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;
-20
View File
@@ -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);
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE IF EXISTS usage_quotas;
DROP TABLE IF EXISTS usage_counters;
+15
View File
@@ -0,0 +1,15 @@
-- Nutzungszaehler & Quotas je Tenant (LIC-03, siehe core-kanban/tickets/LIC-03.md).
CREATE TABLE usage_counters (
tenant_id UUID NOT NULL,
metric TEXT NOT NULL,
value BIGINT NOT NULL DEFAULT 0,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, metric)
);
CREATE TABLE usage_quotas (
tenant_id UUID NOT NULL,
metric TEXT NOT NULL,
limit_value BIGINT NOT NULL,
PRIMARY KEY (tenant_id, metric)
);
@@ -1,2 +0,0 @@
DROP TABLE IF EXISTS in_app_notifications;
DROP TABLE IF EXISTS notification_templates;
@@ -1,28 +0,0 @@
-- Benachrichtigungs-Kanaele: Vorlagen (E-Mail) + In-App-Nachrichten
-- (CFG-03, siehe core-kanban/tickets/CFG-03.md). Beide leben in der
-- Registry-DB, analog zu feature_flags/config_values — modulübergreifende
-- Konfiguration/UI-Zustand, keine Mandanten-Geschaeftsdaten.
-- tenant_slug = 'global' ist der Fallback-Wert, wenn ein Tenant keine
-- eigene Vorlage gesetzt hat (Akzeptanzkriterium 3: Vorlagen pro Tenant
-- anpassbar, mit sinnvollem Default).
CREATE TABLE notification_templates (
tenant_slug TEXT NOT NULL,
key TEXT NOT NULL,
subject TEXT NOT NULL,
body TEXT NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_slug, key)
);
CREATE TABLE in_app_notifications (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_slug TEXT NOT NULL,
user_id TEXT NOT NULL,
title TEXT NOT NULL,
body TEXT NOT NULL,
read_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE INDEX in_app_notifications_user_idx ON in_app_notifications (tenant_slug, user_id, created_at);
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE resync_usage_buffer;
DROP TABLE resync_audit_buffer;
+22
View File
@@ -0,0 +1,22 @@
-- Wiederanlauf & Nachsynchronisierung nach Core-Ausfall (API-06, siehe
-- core-kanban/tickets/API-06.md) — lokale, PERSISTENTE Pufferqueue (kein
-- In-Memory) fuer Audit-Events und Nutzungszaehler-Deltas eines Moduls.
CREATE TABLE resync_audit_buffer (
id BIGSERIAL PRIMARY KEY,
seq BIGSERIAL,
tenant_slug TEXT NOT NULL,
actor TEXT NOT NULL,
action TEXT NOT NULL,
target TEXT NOT NULL DEFAULT '',
metadata JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
CREATE TABLE resync_usage_buffer (
id BIGSERIAL PRIMARY KEY,
seq BIGSERIAL,
tenant_slug TEXT NOT NULL,
metric TEXT NOT NULL,
delta BIGINT NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
@@ -1 +0,0 @@
DROP TABLE notification_preferences;
@@ -1,17 +0,0 @@
-- CFG-04: Benachrichtigungspraeferenzen je Benutzer, Ereignistyp und Kanal.
-- Lebt wie notification_templates/in_app_notifications (CFG-03) in der
-- Registry-DB — modulübergreifende Konfiguration, keine Mandanten-Geschaeftsdaten.
--
-- Kein Row = aktiviert (Opt-out-Modell): ein Benutzer verpasst nichts, bis er
-- aktiv einen Kanal/Ereignistyp abschaltet — sicherer Default als Opt-in.
CREATE TABLE notification_preferences (
tenant_slug TEXT NOT NULL,
user_id TEXT NOT NULL,
event_type TEXT NOT NULL,
channel TEXT NOT NULL,
enabled BOOLEAN NOT NULL,
updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_slug, user_id, event_type, channel)
);
CREATE INDEX notification_preferences_tenant_idx ON notification_preferences (tenant_slug);
-1
View File
@@ -1 +0,0 @@
DROP TABLE IF EXISTS users;
-16
View File
@@ -1,16 +0,0 @@
-- 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()
);
@@ -1 +0,0 @@
ALTER TABLE users DROP COLUMN password_hash;
@@ -1,3 +0,0 @@
-- Passwort-Hash-Spalte fuer Login (IAM-02, siehe core-kanban/tickets/IAM-02.md).
-- Enthaelt AUSSCHLIESSLICH den bcrypt-Hash, niemals das Klartext-Passwort.
ALTER TABLE users ADD COLUMN password_hash TEXT NOT NULL DEFAULT '';
+13
View File
@@ -1,11 +1,24 @@
#!/usr/bin/env bash
# Setzt die nexarch-Testumgebung zurueck: loescht die geteilte
# Registry-Tabelle "tenants" in der postgres-Wartungsdatenbank sowie alle
# tenant_*-Datenbanken. Noetig, weil verschiedene Feature-Branches
# unterschiedliche Registry-Schemata erwarten, aber dieselbe physische
# Postgres-Instanz auf dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
#
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/reset-test-env.sh
set -euo pipefail
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
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;"
dbs=$(psql -h localhost -U "$ROLE" -d postgres -tAc "SELECT datname FROM pg_database WHERE datname LIKE 'tenant\_%' ESCAPE '\'")
for db in $dbs; do
psql -h localhost -U "$ROLE" -d postgres -v ON_ERROR_STOP=1 -c "DROP DATABASE IF EXISTS \"${db}\";"
done
echo "Testumgebung zurueckgesetzt: registry-tabelle + $(echo "$dbs" | grep -c . || true) tenant-datenbank(en) entfernt."
+12
View File
@@ -1,12 +1,24 @@
#!/usr/bin/env bash
# Ein-Kommando-Pruefung fuer den aktuellen Code-Stand auf dem Testhost:
# Registry+Tenant-DBs zuruecksetzen, dann build/vet/test in einem Rutsch.
# -p 1 ist Pflicht, da mehrere Pakete dieselbe physische Registry-Tabelle auf
# dem Testhost teilen (siehe [[project-nexarch-test-infra]]).
#
# Aufruf: NEXARCH_TEST_DB_PASSWORD=... ./scripts/run-checks.sh
set -euo pipefail
PASS="${NEXARCH_TEST_DB_PASSWORD:?Setze NEXARCH_TEST_DB_PASSWORD vor dem Aufruf}"
cd "$(dirname "$0")/.."
NEXARCH_TEST_DB_PASSWORD="$PASS" bash scripts/reset-test-env.sh
export TEST_ADMIN_DSN="postgresql://nexarch_test:${PASS}@localhost:5432/postgres?sslmode=disable"
echo "== go build =="
go build ./...
echo "== go vet =="
go vet ./...
echo "== go test (-p 1) =="
go test ./... -p 1 -count=1
-30
View File
@@ -1,30 +0,0 @@
import { ThemeProvider, I18nProvider, ToastProvider, typography } from "@nexarch/shl";
export const metadata = {
title: "NEXARCH Benachrichtigungen",
};
export default function RootLayout({
children,
}: {
children: React.ReactNode;
}) {
return (
<html lang="de">
<body
style={{
fontFamily: typography.fontFamily,
margin: 0,
background: "var(--shl-color-background, #ffffff)",
color: "var(--shl-color-text-primary, #14181f)",
}}
>
<ThemeProvider>
<I18nProvider initialLocale="de">
<ToastProvider>{children}</ToastProvider>
</I18nProvider>
</ThemeProvider>
</body>
</html>
);
}
-43
View File
@@ -1,43 +0,0 @@
"use client";
import { useEffect, useState } from "react";
import { Table } from "@nexarch/shl";
import type { TableColumn } from "@nexarch/shl";
import { Preference, fetchTenantOverview } from "../../lib/api";
// Akzeptanzkriterium 3: Tenant-Admin sieht eine Übersicht der
// Benachrichtigungs-Konfiguration seines Tenants. Zugriffsbeschränkung
// (nur Tenant-Admin) ist RBAC-02s Aufgabe vor diesem Endpunkt, siehe
// internal/notifyprefs/handler.go.
export default function OverviewPage() {
const [prefs, setPrefs] = useState<Preference[] | null>(null);
const [error, setError] = useState<string | null>(null);
useEffect(() => {
fetchTenantOverview()
.then(setPrefs)
.catch(() => setError("Übersicht konnte nicht geladen werden. Nur Tenant-Admins haben Zugriff."));
}, []);
const columns: TableColumn<Preference>[] = [
{ key: "user", header: "Benutzer-ID", render: (p) => p.UserID },
{ key: "event", header: "Ereignistyp", render: (p) => p.EventType },
{ key: "channel", header: "Kanal", render: (p) => p.Channel },
{ key: "enabled", header: "Aktiviert", render: (p) => (p.Enabled ? "Ja" : "Nein") },
];
return (
<main style={{ maxWidth: 800, margin: "40px auto", padding: "0 16px" }}>
<h1>Benachrichtigungs-Übersicht (Mandant)</h1>
{error && <p role="alert">{error}</p>}
{prefs && (
<Table
columns={columns}
rows={prefs}
rowKey={(p) => `${p.UserID}-${p.EventType}-${p.Channel}`}
caption="Von der Standardeinstellung abweichende Benachrichtigungspräferenzen aller Benutzer"
/>
)}
</main>
);
}
-5
View File
@@ -1,5 +0,0 @@
import { redirect } from "next/navigation";
export default function IndexPage() {
redirect("/settings");
}
-97
View File
@@ -1,97 +0,0 @@
"use client";
import { useEffect, useState } from "react";
import { CheckboxField, useToast } from "@nexarch/shl";
import {
ApiError,
KNOWN_CHANNELS,
KNOWN_EVENT_TYPES,
Preference,
fetchMyPreferences,
setPreference,
} from "../../lib/api";
const EVENT_LABELS: Record<string, string> = {
welcome: "Willkommen",
invoice_ready: "Rechnung verfügbar",
password_reset: "Passwort-Zurücksetzung",
security_alert: "Sicherheitshinweis",
};
const CHANNEL_LABELS: Record<string, string> = {
email: "E-Mail",
in_app: "In-App",
};
export default function SettingsPage() {
const { push } = useToast();
const [prefs, setPrefs] = useState<Preference[] | null>(null);
const [loadError, setLoadError] = useState<string | null>(null);
const [pending, setPending] = useState<string | null>(null);
useEffect(() => {
fetchMyPreferences()
.then(setPrefs)
.catch(() => setLoadError("Einstellungen konnten nicht geladen werden. Bitte melden Sie sich an."));
}, []);
// Default: aktiviert, solange keine explizite Praeferenz existiert
// (Akzeptanzkriterium 1) — gleicher Opt-out-Default wie im Backend.
function isEnabled(eventType: string, channel: string): boolean {
const explicit = prefs?.find((p) => p.EventType === eventType && p.Channel === channel);
return explicit ? explicit.Enabled : true;
}
async function onToggle(eventType: string, channel: string, nextEnabled: boolean) {
const key = `${eventType}:${channel}`;
setPending(key);
try {
// Akzeptanzkriterium 2: sofort speichern, kein Sammel-Speichern-Button —
// wirkt unmittelbar auf künftige Zustellungen (EnqueueIfAllowed prüft
// bei jedem Aufruf live, kein Cache dazwischen).
await setPreference(eventType, channel, nextEnabled);
setPrefs((current) => {
const withoutThis = (current ?? []).filter((p) => !(p.EventType === eventType && p.Channel === channel));
return [...withoutThis, { TenantSlug: "", UserID: "", EventType: eventType, Channel: channel, Enabled: nextEnabled }];
});
push(nextEnabled ? "Benachrichtigung aktiviert." : "Benachrichtigung deaktiviert.", "success");
} catch (err) {
push(err instanceof ApiError ? err.message : "Einstellung konnte nicht gespeichert werden.", "danger");
} finally {
setPending(null);
}
}
if (loadError) {
return (
<main style={{ maxWidth: 640, margin: "40px auto", padding: "0 16px" }}>
<p role="alert">{loadError}</p>
</main>
);
}
return (
<main style={{ maxWidth: 640, margin: "40px auto", padding: "0 16px" }}>
<h1>Benachrichtigungseinstellungen</h1>
<p>Wählen Sie je Ereignis, über welche Kanäle Sie benachrichtigt werden möchten.</p>
{KNOWN_EVENT_TYPES.map((eventType) => (
<fieldset key={eventType} style={{ marginBottom: 16 }}>
<legend>{EVENT_LABELS[eventType] ?? eventType}</legend>
{KNOWN_CHANNELS.map((channel) => {
const key = `${eventType}:${channel}`;
return (
<CheckboxField
key={key}
label={CHANNEL_LABELS[channel] ?? channel}
checked={isEnabled(eventType, channel)}
disabled={pending === key}
onChange={(e) => onToggle(eventType, channel, e.target.checked)}
/>
);
})}
</fieldset>
))}
</main>
);
}

Some files were not shown because too many files have changed in this diff Show More