member-console/internal/integrations/discourse/web/webhook.go

// Package web holds the Discourse integration's HTTP handlers: the webhook
// endpoint (this file) and the operator surface. The webhook endpoint is the
// latency layer between sweeps (design.md D3): user_created events link
// brand-new forum accounts within seconds of first login, and
// user_added_to_group/user_removed_from_group events on managed groups trigger targeted
// reconciles that correct forum-side drift. Missing or failed events cost
// only latency — the sweep restores correctness.
package web

import (
	"crypto/hmac"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"

	"git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse/client"
	"git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse/linkage"
	dcmod "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/discourse/store"
)

// WebhookPath is the Discourse webhook endpoint, declared CSRF-exempt by the
// adapter when (and only when) the webhook secret is configured.
const WebhookPath = "/webhooks/discourse"

// StarterFunc starts a person's targeted reconcile. Production wraps the
// Temporal client (per-person workflow ID serialization); tests inject a
// recorder.
type StarterFunc func(personID string) error

// WebhookHandlerConfig configures the webhook endpoint.
type WebhookHandlerConfig struct {
	DB     *sql.DB
	Logger *slog.Logger
	// Secret is the shared webhook secret (discourse-webhook-secret).
	Secret string
	// Mode selects reverse resolution for user_created events.
	Mode linkage.Mode
	// Client resolves admin user records (oidc/discourseconnect reverse
	// resolution).
	Client *client.Client
	// StartReconcile launches the targeted per-person reconcile.
	StartReconcile StarterFunc
}

// WebhookHandler verifies, dedupes, and dispatches Discourse webhook events.
type WebhookHandler struct {
	cfg WebhookHandlerConfig
}

// NewWebhookHandler constructs the handler.
func NewWebhookHandler(cfg WebhookHandlerConfig) *WebhookHandler {
	return &WebhookHandler{cfg: cfg}
}

// RegisterRoutes mounts the webhook endpoint.
func (h *WebhookHandler) RegisterRoutes(mux *http.ServeMux) {
	mux.HandleFunc("POST "+WebhookPath, h.serve)
}

// verifySignature checks X-Discourse-Event-Signature ("sha256=<hex>") as
// HMAC-SHA256 over the raw body, in constant time.
func (h *WebhookHandler) verifySignature(header string, body []byte) bool {
	const prefix = "sha256="
	if len(header) <= len(prefix) || header[:len(prefix)] != prefix {
		return false
	}
	want, err := hex.DecodeString(header[len(prefix):])
	if err != nil {
		return false
	}
	mac := hmac.New(sha256.New, []byte(h.cfg.Secret))
	mac.Write(body)
	return hmac.Equal(mac.Sum(nil), want)
}

func (h *WebhookHandler) serve(w http.ResponseWriter, r *http.Request) {
	const maxBodyBytes = 262144
	body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes))
	if err != nil {
		http.Error(w, "Bad request", http.StatusBadRequest)
		return
	}

	// Reject unsigned/mis-signed requests before any side effect (spec:
	// "rejected without side effects").
	if !h.verifySignature(r.Header.Get("X-Discourse-Event-Signature"), body) {
		h.cfg.Logger.Warn("discourse webhook rejected: invalid or missing signature")
		http.Error(w, "Invalid signature", http.StatusForbidden)
		return
	}

	eventName := r.Header.Get("X-Discourse-Event")
	switch eventName {
	case "user_created", "user_added_to_group", "user_removed_from_group":
	default:
		// Uninteresting event types are acknowledged and dropped so
		// Discourse doesn't retry them.
		w.WriteHeader(http.StatusOK)
		return
	}

	// Dedupe on the Discourse-assigned event id (fallback: body hash, which
	// still collapses identical redeliveries).
	eventID := r.Header.Get("X-Discourse-Event-Id")
	if eventID == "" {
		digest := sha256.Sum256(body)
		eventID = hex.EncodeToString(digest[:])
	}
	// NOT EXISTS rather than ON CONFLICT: the table's unique constraint
	// includes received_at (a partitioning requirement), so ON CONFLICT
	// would only collapse same-instant duplicates — a redelivery minutes
	// later would insert and reprocess. The remaining concurrent-delivery
	// race is harmless: reconcile starts are per-person serialized and
	// idempotent. (Recorded as findings.md #6 — the Stripe handler shares
	// this constraint shape.)
	res, err := h.cfg.DB.ExecContext(r.Context(),
		`INSERT INTO core.webhook_events (provider, provider_event_id, event_type, payload, status)
		 SELECT $1, $2, $3, $4, 'received'
		 WHERE NOT EXISTS (
		     SELECT 1 FROM core.webhook_events
		     WHERE provider = $1 AND provider_event_id = $2
		 )`,
		"discourse", eventID, eventName, body)
	if err != nil {
		h.cfg.Logger.Error("discourse webhook event insert failed", slog.Any("error", err))
		// Return 500 so Discourse retries — nothing was recorded.
		http.Error(w, "Internal error", http.StatusInternalServerError)
		return
	}
	if rows, _ := res.RowsAffected(); rows == 0 {
		// Duplicate delivery: already recorded (and processed or in
		// flight) — acknowledge without reprocessing.
		w.WriteHeader(http.StatusOK)
		return
	}

	status, procErr := h.process(r, eventName, body)
	if procErr != nil {
		h.cfg.Logger.Warn("discourse webhook processing failed",
			slog.String("event", eventName), slog.Any("error", procErr))
	}
	if _, err := h.cfg.DB.ExecContext(r.Context(),
		`UPDATE core.webhook_events SET status = $3, processed_at = NOW(), error_message = $4
		 WHERE provider = $1 AND provider_event_id = $2`,
		"discourse", eventID, status, errText(procErr)); err != nil {
		h.cfg.Logger.Error("discourse webhook status update failed", slog.Any("error", err))
	}

	// Always 200 once recorded: processing failures are visible in
	// webhook_events and the sweep is the correctness net; a 5xx would only
	// make Discourse redeliver an event we already recorded.
	w.WriteHeader(http.StatusOK)
}

