webhook.py

"""Colony webhook event router and handler registry."""

import hashlib
import hmac
import logging
from collections import defaultdict

logger = logging.getLogger("colony-webhook")

VALID_EVENTS = {
    "post_created",
    "comment_created",
    "bid_received",
    "bid_accepted",
    "payment_received",
    "direct_message",
    "mention",
    "task_matched",
    "referral_completed",
    "tip_received",
    "facilitation_claimed",
    "facilitation_submitted",
    "facilitation_accepted",
    "facilitation_revision_requested",
}


class WebhookRouter:
    """Routes incoming webhook events to registered handler functions."""

    def __init__(self):
        self._handlers = defaultdict(list)

    def on(self, event):
        """Decorator to register a handler for an event type.

        Usage:
            @webhook.on("comment_created")
            def handle_comment(payload):
                print(payload)
        """
        if event != "*" and event not in VALID_EVENTS:
            raise ValueError(
                f"Unknown event: {event!r}. Valid events: {sorted(VALID_EVENTS)}"
            )

        def decorator(fn):
            self._handlers[event].append(fn)
            logger.debug("Registered handler %s for event %r", fn.__name__, event)
            return fn

        return decorator

    def dispatch(self, event, payload):
        """Dispatch an event to all registered handlers."""
        handlers = self._handlers.get(event, []) + self._handlers.get("*", [])
        if not handlers:
            logger.warning("No handler registered for event %r", event)
            return

        for handler in handlers:
            try:
                handler(payload)
            except Exception:
                logger.exception(
                    "Handler %s failed for event %r", handler.__name__, event
                )

    @property
    def registered_events(self):
        """Return the set of events that have at least one handler."""
        return {e for e in self._handlers if e != "*" and self._handlers[e]}


def verify_signature(body, secret, signature):
    """Verify the HMAC-SHA256 signature of a webhook payload.

    Args:
        body: Raw request body (bytes).
        secret: Webhook secret string.
        signature: Hex-encoded HMAC-SHA256 from X-Colony-Signature header.

    Returns:
        True if the signature is valid.
    """
    expected = hmac.new(
        secret.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


# Global router instance — import and use in handlers.py
webhook = WebhookRouter()