diff --git a/internal/moduletrust/cache.go b/internal/moduletrust/cache.go new file mode 100644 index 0000000..246b21a --- /dev/null +++ b/internal/moduletrust/cache.go @@ -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 +} diff --git a/internal/moduletrust/issuer.go b/internal/moduletrust/issuer.go new file mode 100644 index 0000000..da2028f --- /dev/null +++ b/internal/moduletrust/issuer.go @@ -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 +} diff --git a/internal/moduletrust/keys.go b/internal/moduletrust/keys.go new file mode 100644 index 0000000..a3dd1c4 --- /dev/null +++ b/internal/moduletrust/keys.go @@ -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 +} diff --git a/internal/moduletrust/moduletrust_test.go b/internal/moduletrust/moduletrust_test.go new file mode 100644 index 0000000..5cd532a --- /dev/null +++ b/internal/moduletrust/moduletrust_test.go @@ -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) {} diff --git a/internal/moduletrust/verifier.go b/internal/moduletrust/verifier.go new file mode 100644 index 0000000..5a2721a --- /dev/null +++ b/internal/moduletrust/verifier.go @@ -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 +} diff --git a/internal/resync/buffer.go b/internal/resync/buffer.go new file mode 100644 index 0000000..0a587be --- /dev/null +++ b/internal/resync/buffer.go @@ -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 +} diff --git a/internal/resync/handler.go b/internal/resync/handler.go new file mode 100644 index 0000000..8e3e88a --- /dev/null +++ b/internal/resync/handler.go @@ -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) +} diff --git a/internal/resync/resync_test.go b/internal/resync/resync_test.go new file mode 100644 index 0000000..f6b402e --- /dev/null +++ b/internal/resync/resync_test.go @@ -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) + } +} diff --git a/internal/resync/worker.go b/internal/resync/worker.go new file mode 100644 index 0000000..66abd65 --- /dev/null +++ b/internal/resync/worker.go @@ -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) + } + } +} diff --git a/migrations/0006_resync_buffers.down.sql b/migrations/0006_resync_buffers.down.sql new file mode 100644 index 0000000..b4f73ae --- /dev/null +++ b/migrations/0006_resync_buffers.down.sql @@ -0,0 +1,2 @@ +DROP TABLE resync_usage_buffer; +DROP TABLE resync_audit_buffer; diff --git a/migrations/0006_resync_buffers.up.sql b/migrations/0006_resync_buffers.up.sql new file mode 100644 index 0000000..5de9fc1 --- /dev/null +++ b/migrations/0006_resync_buffers.up.sql @@ -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() +);