nexarch/connect/internal/webhooksub/webhooksub.go

// 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
}