cgalo5758
88db730fcc
Introduce a commercial license option alongside AGPL-3.0-only, require a CLA for contributors, and document the terms in COMMERCIAL.md and NOTICE. Add a script to stamp SPDX headers on Go files and apply it across the tree.
293 lines
10 KiB
Go
293 lines
10 KiB
Go
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
|
|
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
|
|
|
|
// 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)
|
|
}
|