func errText(err error) sql.NullString {
	if err == nil {
		return sql.NullString{}
	}
	return sql.NullString{String: err.Error(), Valid: true}
}

// process resolves the affected person and starts their targeted reconcile.
func (h *WebhookHandler) process(r *http.Request, eventName string, body []byte) (string, error) {
	// Shapes verified against live Discourse 3.5.3 (change dir,
	// live-verification.md): user events carry a full user serializer under
	// "user"; group-membership events carry the bare GroupUser join row under
	// "group_user" — forum group id + user id only, no names.
	var payload struct {
		User *struct {
			ID       int64  `json:"id"`
			Username string `json:"username"`
			Email    string `json:"email"`
		} `json:"user"`
		GroupUser *struct {
			GroupID int64 `json:"group_id"`
			UserID  int64 `json:"user_id"`
		} `json:"group_user"`
	}
	if err := json.Unmarshal(body, &payload); err != nil {
		return "failed", fmt.Errorf("parse payload: %w", err)
	}

	q := dcmod.New(h.cfg.DB)
	ctx := r.Context()

	switch eventName {
	case "user_created":
		if payload.User == nil {
			return "failed", errors.New("user event payload has no user object")
		}
		personID, err := h.resolvePerson(r, payload.User.ID, payload.User.Email)
		if err != nil {
			return "failed", err
		}
		if personID == "" {
			// No member-console person for this forum account — normal for
			// organic signups; nothing to reconcile.
			return "processed", nil
		}
		if err := h.cfg.StartReconcile(personID); err != nil {
			return "failed", fmt.Errorf("start reconcile: %w", err)
		}
		return "processed", nil

	case "user_added_to_group", "user_removed_from_group":
		if payload.GroupUser == nil {
			return "failed", errors.New("group event payload has no group_user object")
		}
		// Only managed groups matter; and only linked persons can be
		// reconciled (unlinked drift is the sweep's job). Automatic groups
		// (trust_level_*) fire these events on every signup, so the mapping
		// filter is load-bearing, not just hygiene.
		if _, err := q.GetGroupMappingByDiscourseGroupID(ctx, payload.GroupUser.GroupID); err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return "processed", nil
			}
			return "failed", fmt.Errorf("mapping lookup: %w", err)
		}
		link, err := q.GetUserLinkByDiscourseUserID(ctx, payload.GroupUser.UserID)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				return "processed", nil
			}
			return "failed", fmt.Errorf("link lookup: %w", err)
		}
		if err := h.cfg.StartReconcile(link.PersonID); err != nil {
			return "failed", fmt.Errorf("start reconcile: %w", err)
		}
		return "processed", nil
	}
	return "failed", fmt.Errorf("unhandled event %q", eventName)
}

// resolvePerson reverse-maps a forum user to a member-console person under
// the configured mode: oidc/discourseconnect read the admin record's
// external ids; email matches the verified primary email. An unresolvable
// user returns ("", nil) — not an error.
func (h *WebhookHandler) resolvePerson(r *http.Request, discourseUserID int64, email string) (string, error) {
	q := dcmod.New(h.cfg.DB)
	ctx := r.Context()

	switch h.cfg.Mode {
	case linkage.ModeOIDC, linkage.ModeDiscourseConnect:
		rec, err := h.cfg.Client.AdminUser(ctx, discourseUserID)
		if err != nil {
			return "", fmt.Errorf("admin user lookup: %w", err)
		}
		if rec == nil {
			return "", nil
		}
		var subject string
		if h.cfg.Mode == linkage.ModeOIDC {
			subject = rec.ExternalIDs["oidc"]
		} else if rec.SingleSignOnRecord != nil {
			subject = rec.SingleSignOnRecord.ExternalID
		}
		if subject == "" {
			return "", nil
		}
		person, err := q.FindPersonByOIDCSubject(ctx, subject)
		if errors.Is(err, sql.ErrNoRows) {
			return "", nil
		}
		if err != nil {
			return "", fmt.Errorf("person by subject: %w", err)
		}
		return person.PersonID, nil

	case linkage.ModeEmail:
		if email == "" {
			return "", nil
		}
		person, err := q.FindPersonByVerifiedEmail(ctx, email)
		if errors.Is(err, sql.ErrNoRows) {
			return "", nil
		}
		if err != nil {
			return "", fmt.Errorf("person by email: %w", err)
		}
		return person.PersonID, nil
	}
	return "", fmt.Errorf("unknown linkage mode %q", h.cfg.Mode)
}