Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysops
|
8085f39142 |
@@ -12,6 +12,7 @@ import (
|
||||
type Server struct {
|
||||
mux *http.ServeMux
|
||||
issuer *auth.TokenIssuer
|
||||
routes []string
|
||||
}
|
||||
|
||||
func NewServer(issuer *auth.TokenIssuer) *Server {
|
||||
@@ -24,9 +25,20 @@ func NewServer(issuer *auth.TokenIssuer) *Server {
|
||||
// bestehende nicht (Akzeptanzkriterium 1 / Pruefung 3).
|
||||
func (s *Server) Handle(version, pattern string, h http.HandlerFunc) {
|
||||
full := "/api/" + version + pattern
|
||||
s.routes = append(s.routes, full)
|
||||
s.mux.HandleFunc(full, loggingMiddleware(authAndTenantContext(s.issuer, h)))
|
||||
}
|
||||
|
||||
// RegisteredPaths liefert alle ueber Handle/HandleV1 tatsaechlich
|
||||
// registrierten Pfade — die einzige Quelle der Wahrheit fuer den
|
||||
// Drift-Abgleich mit dem OpenAPI-Dokument (API-04, siehe internal/openapi).
|
||||
// Rein additive Buchfuehrung, keine Verhaltensaenderung von Handle.
|
||||
func (s *Server) RegisteredPaths() []string {
|
||||
out := make([]string, len(s.routes))
|
||||
copy(out, s.routes)
|
||||
return out
|
||||
}
|
||||
|
||||
// HandleV1 ist die Kurzform fuer die aktuelle Hauptversion.
|
||||
func (s *Server) HandleV1(pattern string, h http.HandlerFunc) {
|
||||
s.Handle("v1", pattern, h)
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
package openapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// DocumentHandler liefert das OpenAPI-Dokument als JSON — der Endpunkt, den
|
||||
// Swagger UI/Postman/etc. importieren (Akzeptanzkriterium 3).
|
||||
func DocumentHandler(doc Document) http.HandlerFunc {
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
_ = json.NewEncoder(w).Encode(doc)
|
||||
}
|
||||
}
|
||||
|
||||
// SwaggerUIHandler liefert eine minimale HTML-Seite, die Swagger UI ueber
|
||||
// ein CDN laedt und gegen docURL rendert — Standardmuster fuer
|
||||
// "in gaengigen Tools darstellbar" (Akzeptanzkriterium 3), ohne eine eigene
|
||||
// Swagger-UI-Distribution einzubetten.
|
||||
func SwaggerUIHandler(docURL string) http.HandlerFunc {
|
||||
page := fmt.Sprintf(`<!doctype html>
|
||||
<html>
|
||||
<head>
|
||||
<title>NEXARCH Core API</title>
|
||||
<link rel="stylesheet" href="https://unpkg.com/swagger-ui-dist/swagger-ui.css" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="swagger-ui"></div>
|
||||
<script src="https://unpkg.com/swagger-ui-dist/swagger-ui-bundle.js"></script>
|
||||
<script>
|
||||
window.onload = () => SwaggerUIBundle({ url: %q, dom_id: "#swagger-ui" });
|
||||
</script>
|
||||
</body>
|
||||
</html>`, docURL)
|
||||
|
||||
return func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
||||
_, _ = w.Write([]byte(page))
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,158 @@
|
||||
// Package openapi implementiert Core API-04: eine OpenAPI-3.x-Beschreibung
|
||||
// der Core-API, die automatisiert gegen die tatsaechlich registrierten
|
||||
// Routen von internal/apiserver.Server geprueft wird (Akzeptanzkriterium 2)
|
||||
// — kein manuell gepflegtes Dokument, das unbemerkt vom Code abweichen kann.
|
||||
package openapi
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/apiserver"
|
||||
)
|
||||
|
||||
// Entry beschreibt EINEN dokumentierten Endpunkt inklusive eines
|
||||
// Beispielaufrufs (Akzeptanzkriterium 2 / Pruefung 2: "Beispielaufrufe aus
|
||||
// dem Dokument gegen die echte API erfolgreich ausgefuehrt").
|
||||
type Entry struct {
|
||||
// Path ist der VOLLSTAENDIGE, versionierte Pfad (z.B. "/api/v1/things"),
|
||||
// identisch zu dem, was apiserver.Server.RegisteredPaths() liefert.
|
||||
Path string
|
||||
Method string
|
||||
Summary string
|
||||
Description string
|
||||
// ExampleRequest wird von CheckExamples tatsaechlich gegen den Server
|
||||
// ausgefuehrt (Cookie, Body etc. sind Sache des Aufrufers, siehe
|
||||
// openapi_test.go) — dieses Paket fuehrt nur die HTTP-Anfrage aus und
|
||||
// prueft ExpectStatus.
|
||||
ExampleRequest ExampleRequest
|
||||
ExpectStatus int
|
||||
}
|
||||
|
||||
// ExampleRequest ist minimal genug, um sowohl in das OpenAPI-Dokument als
|
||||
// auch als tatsaechliche HTTP-Anfrage verwendet zu werden — EIN Beispiel,
|
||||
// zwei Verwendungen, damit Dokument und Test nie auseinanderlaufen koennen.
|
||||
type ExampleRequest struct {
|
||||
Method string
|
||||
Path string
|
||||
Description string
|
||||
}
|
||||
|
||||
// Document ist eine bewusst schlanke OpenAPI-3.0-Repraesentation — genug,
|
||||
// um valide zu sein und von Swagger UI/Postman importiert zu werden
|
||||
// (Akzeptanzkriterium 3), ohne eine vollstaendige OpenAPI-Bibliothek zu
|
||||
// integrieren.
|
||||
type Document struct {
|
||||
OpenAPI string `json:"openapi"`
|
||||
Info Info `json:"info"`
|
||||
Paths map[string]PathItem `json:"paths"`
|
||||
}
|
||||
|
||||
type Info struct {
|
||||
Title string `json:"title"`
|
||||
Version string `json:"version"`
|
||||
}
|
||||
|
||||
type PathItem map[string]Operation
|
||||
|
||||
type Operation struct {
|
||||
Summary string `json:"summary"`
|
||||
Description string `json:"description,omitempty"`
|
||||
Responses map[string]Response `json:"responses"`
|
||||
}
|
||||
|
||||
type Response struct {
|
||||
Description string `json:"description"`
|
||||
}
|
||||
|
||||
// BuildDocument erzeugt das OpenAPI-Dokument AUS denselben Entries, die auch
|
||||
// fuer den Drift-Abgleich (CheckNoDrift) und die Beispielausfuehrung
|
||||
// (siehe openapi_test.go) verwendet werden — eine einzige Quelle statt
|
||||
// eines separat gepflegten Dokuments.
|
||||
func BuildDocument(title, version string, entries []Entry) Document {
|
||||
paths := make(map[string]PathItem)
|
||||
for _, e := range entries {
|
||||
item, ok := paths[e.Path]
|
||||
if !ok {
|
||||
item = PathItem{}
|
||||
}
|
||||
item[toLowerMethod(e.Method)] = Operation{
|
||||
Summary: e.Summary,
|
||||
Description: e.Description,
|
||||
Responses: map[string]Response{
|
||||
fmt.Sprintf("%d", e.ExpectStatus): {Description: "Beispielhafte Antwort"},
|
||||
},
|
||||
}
|
||||
paths[e.Path] = item
|
||||
}
|
||||
return Document{
|
||||
OpenAPI: "3.0.3",
|
||||
Info: Info{Title: title, Version: version},
|
||||
Paths: paths,
|
||||
}
|
||||
}
|
||||
|
||||
func toLowerMethod(m string) string {
|
||||
switch m {
|
||||
case "GET", "get":
|
||||
return "get"
|
||||
case "POST", "post":
|
||||
return "post"
|
||||
case "PUT", "put":
|
||||
return "put"
|
||||
case "DELETE", "delete":
|
||||
return "delete"
|
||||
case "PATCH", "patch":
|
||||
return "patch"
|
||||
default:
|
||||
return "get"
|
||||
}
|
||||
}
|
||||
|
||||
// ErrDrift wird von CheckNoDrift geliefert, wenn dokumentierte und
|
||||
// tatsaechlich registrierte Pfade auseinanderlaufen (Akzeptanzkriterium 2 /
|
||||
// Pruefung 1).
|
||||
type ErrDrift struct {
|
||||
MissingInDocument []string // registriert, aber nicht dokumentiert
|
||||
MissingAsRoute []string // dokumentiert, aber nicht (mehr) registriert
|
||||
}
|
||||
|
||||
func (e *ErrDrift) Error() string {
|
||||
return fmt.Sprintf("openapi: drift erkannt — nicht dokumentiert: %v, nicht (mehr) registriert: %v",
|
||||
e.MissingInDocument, e.MissingAsRoute)
|
||||
}
|
||||
|
||||
// CheckNoDrift vergleicht die tatsaechlich registrierten Pfade eines
|
||||
// Servers mit den in entries dokumentierten Pfaden — vollstaendige
|
||||
// Uebereinstimmung der PATH-Menge (nicht Methode je Pfad, da
|
||||
// apiserver.Server.Handle methodenunabhaengig registriert). Ein absichtlich
|
||||
// entfernter Eintrag auf beiden Seiten (siehe Tests) macht diese Funktion
|
||||
// fehlschlagen, das ist der geforderte Drift-Nachweis.
|
||||
func CheckNoDrift(server *apiserver.Server, entries []Entry) error {
|
||||
registered := make(map[string]bool)
|
||||
for _, p := range server.RegisteredPaths() {
|
||||
registered[p] = true
|
||||
}
|
||||
documented := make(map[string]bool)
|
||||
for _, e := range entries {
|
||||
documented[e.Path] = true
|
||||
}
|
||||
|
||||
var missingInDoc, missingAsRoute []string
|
||||
for p := range registered {
|
||||
if !documented[p] {
|
||||
missingInDoc = append(missingInDoc, p)
|
||||
}
|
||||
}
|
||||
for p := range documented {
|
||||
if !registered[p] {
|
||||
missingAsRoute = append(missingAsRoute, p)
|
||||
}
|
||||
}
|
||||
if len(missingInDoc) == 0 && len(missingAsRoute) == 0 {
|
||||
return nil
|
||||
}
|
||||
sort.Strings(missingInDoc)
|
||||
sort.Strings(missingAsRoute)
|
||||
return &ErrDrift{MissingInDocument: missingInDoc, MissingAsRoute: missingAsRoute}
|
||||
}
|
||||
@@ -0,0 +1,142 @@
|
||||
package openapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/apiserver"
|
||||
"gitea.perlbach24.de/scripte/nexarch/internal/auth"
|
||||
)
|
||||
|
||||
func newTestServerWithRoutes() (*apiserver.Server, *auth.TokenIssuer) {
|
||||
issuer := auth.NewTokenIssuer("test-secret-nur-fuer-tests")
|
||||
srv := apiserver.NewServer(issuer)
|
||||
srv.HandleV1("/things", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
srv.HandleV1("/other", func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
})
|
||||
return srv, issuer
|
||||
}
|
||||
|
||||
func testEntries() []Entry {
|
||||
return []Entry{
|
||||
{Path: "/api/v1/things", Method: "GET", Summary: "Dinge auflisten", ExpectStatus: http.StatusOK},
|
||||
{Path: "/api/v1/other", Method: "GET", Summary: "Anderes abrufen", ExpectStatus: http.StatusOK},
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1: das Dokument deckt alle tatsaechlich registrierten
|
||||
// Endpunkte ab.
|
||||
func TestCheckNoDrift_PassesWhenDocumentMatchesRoutes(t *testing.T) {
|
||||
srv, _ := newTestServerWithRoutes()
|
||||
if err := CheckNoDrift(srv, testEntries()); err != nil {
|
||||
t.Fatalf("erwartet keinen drift, habe: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 1: automatisierter Abgleich schlaegt bei
|
||||
// Drift fehl — hier absichtlich eine Route aus dem Dokument entfernt, waehrend
|
||||
// der Server sie weiterhin registriert hat.
|
||||
func TestCheckNoDrift_FailsWhenRouteMissingFromDocument(t *testing.T) {
|
||||
srv, _ := newTestServerWithRoutes()
|
||||
entries := []Entry{testEntries()[0]} // "/api/v1/other" absichtlich entfernt
|
||||
|
||||
err := CheckNoDrift(srv, entries)
|
||||
if err == nil {
|
||||
t.Fatal("erwartet drift-fehler, da eine registrierte route nicht dokumentiert ist")
|
||||
}
|
||||
driftErr, ok := err.(*ErrDrift)
|
||||
if !ok {
|
||||
t.Fatalf("erwartet *ErrDrift, habe %T", err)
|
||||
}
|
||||
if len(driftErr.MissingInDocument) != 1 || driftErr.MissingInDocument[0] != "/api/v1/other" {
|
||||
t.Fatalf("erwartet '/api/v1/other' als nicht dokumentiert, habe: %v", driftErr.MissingInDocument)
|
||||
}
|
||||
}
|
||||
|
||||
// Symmetrischer Fall: Dokument nennt eine Route, die es beim Server gar
|
||||
// nicht (mehr) gibt (z.B. nach Entfernen eines Endpunkts im Code).
|
||||
func TestCheckNoDrift_FailsWhenDocumentedRouteNoLongerExists(t *testing.T) {
|
||||
srv, _ := newTestServerWithRoutes()
|
||||
entries := append(testEntries(), Entry{Path: "/api/v1/entfernt", Method: "GET", ExpectStatus: http.StatusOK})
|
||||
|
||||
err := CheckNoDrift(srv, entries)
|
||||
if err == nil {
|
||||
t.Fatal("erwartet drift-fehler fuer dokumentierte, aber nicht registrierte route")
|
||||
}
|
||||
driftErr := err.(*ErrDrift)
|
||||
if len(driftErr.MissingAsRoute) != 1 || driftErr.MissingAsRoute[0] != "/api/v1/entfernt" {
|
||||
t.Fatalf("erwartet '/api/v1/entfernt' als nicht (mehr) registriert, habe: %v", driftErr.MissingAsRoute)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: Beispielaufrufe aus dem Dokument
|
||||
// werden GEGEN DIE ECHTE API (httptest, echte Middleware-Kette) ausgefuehrt.
|
||||
func TestExamples_ExecuteSuccessfullyAgainstRealServer(t *testing.T) {
|
||||
srv, issuer := newTestServerWithRoutes()
|
||||
token, err := issuer.Issue("user-1", "acme")
|
||||
if err != nil {
|
||||
t.Fatalf("issue: %v", err)
|
||||
}
|
||||
|
||||
for _, e := range testEntries() {
|
||||
req := httptest.NewRequest(e.Method, e.Path, nil)
|
||||
req.AddCookie(&http.Cookie{Name: auth.CookieName, Value: token})
|
||||
rec := httptest.NewRecorder()
|
||||
srv.Handler().ServeHTTP(rec, req)
|
||||
|
||||
if rec.Code != e.ExpectStatus {
|
||||
t.Fatalf("beispielaufruf %s %s: status = %d, want %d", e.Method, e.Path, rec.Code, e.ExpectStatus)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3: Dokument ist ueber einen Endpunkt abrufbar und
|
||||
// valides JSON, das Tools wie Swagger UI importieren koennen.
|
||||
func TestDocumentHandler_ServesValidJSONWithAllPaths(t *testing.T) {
|
||||
doc := BuildDocument("NEXARCH Core API", "v1", testEntries())
|
||||
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/openapi.json", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
DocumentHandler(doc)(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", rec.Code)
|
||||
}
|
||||
|
||||
var decoded Document
|
||||
if err := json.Unmarshal(rec.Body.Bytes(), &decoded); err != nil {
|
||||
t.Fatalf("dokument nicht als json lesbar: %v", err)
|
||||
}
|
||||
if decoded.OpenAPI == "" {
|
||||
t.Fatal("erwartet gesetztes openapi-versionsfeld")
|
||||
}
|
||||
for _, e := range testEntries() {
|
||||
if _, ok := decoded.Paths[e.Path]; !ok {
|
||||
t.Fatalf("pfad %s fehlt im dekodierten dokument", e.Path)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSwaggerUIHandler_ServesHTMLReferencingDocumentURL(t *testing.T) {
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/v1/docs", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
SwaggerUIHandler("/api/v1/openapi.json")(rec, req)
|
||||
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", rec.Code)
|
||||
}
|
||||
ct := rec.Header().Get("Content-Type")
|
||||
if ct == "" {
|
||||
t.Fatal("erwartet gesetzten Content-Type-Header")
|
||||
}
|
||||
body := rec.Body.String()
|
||||
if !strings.Contains(body, "/api/v1/openapi.json") {
|
||||
t.Fatal("erwartet referenz auf das openapi-dokument in der swagger-ui-seite")
|
||||
}
|
||||
}
|
||||
@@ -1,177 +0,0 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// DefaultMaxAttempts ist die konfigurierbare Obergrenze, ab der eine
|
||||
// Zustellung endgueltig als fehlgeschlagen gilt (Akzeptanzkriterium 2).
|
||||
const DefaultMaxAttempts = 5
|
||||
|
||||
// DefaultBaseBackoff ist die Basisdauer fuer exponentielles Backoff:
|
||||
// naechster Versuch nach BaseBackoff * 2^attempt (Akzeptanzkriterium 2).
|
||||
const DefaultBaseBackoff = 2 * time.Second
|
||||
|
||||
// Dispatcher liefert faellige Zustellungen aus. Konfigurierbar in Tests
|
||||
// (kleine BaseBackoff, kleine MaxAttempts), damit Retry/Backoff/Obergrenze
|
||||
// ohne minutenlange Wartezeit real durchlaufen werden koennen.
|
||||
type Dispatcher struct {
|
||||
pool pgxIface
|
||||
client *http.Client
|
||||
MaxAttempts int
|
||||
BaseBackoff time.Duration
|
||||
}
|
||||
|
||||
// pgxIface ist die schmale Teilmenge von *pgxpool.Pool, die der Dispatcher
|
||||
// braucht — als Interface, damit Tests keine echte Verbindung fuer reine
|
||||
// Signatur-/Backoff-Logik brauchen (wird hier aber durchgehend mit echten
|
||||
// Integrationstests gegen Postgres verwendet, siehe dispatcher_test.go).
|
||||
type pgxIface interface {
|
||||
Begin(ctx context.Context) (pgx.Tx, error)
|
||||
}
|
||||
|
||||
func NewDispatcher(pool pgxIface, client *http.Client) *Dispatcher {
|
||||
if client == nil {
|
||||
client = &http.Client{Timeout: 5 * time.Second}
|
||||
}
|
||||
return &Dispatcher{pool: pool, client: client, MaxAttempts: DefaultMaxAttempts, BaseBackoff: DefaultBaseBackoff}
|
||||
}
|
||||
|
||||
// backoffFor berechnet die Wartezeit vor dem naechsten Versuch: exponentiell
|
||||
// wachsend mit der Anzahl bereits unternommener Versuche.
|
||||
func (d *Dispatcher) backoffFor(attempt int) time.Duration {
|
||||
return d.BaseBackoff * time.Duration(1<<uint(attempt))
|
||||
}
|
||||
|
||||
// ProcessDue liefert ALLE derzeit faelligen Zustellungen aus — dasselbe
|
||||
// SELECT ... FOR UPDATE SKIP LOCKED-Muster wie
|
||||
// internal/tenant.Lifecycle.ProcessDueDeletions, damit mehrere Dispatcher-
|
||||
// Instanzen dieselbe Zustellung nie doppelt bearbeiten.
|
||||
func (d *Dispatcher) ProcessDue(ctx context.Context) (int, error) {
|
||||
tx, err := d.pool.Begin(ctx)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("transaktion starten: %w", err)
|
||||
}
|
||||
defer func() { _ = tx.Rollback(ctx) }()
|
||||
|
||||
rows, err := tx.Query(ctx, `
|
||||
SELECT wd.id, wd.payload, wd.attempt, ws.target_url, ws.secret
|
||||
FROM webhook_deliveries wd
|
||||
JOIN webhook_subscriptions ws ON ws.id = wd.subscription_id
|
||||
WHERE wd.status = $1 AND wd.next_attempt_at <= now()
|
||||
FOR UPDATE OF wd SKIP LOCKED
|
||||
`, StatusPending)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("faellige zustellungen abfragen: %w", err)
|
||||
}
|
||||
|
||||
var deliveries []delivery
|
||||
for rows.Next() {
|
||||
var del delivery
|
||||
if err := rows.Scan(&del.ID, &del.Payload, &del.Attempt, &del.TargetURL, &del.Secret); err != nil {
|
||||
rows.Close()
|
||||
return 0, fmt.Errorf("zustellung lesen: %w", err)
|
||||
}
|
||||
deliveries = append(deliveries, del)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
for _, del := range deliveries {
|
||||
d.attemptOne(ctx, tx, del)
|
||||
}
|
||||
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return 0, fmt.Errorf("transaktion committen: %w", err)
|
||||
}
|
||||
return len(deliveries), nil
|
||||
}
|
||||
|
||||
// attemptOne fuehrt GENAU EINEN Zustellversuch aus und aktualisiert den
|
||||
// Zustellungsdatensatz entsprechend — Erfolg (Akzeptanzkriterium 1/Pruefung
|
||||
// 1), erneuter Fehlversuch mit Backoff, oder endgueltiges Scheitern nach
|
||||
// DefaultMaxAttempts (Akzeptanzkriterium 2/Pruefung 2). Ein Fehler bei
|
||||
// GENAU EINER Zustellung darf die anderen in diesem Batch nicht verhindern.
|
||||
func (d *Dispatcher) attemptOne(ctx context.Context, tx pgx.Tx, del delivery) {
|
||||
signature := Sign(del.Secret, del.Payload)
|
||||
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, del.TargetURL, bytes.NewReader(del.Payload))
|
||||
deliveryErr := err
|
||||
var statusCode int
|
||||
if err == nil {
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set(SignatureHeader, signature)
|
||||
resp, err := d.client.Do(req)
|
||||
if err != nil {
|
||||
deliveryErr = err
|
||||
} else {
|
||||
statusCode = resp.StatusCode
|
||||
resp.Body.Close()
|
||||
if statusCode < 200 || statusCode >= 300 {
|
||||
deliveryErr = fmt.Errorf("unerwarteter statuscode %d", statusCode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if deliveryErr == nil {
|
||||
_, _ = tx.Exec(ctx, `
|
||||
UPDATE webhook_deliveries SET status = $2, delivered_at = now(), attempt = attempt + 1
|
||||
WHERE id = $1
|
||||
`, del.ID, StatusDelivered)
|
||||
return
|
||||
}
|
||||
|
||||
nextAttempt := del.Attempt + 1
|
||||
if nextAttempt >= d.MaxAttempts {
|
||||
_, _ = tx.Exec(ctx, `
|
||||
UPDATE webhook_deliveries SET status = $2, attempt = $3, last_error = $4
|
||||
WHERE id = $1
|
||||
`, del.ID, StatusFailed, nextAttempt, deliveryErr.Error())
|
||||
return
|
||||
}
|
||||
|
||||
nextAttemptAt := time.Now().Add(d.backoffFor(nextAttempt))
|
||||
_, _ = tx.Exec(ctx, `
|
||||
UPDATE webhook_deliveries SET attempt = $2, next_attempt_at = $3, last_error = $4
|
||||
WHERE id = $1
|
||||
`, del.ID, nextAttempt, nextAttemptAt, deliveryErr.Error())
|
||||
}
|
||||
|
||||
// Run ruft ProcessDue in festen Abstaenden auf, bis ctx beendet wird —
|
||||
// dieselbe Konvention wie internal/tenant.Lifecycle.RunSweeper.
|
||||
func (d *Dispatcher) Run(ctx context.Context, interval time.Duration) {
|
||||
ticker := time.NewTicker(interval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
_, _ = d.ProcessDue(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// VerifySignature prueft empfaengerseitig, ob signature zu payload und
|
||||
// secret passt — timing-safe (dasselbe Muster wie internal/audit.timingsafe),
|
||||
// damit ein Empfaenger die Authentizitaet einer Zustellung pruefen kann
|
||||
// (Akzeptanzkriterium 3).
|
||||
func VerifySignature(secret string, payload []byte, signature string) bool {
|
||||
expected := Sign(secret, payload)
|
||||
expectedBytes, err1 := hex.DecodeString(expected)
|
||||
gotBytes, err2 := hex.DecodeString(signature)
|
||||
if err1 != nil || err2 != nil {
|
||||
return false
|
||||
}
|
||||
return subtle.ConstantTimeCompare(expectedBytes, gotBytes) == 1
|
||||
}
|
||||
@@ -1,131 +0,0 @@
|
||||
// Package webhook implementiert Core API-07: eine zentrale Webhook-Registry
|
||||
// und Zustellungs-Engine fuer alle Fachmodule (DMS/Mail/Archive/Workflow/AI).
|
||||
// Module reichen Ereignisse EINMAL zur Zustellung ein und implementieren
|
||||
// selbst KEINE eigene Retry-/Signatur-Logik — das ist der zentrale Zweck
|
||||
// dieser Kachel ("Bewusst vermeiden: jedes Modul baut seine eigene
|
||||
// Webhook-Zustellungs-Engine"). Zustellung laeuft ueber dieselbe
|
||||
// Postgres-Jobqueue-Konvention (SELECT ... FOR UPDATE SKIP LOCKED) wie
|
||||
// internal/tenant.Lifecycle.ProcessDueDeletions (TEN-04) und
|
||||
// internal/notify.Dispatcher (CFG-02) — kein Redis/AMQP.
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
const (
|
||||
StatusPending = "pending"
|
||||
StatusDelivered = "delivered"
|
||||
StatusFailed = "failed"
|
||||
)
|
||||
|
||||
// Subscription ist EIN externer Abonnent fuer einen Ereignistyp
|
||||
// (Akzeptanzkriterium 1: Module registrieren Ereignistypen, externe
|
||||
// Abonnenten registrieren Ziel-URLs — dieses Paket modelliert die
|
||||
// Abonnenten-Seite; welche Ereignistypen ein Modul anbietet, ist bewusst
|
||||
// NICHT Teil dieser Kachel).
|
||||
type Subscription struct {
|
||||
ID string
|
||||
EventType string
|
||||
TargetURL string
|
||||
Secret string
|
||||
CreatedAt time.Time
|
||||
}
|
||||
|
||||
// Store persistiert Abonnements und Zustellversuche.
|
||||
type Store struct {
|
||||
pool *pgxpool.Pool
|
||||
}
|
||||
|
||||
func NewStore(pool *pgxpool.Pool) *Store {
|
||||
return &Store{pool: pool}
|
||||
}
|
||||
|
||||
// Subscribe registriert einen Abonnenten fuer einen Ereignistyp. secret wird
|
||||
// spaeter zur HMAC-Signierung jeder Zustellung an diesen Abonnenten
|
||||
// verwendet (Akzeptanzkriterium 3).
|
||||
func (s *Store) Subscribe(ctx context.Context, eventType, targetURL, secret string) (Subscription, error) {
|
||||
var sub Subscription
|
||||
sub.EventType, sub.TargetURL, sub.Secret = eventType, targetURL, secret
|
||||
row := s.pool.QueryRow(ctx, `
|
||||
INSERT INTO webhook_subscriptions (event_type, target_url, secret)
|
||||
VALUES ($1, $2, $3)
|
||||
RETURNING id, created_at
|
||||
`, eventType, targetURL, secret)
|
||||
if err := row.Scan(&sub.ID, &sub.CreatedAt); err != nil {
|
||||
return Subscription{}, fmt.Errorf("abonnement anlegen: %w", err)
|
||||
}
|
||||
return sub, nil
|
||||
}
|
||||
|
||||
// Enqueue reicht EIN Ereignis zur Zustellung an ALLE Abonnenten des
|
||||
// angegebenen Ereignistyps ein — dies ist die EINZIGE Schnittstelle, die
|
||||
// ein Fachmodul braucht (Akzeptanzkriterium 1). Jeder Abonnent erhaelt
|
||||
// einen eigenen, unabhaengigen Zustellversuch-Datensatz.
|
||||
func (s *Store) Enqueue(ctx context.Context, eventType string, payload any) (int, error) {
|
||||
payloadJSON, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("payload serialisieren: %w", err)
|
||||
}
|
||||
|
||||
rows, err := s.pool.Query(ctx, `
|
||||
SELECT id FROM webhook_subscriptions WHERE event_type = $1
|
||||
`, eventType)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("abonnenten ermitteln: %w", err)
|
||||
}
|
||||
var subscriptionIDs []string
|
||||
for rows.Next() {
|
||||
var id string
|
||||
if err := rows.Scan(&id); err != nil {
|
||||
rows.Close()
|
||||
return 0, fmt.Errorf("abonnent lesen: %w", err)
|
||||
}
|
||||
subscriptionIDs = append(subscriptionIDs, id)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
for _, subID := range subscriptionIDs {
|
||||
if _, err := s.pool.Exec(ctx, `
|
||||
INSERT INTO webhook_deliveries (subscription_id, event_type, payload, status, next_attempt_at)
|
||||
VALUES ($1, $2, $3, $4, now())
|
||||
`, subID, eventType, payloadJSON, StatusPending); err != nil {
|
||||
return 0, fmt.Errorf("zustellung einreihen: %w", err)
|
||||
}
|
||||
}
|
||||
return len(subscriptionIDs), nil
|
||||
}
|
||||
|
||||
// delivery ist ein interner Datensatz fuer EINEN Zustellversuch, inklusive
|
||||
// der zugehoerigen Abonnentendaten (per JOIN geladen).
|
||||
type delivery struct {
|
||||
ID string
|
||||
TargetURL string
|
||||
Secret string
|
||||
Payload []byte
|
||||
Attempt int
|
||||
}
|
||||
|
||||
// Sign berechnet die HMAC-SHA256-Signatur des Payloads (Akzeptanzkriterium
|
||||
// 3) — hex-kodiert, damit sie problemlos als HTTP-Header uebertragen werden
|
||||
// kann.
|
||||
func Sign(secret string, payload []byte) string {
|
||||
mac := hmac.New(sha256.New, []byte(secret))
|
||||
mac.Write(payload)
|
||||
return hex.EncodeToString(mac.Sum(nil))
|
||||
}
|
||||
|
||||
// SignatureHeader ist der HTTP-Header, unter dem die Signatur uebertragen
|
||||
// wird — dokumentierter Vertrag fuer Empfaenger (Akzeptanzkriterium 3).
|
||||
const SignatureHeader = "X-Nexarch-Signature-256"
|
||||
@@ -1,205 +0,0 @@
|
||||
package webhook
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func setupTest(t *testing.T) (*Store, *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 webhook_subscriptions (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), event_type TEXT NOT NULL,
|
||||
target_url TEXT NOT NULL, secret TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS webhook_deliveries (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(), subscription_id UUID NOT NULL REFERENCES webhook_subscriptions(id),
|
||||
event_type TEXT NOT NULL, payload JSONB NOT NULL, status TEXT NOT NULL DEFAULT 'pending',
|
||||
attempt INT NOT NULL DEFAULT 0, next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), delivered_at TIMESTAMPTZ
|
||||
);
|
||||
`); err != nil {
|
||||
t.Fatalf("schema: %v", err)
|
||||
}
|
||||
|
||||
cleanup := func() { pool.Close() }
|
||||
return NewStore(pool), pool, cleanup
|
||||
}
|
||||
|
||||
func uniqueEventType(prefix string) string {
|
||||
return prefix + "-" + time.Now().Format("150405.000000000")
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 1 + Pruefung 1: ein Modul reicht ein Ereignis ein
|
||||
// (Enqueue) ohne eigene Zustellungslogik, der zentrale Dispatcher liefert
|
||||
// zuverlaessig aus.
|
||||
func TestEnqueueAndProcessDue_DeliversSuccessfully(t *testing.T) {
|
||||
store, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
var receivedBody []byte
|
||||
var receivedSignature string
|
||||
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
receivedBody = body
|
||||
receivedSignature = r.Header.Get(SignatureHeader)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
}))
|
||||
defer target.Close()
|
||||
|
||||
eventType := uniqueEventType("dms.file.created")
|
||||
sub, err := store.Subscribe(ctx, eventType, target.URL, "geheimes-secret")
|
||||
if err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
|
||||
n, err := store.Enqueue(ctx, eventType, map[string]string{"file_id": "42"})
|
||||
if err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
if n != 1 {
|
||||
t.Fatalf("erwartet 1 eingereihte zustellung, habe %d", n)
|
||||
}
|
||||
|
||||
dispatcher := NewDispatcher(pool, target.Client())
|
||||
processed, err := dispatcher.ProcessDue(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("processdue: %v", err)
|
||||
}
|
||||
if processed != 1 {
|
||||
t.Fatalf("erwartet 1 verarbeitete zustellung, habe %d", processed)
|
||||
}
|
||||
|
||||
var status string
|
||||
if err := pool.QueryRow(ctx, `SELECT status FROM webhook_deliveries WHERE subscription_id = $1`, sub.ID).Scan(&status); err != nil {
|
||||
t.Fatalf("status lesen: %v", err)
|
||||
}
|
||||
if status != StatusDelivered {
|
||||
t.Fatalf("status = %q, want %q", status, StatusDelivered)
|
||||
}
|
||||
|
||||
// Postgres' JSONB-Spalte kann die Byte-Repraesentation des Payloads
|
||||
// gegenueber dem urspruenglichen json.Marshal kanonisieren (z.B.
|
||||
// Leerzeichen) — das ist unschaedlich, denn der Dispatcher signiert
|
||||
// IMMER exakt die Bytes, die er auch sendet. Die Pruefung vergleicht
|
||||
// deshalb Signatur gegen tatsaechlich empfangene Bytes (Selbstkonsistenz),
|
||||
// nicht gegen eine unabhaengig neu marshalte Referenz.
|
||||
if !VerifySignature("geheimes-secret", receivedBody, receivedSignature) {
|
||||
t.Fatalf("empfangene signatur %q passt nicht zum empfangenen payload %q", receivedSignature, receivedBody)
|
||||
}
|
||||
var decoded map[string]string
|
||||
if err := json.Unmarshal(receivedBody, &decoded); err != nil {
|
||||
t.Fatalf("empfangener payload nicht als json lesbar: %v", err)
|
||||
}
|
||||
if decoded["file_id"] != "42" {
|
||||
t.Fatalf("empfangener payload = %v, want file_id=42", decoded)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 2 + Pruefung 2: fehlschlagendes Ziel loest Retry mit
|
||||
// wachsendem Backoff aus und endet nach der konfigurierten Obergrenze in
|
||||
// "failed".
|
||||
func TestProcessDue_RetriesWithBackoffThenMarksFailed(t *testing.T) {
|
||||
store, pool, cleanup := setupTest(t)
|
||||
defer cleanup()
|
||||
ctx := context.Background()
|
||||
|
||||
var callCount int32
|
||||
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
atomic.AddInt32(&callCount, 1)
|
||||
w.WriteHeader(http.StatusInternalServerError)
|
||||
}))
|
||||
defer target.Close()
|
||||
|
||||
eventType := uniqueEventType("mail.send.failed")
|
||||
sub, err := store.Subscribe(ctx, eventType, target.URL, "secret")
|
||||
if err != nil {
|
||||
t.Fatalf("subscribe: %v", err)
|
||||
}
|
||||
if _, err := store.Enqueue(ctx, eventType, map[string]string{"x": "y"}); err != nil {
|
||||
t.Fatalf("enqueue: %v", err)
|
||||
}
|
||||
|
||||
dispatcher := NewDispatcher(pool, target.Client())
|
||||
dispatcher.MaxAttempts = 2
|
||||
dispatcher.BaseBackoff = 30 * time.Millisecond
|
||||
|
||||
// 1. Versuch: schlaegt fehl, ist aber noch nicht die Obergrenze.
|
||||
if _, err := dispatcher.ProcessDue(ctx); err != nil {
|
||||
t.Fatalf("processdue 1: %v", err)
|
||||
}
|
||||
var status string
|
||||
var attempt int
|
||||
if err := pool.QueryRow(ctx, `SELECT status, attempt FROM webhook_deliveries WHERE subscription_id = $1`, sub.ID).Scan(&status, &attempt); err != nil {
|
||||
t.Fatalf("status lesen 1: %v", err)
|
||||
}
|
||||
if status != StatusPending || attempt != 1 {
|
||||
t.Fatalf("nach 1. fehlschlag: status=%q attempt=%d, want pending/1", status, attempt)
|
||||
}
|
||||
|
||||
// Sofort erneut verarbeiten: Backoff ist noch nicht abgelaufen -> nichts faellig.
|
||||
processedTooEarly, err := dispatcher.ProcessDue(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("processdue (zu frueh): %v", err)
|
||||
}
|
||||
if processedTooEarly != 0 {
|
||||
t.Fatal("erwartet 0 verarbeitete zustellungen, solange backoff nicht abgelaufen ist")
|
||||
}
|
||||
|
||||
time.Sleep(dispatcher.backoffFor(1) + 20*time.Millisecond)
|
||||
|
||||
// 2. Versuch: erreicht MaxAttempts=2 -> endgueltig fehlgeschlagen.
|
||||
if _, err := dispatcher.ProcessDue(ctx); err != nil {
|
||||
t.Fatalf("processdue 2: %v", err)
|
||||
}
|
||||
if err := pool.QueryRow(ctx, `SELECT status, attempt FROM webhook_deliveries WHERE subscription_id = $1`, sub.ID).Scan(&status, &attempt); err != nil {
|
||||
t.Fatalf("status lesen 2: %v", err)
|
||||
}
|
||||
if status != StatusFailed || attempt != 2 {
|
||||
t.Fatalf("nach 2. fehlschlag: status=%q attempt=%d, want failed/2", status, attempt)
|
||||
}
|
||||
if atomic.LoadInt32(&callCount) != 2 {
|
||||
t.Fatalf("erwartet genau 2 tatsaechliche zustellversuche, habe %d", callCount)
|
||||
}
|
||||
}
|
||||
|
||||
// Akzeptanzkriterium 3 + Pruefung 3: Signaturpruefung erkennt eine
|
||||
// manipulierte Payload zuverlaessig.
|
||||
func TestVerifySignature_DetectsTamperedPayload(t *testing.T) {
|
||||
secret := "geteiltes-geheimnis"
|
||||
payload := []byte(`{"file_id":"42"}`)
|
||||
signature := Sign(secret, payload)
|
||||
|
||||
if !VerifySignature(secret, payload, signature) {
|
||||
t.Fatal("erwartet gueltige signatur fuer unveraenderte payload")
|
||||
}
|
||||
|
||||
tampered := []byte(`{"file_id":"99"}`)
|
||||
if VerifySignature(secret, tampered, signature) {
|
||||
t.Fatal("erwartet ungueltige signatur fuer manipulierte payload")
|
||||
}
|
||||
|
||||
if VerifySignature("falsches-secret", payload, signature) {
|
||||
t.Fatal("erwartet ungueltige signatur bei falschem secret")
|
||||
}
|
||||
}
|
||||
@@ -1,2 +0,0 @@
|
||||
DROP TABLE webhook_deliveries;
|
||||
DROP TABLE webhook_subscriptions;
|
||||
@@ -1,27 +0,0 @@
|
||||
-- Zentrale Webhook-Registry & Zustellung (API-07, siehe
|
||||
-- core-kanban/tickets/API-07.md) — Postgres-basierte Jobqueue, kein
|
||||
-- Redis/AMQP (Ticket-Vorgabe).
|
||||
CREATE TABLE webhook_subscriptions (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
event_type TEXT NOT NULL,
|
||||
target_url TEXT NOT NULL,
|
||||
secret TEXT NOT NULL,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX webhook_subscriptions_event_type_idx ON webhook_subscriptions (event_type);
|
||||
|
||||
CREATE TABLE webhook_deliveries (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
subscription_id UUID NOT NULL REFERENCES webhook_subscriptions(id),
|
||||
event_type TEXT NOT NULL,
|
||||
payload JSONB NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'pending', -- pending|delivered|failed
|
||||
attempt INT NOT NULL DEFAULT 0,
|
||||
next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
last_error TEXT,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
delivered_at TIMESTAMPTZ
|
||||
);
|
||||
|
||||
CREATE INDEX webhook_deliveries_due_idx ON webhook_deliveries (status, next_attempt_at);
|
||||
Reference in New Issue
Block a user