API-03: 03 externe webhook abonnements nutzt core api 07

Connect-seitige Webhook-Abonnements ueber Core API-07 (zentrale Webhook-Registry): connect/internal/webhookclient als schlanker HTTP-Client gegen die Registry (Subscribe/Unsubscribe), connect/internal/ webhooksub mit lokaler, verschluesselter Abonnementverwaltung und HTTP-Zustellempfaenger inkl. HMAC-Signaturpruefung. Core API-07 exponiert bislang keinen HTTP-Wrapper fuer physisch getrennte Module (anders als z.B. AUD-08) -- Tests laufen deshalb gegen einen den dokumentierten Core-Vertrag nachbildenden Stub, echte Postgres-Testdatenbank, echte HTTP-Verbindungen. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-10-16 13:18:10 +00:00
1 parent dac7fff504
commit 0fef62b2a3
@@ -0,0 +1,80 @@
+# API-03 – Prüfprotokoll: Externe Webhook-Abonnements (nutzt Core API-07)
+
+Datum: 2026-10-16. Host: 204 (Wegwerf-Postgres, `TEST_TENANT_DSN`, isoliertes Schema pro Testlauf über `connect/internal/testdb`).
+
+## Ausgangslage: Schnittstelle zu Core API-07
+
+Core API-07 (`internal/webhook`, Modul `gitea.perlbach24.de/scripte/nexarch`) implementiert Registrierung (`Store.Subscribe`), Zustellung mit Retry/Backoff (`Dispatcher`) und HMAC-Signatur (`Sign`/`VerifySignature`) als reines Go-Paket. Recherche im Core-Modul (gleicher Checkout, `internal/webhook/*.go`, `internal/contracttest/contracttest_test.go`) ergab: **es existiert aktuell kein HTTP-Handler/`cmd/*`-Dienst, der `Store.Subscribe` für physisch getrennte Module (Connect, DMS, Mail, Archive) über HTTP erreichbar macht** — anders als z. B. AUD-08 (`internal/audit/ingest_handler.go` + `cmd/auditlog-devserver`), das exakt dieses Muster für den Audit-Log bereits umsetzt. Core-seitig fehlt zusätzlich eine `Unsubscribe`-Methode auf `Store` völlig.
+
+Das ist eine Lücke auf Core-Seite (core-kanban), nicht Gegenstand dieser Kachel (Connect ist laut Auftrag physisch getrennt und darf Cores `internal/`-Pakete nicht importieren — nur über HTTP koppeln). Diese Kachel erlaubt für genau diesen Fall ausdrücklich einen **„minimal gestubten Core-Webhook-Registry-Endpunkt"** für den End-zu-Ende-Test. Umgesetzt wurde daher:
+
+- Der HTTP-Vertrag, den ein künftiger Core-HTTP-Wrapper für API-07 erfüllen müsste (`POST /webhooks/subscriptions`, `DELETE /webhooks/subscriptions/{id}`, `X-Service-Token`-Auth, Felder `tenant_slug`/`event_type`/`target_url`/`secret`), dokumentiert als Vertrag in `connect/internal/webhookclient`.
+- Alle Tests laufen über einen echten `httptest.Server`, der genau diesen Vertrag nachbildet (echte HTTP-Verbindung über Loopback-TCP, keine Mock-Schnittstelle des Clients selbst) — dasselbe Muster wie `connect/internal/policyclient`/`connect/internal/auditclient` bereits für andere Core-Schnittstellen verwenden.
+- HMAC-Signaturvertrag (`X-Nexarch-Signature-256`, hex-kodiertes HMAC-SHA256) 1:1 aus `internal/webhook.SignatureHeader`/`VerifySignature` übernommen, da Connect das Core-Paket nicht importieren kann (physische Modultrennung).
+
+## Umsetzung
+
+- `connect/internal/webhookclient` — schlanker HTTP-Client gegen Core API-07 (`Subscribe`, `Unsubscribe`), Muster aus `connect/internal/auditclient`/`dms/internal/retentionclient`.
+- `connect/internal/webhooksub` — Connect-seitige Verwaltung:
+  - `Manager.RegisterAction(connectorName, eventType, action)`: verdrahtet prozesslokal, welche Connector-Aktion ein Ereignis auslöst (Muster wie `registry.Registry`).
+  - `Manager.Subscribe`: registriert bei Core (AK1), verschlüsselt das Zustellgeheimnis per `connect/internal/crypto` (Envelope-Verschlüsselung, gleiches Muster wie `credential.Store`) und speichert erst danach lokal. Schlägt die Core-Registrierung fehl, bleibt nichts lokal zurück.
+  - `Manager.Unsubscribe`: bestellt zuerst bei Core ab, löscht die lokale Zeile erst danach vollständig (AK3). Schlägt die Core-Abbestellung fehl, bleibt die lokale Zeile für einen Wiederholungsversuch erhalten statt ein verwaistes Core-Abonnement zu hinterlassen.
+  - `Manager.RegisterRoutes`/`handleDelivery` (`delivery.go`): HTTP-Endpunkt `POST /connector-webhooks/deliveries/{id}`, an den Core laut `target_url` zustellt. Prüft die HMAC-Signatur zeitkonstant (`crypto/subtle.ConstantTimeCompare`) gegen das entschlüsselte Geheimnis, löst bei Erfolg die registrierte Connector-Aktion aus (AK2).
+- Migration `0013_connector_webhook_subscriptions` — Tabelle `connector_webhook_subscriptions` (lokale id = Zustellpfad-Segment, `core_subscription_id`, verschlüsseltes Geheimnis, `UNIQUE(tenant_slug, connector_name, event_type)`).
+- `connect/scripts/check-security-test-coverage.sh` — `internal/webhookclient`/`internal/webhooksub` ergänzt (beide verarbeiten tenant-gebundene Geheimnisse/Core-Authentifizierung, gleiche Kategorie wie `credential`/`auditclient`).
+
+## Akzeptanzkriterien
+
+1. **Connector kann ein Webhook-Ereignis über die Core-Webhook-Registry abonnieren** — erfüllt: `Manager.Subscribe` ruft `webhookclient.Subscribe` (POST an die Registry) auf und verlangt vorab eine registrierte Connector-Aktion.
+2. **Zugestelltes Ereignis löst die zugeordnete Connector-Aktion aus** — erfüllt: `handleDelivery` prüft Signatur und ruft die per `RegisterAction` hinterlegte Funktion mit dem zugestellten Payload auf.
+3. **Abbestellen entfernt das Abonnement vollständig** — erfüllt: `Manager.Unsubscribe` entfernt sowohl bei Core (`webhookclient.Unsubscribe`) als auch die lokale Zeile; eine erneute Zustellung an die alte URL schlägt danach mit 404 fehl.
+
+## Prüfungen vor der Abnahme
+
+| # | Prüfung | Ergebnis |
+|---|---|---|
+| 1 | Testabonnement über Core-Webhook-Registry im Test erfolgreich registriert | **Bestanden.** `TestSubscribe_RegistersAtCoreAndPersistsLocally`: echter HTTP-POST an den (gestubten) Core-Endpunkt, lokale Zeile mit `core_subscription_id` entsteht, Core-Server bestätigt das Abonnement. Negativ abgesichert durch `TestSubscribe_WithoutRegisteredAction_Fails` und `TestSubscribe_CoreFailure_PersistsNothingLocally` (kein verwaister lokaler Datensatz bei Core-Ausfall). |
+| 2 | Testereignis löst im Test die erwartete Connector-Aktion aus | **Bestanden.** `TestHandleDelivery_ValidSignature_TriggersConnectorAction`: echter HTTP-POST (eigener `httptest.Server`, echte TCP-Verbindung) mit gültiger HMAC-Signatur an die beim Subscribe erzeugte Zustell-URL löst die registrierte Aktion mit dem exakten Payload aus. Negativ abgesichert durch `TestHandleDelivery_InvalidSignature_Rejected` (401, Aktion bleibt aus) und `TestHandleDelivery_UnknownSubscription_NotFound`. |
+| 3 | Abbestellung im Test vollständig entfernt | **Bestanden.** `TestUnsubscribe_RemovesAtCoreAndLocally_Completely`: nach `Unsubscribe` ist die lokale Zeile weg und der Core-Stub bestätigt das entfernte Abonnement. Zusätzlich `TestHandleDelivery_AfterUnsubscribe_NotFound` (Zustellung an die alte URL schlägt danach mit 404 fehl) und `TestUnsubscribe_CoreFailure_KeepsLocalRowForRetry` (kein Datenverlust bei Core-Ausfall während der Abbestellung). |
+
+## Testlauf (real, Host 204, gegen echte Postgres-Testdatenbank)
+
+```
+go build ./...                                                   → keine Ausgabe (sauber)
+go vet ./...                                                      → keine Ausgabe (sauber)
+gofmt -l internal/webhookclient internal/webhooksub                → keine Ausgabe (bereits formatiert)
+
+go test ./internal/webhookclient/... -v
+  PASS  TestSubscribe_SendsRequestFields_ReturnsSubscription
+  PASS  TestSubscribe_UnexpectedStatus_ReturnsError
+  PASS  TestSubscribe_TransportError_ReturnsError
+  PASS  TestUnsubscribe_SendsServiceTokenAndTenant_Succeeds
+  PASS  TestUnsubscribe_UnexpectedStatus_ReturnsError
+  PASS  TestUnsubscribe_TransportError_ReturnsError
+ok    connect/internal/webhookclient   0.007s
+
+go test ./internal/webhooksub/... -v   (TEST_TENANT_DSN gesetzt, echte Postgres-DB)
+  PASS  TestHandleDelivery_ValidSignature_TriggersConnectorAction
+  PASS  TestHandleDelivery_InvalidSignature_Rejected
+  PASS  TestHandleDelivery_UnknownSubscription_NotFound
+  PASS  TestHandleDelivery_AfterUnsubscribe_NotFound
+  PASS  TestSubscribe_RegistersAtCoreAndPersistsLocally
+  PASS  TestSubscribe_WithoutRegisteredAction_Fails
+  PASS  TestSubscribe_CoreFailure_PersistsNothingLocally
+  PASS  TestUnsubscribe_RemovesAtCoreAndLocally_Completely
+  PASS  TestUnsubscribe_UnknownSubscription_ReturnsNotFound
+  PASS  TestUnsubscribe_CoreFailure_KeepsLocalRowForRetry
+ok    connect/internal/webhooksub      1.572s–1.778s (mehrere Läufe)
+
+go test ./... (gesamtes connect-Modul, TEST_TENANT_DSN gesetzt) → alle Pakete ok, kein FAIL
+
+bash scripts/check-security-test-coverage.sh → "Merge-Gate bestanden: alle sicherheitsrelevanten Pakete haben Tests."
+```
+
+Keine Regressionen: alle zuvor bestehenden Pakete (u. a. `credential`, `statusapi`, `syncjob`, `audittrail`) bleiben unverändert grün.
+
+## Nicht Bestandteil dieser Kachel
+
+- `QA-02` Prüfgate Connector-Engine — separates Folgeticket, nicht umgesetzt.
+- Core-seitiger HTTP-Wrapper für API-07 (`Store.Subscribe`/neue `Unsubscribe`-Methode über `cmd/*-devserver`, analog AUD-08) — Lücke auf Core-Seite, gehört ins core-kanban-Board, nicht in diese Kachel. Bis dahin bindet `connect/internal/webhookclient` an den hier dokumentierten, noch nicht Core-seitig existierenden Vertrag; Umstellung auf die reale Core-Basis-URL ist eine reine Konfigurationsänderung (`NEXARCH_*_BASE_URL`), keine Codeänderung.
+- Verdrahtung in `connect-app`s `main.go` (Produktions-`KEKProvider` gegen Core API-10) — bewusst unterlassen: auch `connect/internal/credential` (ältere, grundlegendere Kachel) ist dort noch nicht verdrahtet; ein produktiver `KEKProvider` für Connect existiert im Modul noch nirgends und wäre eine eigene, hier nicht beauftragte Erweiterung.
@@ -0,0 +1,108 @@
+// Package webhookclient ist ein schlanker HTTP-Client fuer Core API-07
+// (internal/webhook, zentrale Webhook-Registry). Connect ist ein physisch
+// getrenntes Go-Modul und kann Cores internal/webhook daher nicht direkt
+// importieren -- nur ueber dessen HTTP-Schnittstelle (gleiches Muster wie
+// connect/internal/auditclient fuer AUD-08 und dms/internal/retentionclient
+// fuer Archive RET-05).
+package webhookclient
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/url"
+	"time"
+
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/httpclient"
+)
+
+type Client struct {
+	BaseURL      string
+	ServiceToken string
+	HTTPClient   *http.Client
+}
+
+func New(baseURL, serviceToken string) *Client {
+	return &Client{BaseURL: baseURL, ServiceToken: serviceToken, HTTPClient: httpclient.Default()}
+}
+
+// Subscription spiegelt Core API-07s Antwort auf eine Registrierung.
+type Subscription struct {
+	ID        string    `json:"id"`
+	EventType string    `json:"event_type"`
+	TargetURL string    `json:"target_url"`
+	CreatedAt time.Time `json:"created_at"`
+}
+
+type subscribeRequest struct {
+	TenantSlug string `json:"tenant_slug"`
+	EventType  string `json:"event_type"`
+	TargetURL  string `json:"target_url"`
+	Secret     string `json:"secret"`
+}
+
+// Subscribe registriert targetURL bei Core API-07 fuer eventType. Jeder
+// Fehler (Transport, Timeout, Nicht-2xx) wird als Fehler zurueckgegeben --
+// der Aufrufer entscheidet, ob/wie er das nachbehandelt (Akzeptanzkriterium
+// 1).
+func (c *Client) Subscribe(ctx context.Context, tenantSlug, eventType, targetURL, secret string) (Subscription, error) {
+	body, err := json.Marshal(subscribeRequest{
+		TenantSlug: tenantSlug, EventType: eventType, TargetURL: targetURL, Secret: secret,
+	})
+	if err != nil {
+		return Subscription{}, fmt.Errorf("webhookclient: request kodieren: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+"/webhooks/subscriptions", bytes.NewReader(body))
+	if err != nil {
+		return Subscription{}, fmt.Errorf("webhookclient: request bauen: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-Service-Token", c.ServiceToken)
+
+	resp, err := c.httpClient().Do(req)
+	if err != nil {
+		return Subscription{}, fmt.Errorf("webhookclient: aufruf fehlgeschlagen: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusCreated {
+		return Subscription{}, fmt.Errorf("webhookclient: unerwarteter status %d", resp.StatusCode)
+	}
+	var out Subscription
+	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+		return Subscription{}, fmt.Errorf("webhookclient: antwort dekodieren: %w", err)
+	}
+	return out, nil
+}
+
+// Unsubscribe entfernt subscriptionID vollstaendig aus Core API-07
+// (Akzeptanzkriterium 3).
+func (c *Client) Unsubscribe(ctx context.Context, tenantSlug, subscriptionID string) error {
+	target := c.BaseURL + "/webhooks/subscriptions/" + url.PathEscape(subscriptionID) +
+		"?tenant_slug=" + url.QueryEscape(tenantSlug)
+	req, err := http.NewRequestWithContext(ctx, http.MethodDelete, target, nil)
+	if err != nil {
+		return fmt.Errorf("webhookclient: request bauen: %w", err)
+	}
+	req.Header.Set("X-Service-Token", c.ServiceToken)
+
+	resp, err := c.httpClient().Do(req)
+	if err != nil {
+		return fmt.Errorf("webhookclient: aufruf fehlgeschlagen: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+
+	if resp.StatusCode != http.StatusNoContent {
+		return fmt.Errorf("webhookclient: unerwarteter status %d", resp.StatusCode)
+	}
+	return nil
+}
+
+func (c *Client) httpClient() *http.Client {
+	if c.HTTPClient != nil {
+		return c.HTTPClient
+	}
+	return httpclient.Default()
+}
@@ -0,0 +1,109 @@
+package webhookclient
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+// newFakeCoreWebhookAPI steht fuer Core API-07s echten Vertrag (POST
+// /webhooks/subscriptions, DELETE /webhooks/subscriptions/{id}) -- testet
+// die echte HTTP-Kette (Client -> echter Server -> Antwort-Dekodierung),
+// nicht nur eine wirkungslose lokale Struktur.
+func newFakeCoreWebhookAPI(t *testing.T) (*httptest.Server, *[]string) {
+	t.Helper()
+	var calls []string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls = append(calls, r.Method+" "+r.URL.String())
+		if r.Header.Get("X-Service-Token") == "" {
+			t.Fatalf("Service-Token fehlt bei %s %s", r.Method, r.URL.Path)
+		}
+		switch {
+		case r.Method == http.MethodPost && r.URL.Path == "/webhooks/subscriptions":
+			var in subscribeRequest
+			if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
+				t.Fatalf("body dekodieren: %v", err)
+			}
+			w.WriteHeader(http.StatusCreated)
+			_ = json.NewEncoder(w).Encode(Subscription{
+				ID: "core-sub-1", EventType: in.EventType, TargetURL: in.TargetURL, CreatedAt: time.Now().UTC(),
+			})
+		case r.Method == http.MethodDelete:
+			w.WriteHeader(http.StatusNoContent)
+		default:
+			w.WriteHeader(http.StatusNotFound)
+		}
+	}))
+	return srv, &calls
+}
+
+func TestSubscribe_SendsRequestFields_ReturnsSubscription(t *testing.T) {
+	srv, _ := newFakeCoreWebhookAPI(t)
+	defer srv.Close()
+
+	c := New(srv.URL, "svc-token-123")
+	sub, err := c.Subscribe(context.Background(), "acme", "dms.document.archived", "https://connect.example/d/abc", "shh-secret")
+	if err != nil {
+		t.Fatalf("Subscribe: %v", err)
+	}
+	if sub.ID != "core-sub-1" {
+		t.Fatalf("erwartete core-vergebene id, bekam %q", sub.ID)
+	}
+	if sub.EventType != "dms.document.archived" {
+		t.Fatalf("event_type nicht wie gesendet: %q", sub.EventType)
+	}
+}
+
+func TestSubscribe_UnexpectedStatus_ReturnsError(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	c := New(srv.URL, "svc-token-123")
+	if _, err := c.Subscribe(context.Background(), "acme", "dms.document.archived", "https://connect.example/d/abc", "shh"); err == nil {
+		t.Fatal("erwarteter Fehler bei Status 500 blieb aus")
+	}
+}
+
+func TestSubscribe_TransportError_ReturnsError(t *testing.T) {
+	c := New("http://127.0.0.1:1", "svc-token-123")
+	if _, err := c.Subscribe(context.Background(), "acme", "dms.document.archived", "https://connect.example/d/abc", "shh"); err == nil {
+		t.Fatal("erwarteter Transport-Fehler blieb aus")
+	}
+}
+
+func TestUnsubscribe_SendsServiceTokenAndTenant_Succeeds(t *testing.T) {
+	srv, calls := newFakeCoreWebhookAPI(t)
+	defer srv.Close()
+
+	c := New(srv.URL, "svc-token-123")
+	if err := c.Unsubscribe(context.Background(), "acme", "core-sub-1"); err != nil {
+		t.Fatalf("Unsubscribe: %v", err)
+	}
+	if len(*calls) != 1 || (*calls)[0] != "DELETE /webhooks/subscriptions/core-sub-1?tenant_slug=acme" {
+		t.Fatalf("unerwartete aufrufe: %v", *calls)
+	}
+}
+
+func TestUnsubscribe_UnexpectedStatus_ReturnsError(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer srv.Close()
+
+	c := New(srv.URL, "svc-token-123")
+	if err := c.Unsubscribe(context.Background(), "acme", "core-sub-1"); err == nil {
+		t.Fatal("erwarteter Fehler bei Status 500 blieb aus")
+	}
+}
+
+func TestUnsubscribe_TransportError_ReturnsError(t *testing.T) {
+	c := New("http://127.0.0.1:1", "svc-token-123")
+	if err := c.Unsubscribe(context.Background(), "acme", "core-sub-1"); err == nil {
+		t.Fatal("erwarteter Transport-Fehler blieb aus")
+	}
+}
@@ -0,0 +1,153 @@
+package webhooksub_test
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	connectcrypto "gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/migrate"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/testdb"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/webhookclient"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/webhooksub"
+)
+
+type fakeKEK struct{ kek []byte }
+
+func (f *fakeKEK) TenantKEK(ctx context.Context, tenantSlug string) ([]byte, error) {
+	return f.kek, nil
+}
+
+// fakeCoreWebhookAPI reproduziert Core API-07s Subscribe/Unsubscribe-Vertrag
+// -- der von dieser Kachel ausdruecklich erlaubte "minimal gestubte
+// Core-Webhook-Registry-Endpunkt" fuer den echten End-zu-Ende-Test (echte
+// HTTP-Verbindung, echte Datenbank, kein Mock der Client-Logik selbst).
+type fakeCoreWebhookAPI struct {
+	mu            sync.Mutex
+	subscriptions map[string]bool
+	secrets       map[string]string
+	nextID        int
+}
+
+func newFakeCoreWebhookAPI() *fakeCoreWebhookAPI {
+	return &fakeCoreWebhookAPI{subscriptions: make(map[string]bool), secrets: make(map[string]string)}
+}
+
+func (f *fakeCoreWebhookAPI) hasSubscription(id string) bool {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.subscriptions[id]
+}
+
+// secretFor liefert das Geheimnis, das beim Subscribe fuer coreSubID
+// uebergeben wurde -- in echt kennt nur Core dieses Geheimnis und signiert
+// damit jede Zustellung; der Test braucht es, um eine echte, gueltige
+// Core-Zustellung nachzubilden.
+func (f *fakeCoreWebhookAPI) secretFor(coreSubID string) string {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.secrets[coreSubID]
+}
+
+func (f *fakeCoreWebhookAPI) handle(w http.ResponseWriter, r *http.Request) {
+	if r.Header.Get("X-Service-Token") == "" {
+		http.Error(w, "kein service-token", http.StatusUnauthorized)
+		return
+	}
+	switch {
+	case r.Method == http.MethodPost && r.URL.Path == "/webhooks/subscriptions":
+		var in struct {
+			TenantSlug string `json:"tenant_slug"`
+			EventType  string `json:"event_type"`
+			TargetURL  string `json:"target_url"`
+			Secret     string `json:"secret"`
+		}
+		if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
+			http.Error(w, "ungueltiger body", http.StatusBadRequest)
+			return
+		}
+		f.mu.Lock()
+		f.nextID++
+		id := fmt.Sprintf("core-sub-%d", f.nextID)
+		f.subscriptions[id] = true
+		f.secrets[id] = in.Secret
+		f.mu.Unlock()
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusCreated)
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"id": id, "event_type": in.EventType, "target_url": in.TargetURL, "created_at": time.Now().UTC(),
+		})
+	case r.Method == http.MethodDelete:
+		id := strings.TrimPrefix(r.URL.Path, "/webhooks/subscriptions/")
+		f.mu.Lock()
+		_, existed := f.subscriptions[id]
+		delete(f.subscriptions, id)
+		f.mu.Unlock()
+		if !existed {
+			http.Error(w, "unbekanntes abonnement", http.StatusNotFound)
+			return
+		}
+		w.WriteHeader(http.StatusNoContent)
+	default:
+		http.Error(w, "nicht unterstuetzt", http.StatusNotFound)
+	}
+}
+
+// testEnv buendelt einen Manager gegen eine echte, isolierte Postgres-
+// Testdatenbank und einen echten (gestubten) Core-Webhook-Server.
+type testEnv struct {
+	Manager *webhooksub.Manager
+	Pool    *pgxpool.Pool
+	Core    *fakeCoreWebhookAPI
+}
+
+func setup(t *testing.T) testEnv {
+	t.Helper()
+	dsn := os.Getenv("TEST_TENANT_DSN")
+	if dsn == "" {
+		t.Skip("TEST_TENANT_DSN nicht gesetzt, Integrationstest uebersprungen")
+	}
+	ctx := context.Background()
+	pool := testdb.Pool(t, dsn)
+
+	migrations, err := migrate.Load(findMigrationsDir(t))
+	if err != nil {
+		t.Fatalf("migrationen laden: %v", err)
+	}
+	if _, err := migrate.Up(ctx, pool, migrations); err != nil {
+		t.Fatalf("migrationen anwenden: %v", err)
+	}
+
+	core := newFakeCoreWebhookAPI()
+	coreSrv := httptest.NewServer(http.HandlerFunc(core.handle))
+	t.Cleanup(coreSrv.Close)
+
+	cryptoSvc := connectcrypto.NewService(&fakeKEK{kek: make([]byte, connectcrypto.KEKSize)})
+	client := webhookclient.New(coreSrv.URL, "svc-token-test")
+	mgr := webhooksub.NewManager(pool, cryptoSvc, client, "https://connect.example.test/connector-webhooks/deliveries")
+
+	return testEnv{Manager: mgr, Pool: pool, Core: core}
+}
+
+func findMigrationsDir(t *testing.T) string {
+	t.Helper()
+	dir, _ := os.Getwd()
+	for i := 0; i < 6; i++ {
+		p := dir + "/migrations"
+		if fi, err := os.Stat(p); err == nil && fi.IsDir() {
+			return p
+		}
+		dir += "/.."
+	}
+	t.Fatal("connect/migrations nicht gefunden")
+	return ""
+}
@@ -0,0 +1,92 @@
+package webhooksub
+
+import (
+	"crypto/hmac"
+	"crypto/sha256"
+	"crypto/subtle"
+	"encoding/hex"
+	"errors"
+	"io"
+	"net/http"
+
+	"github.com/jackc/pgx/v5"
+
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto"
+)
+
+// SignatureHeader ist Core API-07s dokumentierter Vertrag
+// (internal/webhook.SignatureHeader) -- hier erneut definiert, weil Connect
+// als physisch getrenntes Go-Modul Cores internal-Paket nicht importieren
+// kann.
+const SignatureHeader = "X-Nexarch-Signature-256"
+
+func verifySignature(secret string, payload []byte, signature string) bool {
+	mac := hmac.New(sha256.New, []byte(secret))
+	mac.Write(payload)
+	expected := mac.Sum(nil)
+	got, err := hex.DecodeString(signature)
+	if err != nil {
+		return false
+	}
+	return subtle.ConstantTimeCompare(expected, got) == 1
+}
+
+const maxDeliveryBody = 1 << 20
+
+// RegisterRoutes verdrahtet Core API-07s Zustellungs-Callback
+// (Akzeptanzkriterium 2): Core ruft diesen Pfad unter der bei Subscribe
+// uebergebenen target_url auf.
+func (m *Manager) RegisterRoutes(mux *http.ServeMux) {
+	mux.HandleFunc("POST /connector-webhooks/deliveries/{id}", m.handleDelivery)
+}
+
+func (m *Manager) handleDelivery(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+
+	var tenantSlug, connectorName, eventType string
+	var ciphertext, wrappedDEK []byte
+	err := m.pool.QueryRow(r.Context(), `
+		SELECT tenant_slug, connector_name, event_type, secret_ciphertext, secret_wrapped_dek
+		FROM connector_webhook_subscriptions WHERE id = $1
+	`, id).Scan(&tenantSlug, &connectorName, &eventType, &ciphertext, &wrappedDEK)
+	if errors.Is(err, pgx.ErrNoRows) {
+		http.Error(w, "unbekanntes abonnement", http.StatusNotFound)
+		return
+	}
+	if err != nil {
+		http.Error(w, "abonnement konnte nicht gelesen werden", http.StatusInternalServerError)
+		return
+	}
+
+	body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxDeliveryBody))
+	if err != nil {
+		http.Error(w, "payload zu gross oder nicht lesbar", http.StatusBadRequest)
+		return
+	}
+
+	secretPlain, err := m.crypto.Open(r.Context(), tenantSlug, &crypto.Sealed{Ciphertext: ciphertext, WrappedDEK: wrappedDEK})
+	if err != nil {
+		http.Error(w, "geheimnis nicht entschluesselbar", http.StatusInternalServerError)
+		return
+	}
+
+	if !verifySignature(string(secretPlain), body, r.Header.Get(SignatureHeader)) {
+		http.Error(w, "signatur ungueltig", http.StatusUnauthorized)
+		return
+	}
+
+	action, ok := m.action(connectorName, eventType)
+	if !ok {
+		// Abonnement existiert, aber im aktuellen Prozess (noch) keine
+		// Aktion registriert -- 503 statt 404, damit Cores Zustellungs-
+		// Retry (API-07 AC2) es erneut versucht statt endgueltig
+		// aufzugeben.
+		http.Error(w, "connector-aktion derzeit nicht verfuegbar", http.StatusServiceUnavailable)
+		return
+	}
+	if err := action(r.Context(), body); err != nil {
+		http.Error(w, "connector-aktion fehlgeschlagen", http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+}
@@ -0,0 +1,160 @@
+package webhooksub_test
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/hex"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/webhooksub"
+)
+
+func sign(secret string, payload []byte) string {
+	mac := hmac.New(sha256.New, []byte(secret))
+	mac.Write(payload)
+	return hex.EncodeToString(mac.Sum(nil))
+}
+
+// subscribeAndGetDeliveryURL abonniert ueber env.Manager und liefert die
+// echte Zustell-URL samt Core-Geheimnis zurueck, wie sie ein echter
+// Core-Zustellversuch verwenden wuerde.
+func subscribeAndGetDeliveryURL(t *testing.T, env testEnv, srv *httptest.Server, connectorName, eventType string) (deliveryURL, secret string) {
+	t.Helper()
+	reg, err := env.Manager.Subscribe(context.Background(), "acme", connectorName, eventType)
+	if err != nil {
+		t.Fatalf("Subscribe: %v", err)
+	}
+	var coreSubID string
+	if err := env.Pool.QueryRow(context.Background(), `
+		SELECT core_subscription_id FROM connector_webhook_subscriptions WHERE id = $1
+	`, reg.ID).Scan(&coreSubID); err != nil {
+		t.Fatalf("core_subscription_id lesen: %v", err)
+	}
+	return srv.URL + "/connector-webhooks/deliveries/" + reg.ID, env.Core.secretFor(coreSubID)
+}
+
+// Pruefung 2 / Akzeptanzkriterium 2: ein zugestelltes Ereignis loest die
+// zugeordnete Connector-Aktion aus -- echter HTTP-Roundtrip gegen den
+// echten Handler, echte Signaturpruefung.
+func TestHandleDelivery_ValidSignature_TriggersConnectorAction(t *testing.T) {
+	env := setup(t)
+
+	var mu sync.Mutex
+	var gotPayload []byte
+	env.Manager.RegisterAction("testcrm", "dms.document.archived", func(_ context.Context, payload []byte) error {
+		mu.Lock()
+		defer mu.Unlock()
+		gotPayload = append([]byte(nil), payload...)
+		return nil
+	})
+
+	mux := http.NewServeMux()
+	env.Manager.RegisterRoutes(mux)
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	deliveryURL, secret := subscribeAndGetDeliveryURL(t, env, srv, "testcrm", "dms.document.archived")
+
+	payload := []byte(`{"document_id":"doc-42","event":"archived"}`)
+	req, _ := http.NewRequest(http.MethodPost, deliveryURL, bytes.NewReader(payload))
+	req.Header.Set(webhooksub.SignatureHeader, sign(secret, payload))
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("zustellung: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("erwartet 200, bekam %d", resp.StatusCode)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	if string(gotPayload) != string(payload) {
+		t.Fatalf("connector-aktion erhielt falschen payload: %q", gotPayload)
+	}
+}
+
+func TestHandleDelivery_InvalidSignature_Rejected(t *testing.T) {
+	env := setup(t)
+
+	triggered := false
+	env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error {
+		triggered = true
+		return nil
+	})
+
+	mux := http.NewServeMux()
+	env.Manager.RegisterRoutes(mux)
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	deliveryURL, _ := subscribeAndGetDeliveryURL(t, env, srv, "testcrm", "dms.document.archived")
+
+	payload := []byte(`{"document_id":"doc-42"}`)
+	req, _ := http.NewRequest(http.MethodPost, deliveryURL, bytes.NewReader(payload))
+	req.Header.Set(webhooksub.SignatureHeader, sign("falsches-geheimnis", payload))
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("zustellung: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusUnauthorized {
+		t.Fatalf("erwartet 401 bei ungueltiger signatur, bekam %d", resp.StatusCode)
+	}
+	if triggered {
+		t.Fatal("connector-aktion darf bei ungueltiger signatur nicht ausgeloest werden")
+	}
+}
+
+func TestHandleDelivery_UnknownSubscription_NotFound(t *testing.T) {
+	env := setup(t)
+	mux := http.NewServeMux()
+	env.Manager.RegisterRoutes(mux)
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	resp, err := http.Post(srv.URL+"/connector-webhooks/deliveries/unbekannt", "application/json", bytes.NewReader([]byte("{}")))
+	if err != nil {
+		t.Fatalf("zustellung: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNotFound {
+		t.Fatalf("erwartet 404, bekam %d", resp.StatusCode)
+	}
+}
+
+// Nach erfolgreichem Abbestellen darf eine (verspaetete) Zustellung an die
+// alte URL nicht mehr ankommen -- bestaetigt die Vollstaendigkeit aus
+// Akzeptanzkriterium 3 auch aus Empfaengersicht.
+func TestHandleDelivery_AfterUnsubscribe_NotFound(t *testing.T) {
+	env := setup(t)
+	env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil })
+
+	mux := http.NewServeMux()
+	env.Manager.RegisterRoutes(mux)
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+
+	deliveryURL, secret := subscribeAndGetDeliveryURL(t, env, srv, "testcrm", "dms.document.archived")
+
+	if err := env.Manager.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err != nil {
+		t.Fatalf("Unsubscribe: %v", err)
+	}
+
+	payload := []byte(`{"document_id":"doc-42"}`)
+	req, _ := http.NewRequest(http.MethodPost, deliveryURL, bytes.NewReader(payload))
+	req.Header.Set(webhooksub.SignatureHeader, sign(secret, payload))
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("zustellung: %v", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusNotFound {
+		t.Fatalf("erwartet 404 nach abbestellung, bekam %d", resp.StatusCode)
+	}
+}
@@ -0,0 +1,157 @@
+// Package webhooksub laesst Connect-Connectoren Ereignisse ueber Core
+// API-07 (zentrale Webhook-Registry) abonnieren (API-03). Connect
+// registriert lediglich Abonnements und empfaengt Zustellungen -- es
+// betreibt selbst keine Zustellungs-Engine (Produkt-DNA dieser Kachel).
+package webhooksub
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/webhookclient"
+)
+
+var (
+	ErrActionNotRegistered = errors.New("webhooksub: keine connector-aktion fuer dieses ereignis registriert")
+	ErrNotFound            = errors.New("webhooksub: abonnement nicht gefunden")
+)
+
+// Action verarbeitet ein zugestelltes Ereignis fuer genau einen Connector
+// und Ereignistyp (Akzeptanzkriterium 2).
+type Action func(ctx context.Context, payload []byte) error
+
+// Registration ist Connects lokale Sicht auf ein bei Core registriertes
+// Abonnement.
+type Registration struct {
+	ID            string
+	TenantSlug    string
+	ConnectorName string
+	EventType     string
+	CreatedAt     time.Time
+}
+
+// Manager verwaltet Connect-seitige Webhook-Abonnements: er registriert sie
+// bei Core API-07, haelt die Zuordnung Ereignistyp -> Connector-Aktion im
+// Prozessspeicher (wie registry.Registry) und verifiziert/verteilt
+// zugestellte Ereignisse.
+type Manager struct {
+	pool            *pgxpool.Pool
+	crypto          *crypto.Service
+	core            *webhookclient.Client
+	callbackBaseURL string
+
+	mu      sync.RWMutex
+	actions map[string]Action
+}
+
+func NewManager(pool *pgxpool.Pool, cryptoSvc *crypto.Service, core *webhookclient.Client, callbackBaseURL string) *Manager {
+	return &Manager{
+		pool: pool, crypto: cryptoSvc, core: core,
+		callbackBaseURL: strings.TrimRight(callbackBaseURL, "/"),
+		actions:         make(map[string]Action),
+	}
+}
+
+func actionKey(connectorName, eventType string) string { return connectorName + "\x00" + eventType }
+
+// RegisterAction verdrahtet die Aktion eines Connectors fuer eventType, vor
+// jedem Subscribe/jeder Zustellung dafuer -- prozesslokal, wie
+// registry.Registry, bei jedem Prozessstart neu aufgebaut.
+func (m *Manager) RegisterAction(connectorName, eventType string, action Action) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.actions[actionKey(connectorName, eventType)] = action
+}
+
+func (m *Manager) action(connectorName, eventType string) (Action, bool) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	a, ok := m.actions[actionKey(connectorName, eventType)]
+	return a, ok
+}
+
+// Subscribe registriert connectorName fuer eventType bei Core API-07
+// (Akzeptanzkriterium 1). Es verlangt eine vorab per RegisterAction
+// hinterlegte Aktion -- ein Abonnement ohne Ziel waere nie zustellbar.
+// Schlaegt die Core-Registrierung fehl, wird nichts lokal gespeichert.
+func (m *Manager) Subscribe(ctx context.Context, tenantSlug, connectorName, eventType string) (Registration, error) {
+	if _, ok := m.action(connectorName, eventType); !ok {
+		return Registration{}, fmt.Errorf("%w: %s/%s", ErrActionNotRegistered, connectorName, eventType)
+	}
+	id, err := newToken(16)
+	if err != nil {
+		return Registration{}, fmt.Errorf("webhooksub: id generieren: %w", err)
+	}
+	secret, err := newToken(32)
+	if err != nil {
+		return Registration{}, fmt.Errorf("webhooksub: geheimnis generieren: %w", err)
+	}
+	targetURL := m.callbackBaseURL + "/" + id
+
+	coreSub, err := m.core.Subscribe(ctx, tenantSlug, eventType, targetURL, secret)
+	if err != nil {
+		return Registration{}, fmt.Errorf("webhooksub: core-registrierung: %w", err)
+	}
+
+	sealed, err := m.crypto.Seal(ctx, tenantSlug, []byte(secret))
+	if err != nil {
+		return Registration{}, fmt.Errorf("webhooksub: geheimnis versiegeln: %w", err)
+	}
+
+	var createdAt time.Time
+	err = m.pool.QueryRow(ctx, `
+		INSERT INTO connector_webhook_subscriptions
+			(id, tenant_slug, connector_name, event_type, core_subscription_id, secret_ciphertext, secret_wrapped_dek)
+		VALUES ($1, $2, $3, $4, $5, $6, $7)
+		RETURNING created_at
+	`, id, tenantSlug, connectorName, eventType, coreSub.ID, sealed.Ciphertext, sealed.WrappedDEK).Scan(&createdAt)
+	if err != nil {
+		return Registration{}, fmt.Errorf("webhooksub: abonnement speichern: %w", err)
+	}
+	return Registration{ID: id, TenantSlug: tenantSlug, ConnectorName: connectorName, EventType: eventType, CreatedAt: createdAt}, nil
+}
+
+// Unsubscribe entfernt connectorNames eventType-Abonnement vollstaendig
+// (Akzeptanzkriterium 3): erst bei Core, dann lokal. Schlaegt die
+// Core-Abbestellung fehl, bleibt die lokale Zeile fuer einen Wiederholungs-
+// versuch bestehen statt ein verwaistes Core-Abonnement zu hinterlassen.
+func (m *Manager) Unsubscribe(ctx context.Context, tenantSlug, connectorName, eventType string) error {
+	var localID, coreSubscriptionID string
+	err := m.pool.QueryRow(ctx, `
+		SELECT id, core_subscription_id FROM connector_webhook_subscriptions
+		WHERE tenant_slug = $1 AND connector_name = $2 AND event_type = $3
+	`, tenantSlug, connectorName, eventType).Scan(&localID, &coreSubscriptionID)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return ErrNotFound
+	}
+	if err != nil {
+		return fmt.Errorf("webhooksub: abonnement lesen: %w", err)
+	}
+
+	if err := m.core.Unsubscribe(ctx, tenantSlug, coreSubscriptionID); err != nil {
+		return fmt.Errorf("webhooksub: core-abbestellung: %w", err)
+	}
+
+	if _, err := m.pool.Exec(ctx, `DELETE FROM connector_webhook_subscriptions WHERE id = $1`, localID); err != nil {
+		return fmt.Errorf("webhooksub: abonnement loeschen: %w", err)
+	}
+	return nil
+}
+
+func newToken(numBytes int) (string, error) {
+	b := make([]byte, numBytes)
+	if _, err := rand.Read(b); err != nil {
+		return "", err
+	}
+	return hex.EncodeToString(b), nil
+}
@@ -0,0 +1,157 @@
+package webhooksub_test
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	connectcrypto "gitea.perlbach24.de/scripte/nexarch/connect/internal/crypto"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/webhookclient"
+	"gitea.perlbach24.de/scripte/nexarch/connect/internal/webhooksub"
+)
+
+// Pruefung 1 / Akzeptanzkriterium 1: Testabonnement wird ueber die
+// (gestubte) Core-Webhook-Registry registriert.
+func TestSubscribe_RegistersAtCoreAndPersistsLocally(t *testing.T) {
+	env := setup(t)
+	env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil })
+
+	reg, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived")
+	if err != nil {
+		t.Fatalf("Subscribe: %v", err)
+	}
+	if reg.ID == "" {
+		t.Fatal("erwartete lokale id")
+	}
+
+	var count int
+	var coreSubID string
+	if err := env.Pool.QueryRow(context.Background(), `
+		SELECT count(*), max(core_subscription_id) FROM connector_webhook_subscriptions WHERE id = $1
+	`, reg.ID).Scan(&count, &coreSubID); err != nil {
+		t.Fatalf("row lesen: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("erwartete genau eine zeile, habe %d", count)
+	}
+	if !env.Core.hasSubscription(coreSubID) {
+		t.Fatalf("core-abonnement %q nicht bei core registriert", coreSubID)
+	}
+}
+
+func TestSubscribe_WithoutRegisteredAction_Fails(t *testing.T) {
+	env := setup(t)
+
+	_, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived")
+	if !errors.Is(err, webhooksub.ErrActionNotRegistered) {
+		t.Fatalf("erwartete ErrActionNotRegistered, bekam %v", err)
+	}
+}
+
+// Negativtest: schlaegt die Core-Registrierung fehl, bleibt lokal nichts
+// zurueck (kein verwaistes Abonnement ohne Core-Gegenstueck).
+func TestSubscribe_CoreFailure_PersistsNothingLocally(t *testing.T) {
+	env := setup(t)
+	env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil })
+
+	failingCore := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer failingCore.Close()
+
+	cryptoSvc := connectcrypto.NewService(&fakeKEK{kek: make([]byte, connectcrypto.KEKSize)})
+	mgr := webhooksub.NewManager(env.Pool, cryptoSvc, webhookclient.New(failingCore.URL, "svc-token-test"), "https://connect.example.test/connector-webhooks/deliveries")
+	mgr.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil })
+
+	if _, err := mgr.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err == nil {
+		t.Fatal("erwarteter fehler bei core-ausfall blieb aus")
+	}
+
+	var count int
+	if err := env.Pool.QueryRow(context.Background(), `
+		SELECT count(*) FROM connector_webhook_subscriptions WHERE tenant_slug = 'acme' AND connector_name = 'testcrm'
+	`).Scan(&count); err != nil {
+		t.Fatalf("zeilen zaehlen: %v", err)
+	}
+	if count != 0 {
+		t.Fatalf("erwartete keine lokale zeile nach core-ausfall, habe %d", count)
+	}
+}
+
+// Pruefung 3 / Akzeptanzkriterium 3: Abbestellen entfernt das Abonnement
+// vollstaendig -- bei Core und lokal.
+func TestUnsubscribe_RemovesAtCoreAndLocally_Completely(t *testing.T) {
+	env := setup(t)
+	env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil })
+	reg, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived")
+	if err != nil {
+		t.Fatalf("Subscribe: %v", err)
+	}
+
+	var coreSubID string
+	if err := env.Pool.QueryRow(context.Background(), `
+		SELECT core_subscription_id FROM connector_webhook_subscriptions WHERE id = $1
+	`, reg.ID).Scan(&coreSubID); err != nil {
+		t.Fatalf("core_subscription_id lesen: %v", err)
+	}
+
+	if err := env.Manager.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err != nil {
+		t.Fatalf("Unsubscribe: %v", err)
+	}
+
+	var count int
+	if err := env.Pool.QueryRow(context.Background(), `
+		SELECT count(*) FROM connector_webhook_subscriptions WHERE id = $1
+	`, reg.ID).Scan(&count); err != nil {
+		t.Fatalf("zeilen zaehlen: %v", err)
+	}
+	if count != 0 {
+		t.Fatalf("erwartete vollstaendig entfernte lokale zeile, habe noch %d", count)
+	}
+	if env.Core.hasSubscription(coreSubID) {
+		t.Fatal("erwartete bei core entferntes abonnement, ist noch vorhanden")
+	}
+}
+
+func TestUnsubscribe_UnknownSubscription_ReturnsNotFound(t *testing.T) {
+	env := setup(t)
+	err := env.Manager.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived")
+	if !errors.Is(err, webhooksub.ErrNotFound) {
+		t.Fatalf("erwartete ErrNotFound, bekam %v", err)
+	}
+}
+
+// Negativtest: schlaegt die Core-Abbestellung fehl, bleibt die lokale Zeile
+// fuer einen Wiederholungsversuch bestehen statt ein verwaistes
+// Core-Abonnement zu hinterlassen.
+func TestUnsubscribe_CoreFailure_KeepsLocalRowForRetry(t *testing.T) {
+	env := setup(t)
+	env.Manager.RegisterAction("testcrm", "dms.document.archived", func(context.Context, []byte) error { return nil })
+	reg, err := env.Manager.Subscribe(context.Background(), "acme", "testcrm", "dms.document.archived")
+	if err != nil {
+		t.Fatalf("Subscribe: %v", err)
+	}
+
+	brokenCore := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer brokenCore.Close()
+	cryptoSvc := connectcrypto.NewService(&fakeKEK{kek: make([]byte, connectcrypto.KEKSize)})
+	mgrWithBrokenCore := webhooksub.NewManager(env.Pool, cryptoSvc, webhookclient.New(brokenCore.URL, "svc-token-test"), "https://connect.example.test/connector-webhooks/deliveries")
+
+	if err := mgrWithBrokenCore.Unsubscribe(context.Background(), "acme", "testcrm", "dms.document.archived"); err == nil {
+		t.Fatal("erwarteter fehler bei core-ausfall blieb aus")
+	}
+
+	var count int
+	if err := env.Pool.QueryRow(context.Background(), `
+		SELECT count(*) FROM connector_webhook_subscriptions WHERE id = $1
+	`, reg.ID).Scan(&count); err != nil {
+		t.Fatalf("zeilen zaehlen: %v", err)
+	}
+	if count != 1 {
+		t.Fatalf("erwartete erhaltene lokale zeile nach core-ausfall, habe %d", count)
+	}
+}
@@ -0,0 +1 @@
+DROP TABLE connector_webhook_subscriptions;
@@ -0,0 +1,16 @@
+-- API-03: lokale Abbildung eines Connect-Connector-Abonnements bei Core
+-- API-07 (zentrale Webhook-Registry). Der lokale id-Wert ist zugleich das
+-- Pfadsegment der an Core uebergebenen target_url, damit eine zugestellte
+-- Anfrage ohne Suche nach event_type+tenant dem richtigen Connector
+-- zugeordnet werden kann.
+CREATE TABLE connector_webhook_subscriptions (
+    id                   TEXT PRIMARY KEY,
+    tenant_slug          TEXT NOT NULL,
+    connector_name       TEXT NOT NULL,
+    event_type           TEXT NOT NULL,
+    core_subscription_id TEXT NOT NULL,
+    secret_ciphertext    BYTEA NOT NULL,
+    secret_wrapped_dek   BYTEA NOT NULL,
+    created_at           TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (tenant_slug, connector_name, event_type)
+);
@@ -25,6 +25,8 @@ SECURITY_PACKAGES=(
  "internal/m365calsync"
  "internal/syncjob"
  "internal/statusapi"
+  "internal/webhookclient"
+  "internal/webhooksub"
 )

 cd "$(dirname "$0")/.."
				`@@ -0,0 +1 @@`
				`DROP TABLE connector_webhook_subscriptions;`