feat(webhook): embedded receiver for CiviCRM-triggered reconciles

.env.example

@@ -11,3 +11,8 @@ UNIFI_API_KEY=replace-me
 # Required only when alert.transport = "smtp":
 # SMTP_USERNAME=replace-me
 # SMTP_PASSWORD=replace-me
+
+# Required only when [webhook] enabled = true. Shared HMAC secret; must match
+# the ORIGIN_HMAC_SECRET configured on the door-webhook Cloudflare Worker.
+# Use a long random value (>= 16 chars), e.g. `openssl rand -hex 32`.
+# WEBHOOK_HMAC_SECRET=replace-me

.pre-commit-config.yaml

@@ -11,4 +11,5 @@ repos:
       - id: pyrefly-check
         name: Pyrefly (type checking)
         pass_filenames: false  # Recommended to do full repo checks. However, you can change this to `true` to only check changed files
-        additional_dependencies: [httpx>=0.28.1, pytest>=9.0.3, pytest-httpx>=0.30]
+        additional_dependencies:
+          [httpx>=0.28.1, pytest>=9.0.3, pytest-httpx>=0.30, flask>=3.1.3, waitress>=3.0.2]

config.example.toml

@@ -82,6 +82,19 @@ rank = 10
 # to = ["admin@example.com"]
 # subject_prefix = "[door-sync]"
 
+# Embedded webhook receiver (optional; disabled by default). When enabled, the
+# daemon runs a loopback-only Flask+waitress server in a second thread that a
+# Cloudflare Tunnel connects to locally — the Pi never exposes a public port.
+# Incoming requests must carry a valid HMAC signature (secret WEBHOOK_HMAC_SECRET
+# in the env file). See docs/configuration.rst and deploy/cloudflared-config.yml.
+# [webhook]
+# enabled = true
+# host = "127.0.0.1"        # loopback only; cloudflared connects locally
+# port = 8787
+# max_body_bytes = 65536    # reject larger request bodies (fail-closed)
+# max_skew_seconds = 300    # reject signed requests older/newer than this (replay guard)
+# debounce_seconds = 2.0    # settle window: a burst of webhooks collapses into one reconcile
+
 [ops]
 # Operational file paths. All three are optional; defaults shown.
 # - audit_jsonl: append-only JSONL of every cycle's outcome (logrotate-friendly).

deploy/cloudflared-config.yml

@@ -0,0 +1,28 @@
+# Cloudflare Tunnel config for the Pi. Routes ONE public hostname to the
+# loopback-only door-sync webhook receiver (config.toml [webhook]).
+#
+# cloudflared runs as its OWN systemd service (see cloudflared.service) —
+# separate from door-sync, so restarting the tunnel never disturbs the
+# reconcile loop, and the door-sync unit keeps exposing no public port.
+#
+# One-time setup (run interactively as the cloudflared user):
+#   cloudflared tunnel login
+#   cloudflared tunnel create door-sync
+#   cloudflared tunnel route dns door-sync door-sync.example.org
+# The `create` step writes a credentials JSON and prints the tunnel UUID.
+# Put the UUID in `tunnel:` below and the JSON at `credentials-file:`.
+#
+# Lock the hostname down in the Cloudflare dashboard with a Zero Trust
+# Access application using a SERVICE TOKEN policy, so only the door-webhook
+# Worker (presenting CF-Access-Client-Id / CF-Access-Client-Secret) can reach
+# it. The webhook's HMAC check (WEBHOOK_HMAC_SECRET) is the second, independent
+# lock.
+
+tunnel: <TUNNEL-UUID>
+credentials-file: /etc/cloudflared/door-sync.json
+
+ingress:
+  - hostname: door-sync.example.org
+    service: http://127.0.0.1:8787
+  # Everything else is refused.
+  - service: http_status:404

deploy/cloudflared.service

@@ -0,0 +1,23 @@
+[Unit]
+Description=cloudflared tunnel for the door-sync webhook receiver
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=cloudflared
+Group=cloudflared
+ExecStart=/usr/local/bin/cloudflared tunnel --config /etc/cloudflared/config.yml run
+Restart=on-failure
+RestartSec=5s
+StandardOutput=journal
+StandardError=journal
+
+# Hardening — the tunnel needs only outbound network + its config/credentials.
+NoNewPrivileges=true
+ProtectSystem=strict
+ProtectHome=true
+PrivateTmp=true
+
+[Install]
+WantedBy=multi-user.target

docs/configuration.rst

@@ -72,6 +72,8 @@ These are required only when the corresponding alert transport is enabled:
      - ``alert.transport = "smtp"``
    * - ``MAILGUN_API_KEY``
      - ``alert.transport = "mailgun"``
+   * - ``WEBHOOK_HMAC_SECRET``
+     - ``webhook.enabled = true``
 
 Example env file:
 
@@ -359,6 +361,40 @@ The Mailgun API key is read from the ``MAILGUN_API_KEY`` env variable.
    email is a convenience layer.
 
 
+``[webhook]`` — Embedded Webhook Receiver
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Optional and **disabled by default**. When enabled, daemon mode runs a small
+sync Flask + waitress HTTP server in a second thread. It exists so a CiviCRM
+change can trigger a reconcile within seconds instead of waiting up to a full
+polling cadence. The server binds **loopback only** — a Cloudflare Tunnel
+(``cloudflared``) connects to it locally, so the Pi never exposes a public
+port. Inbound requests must carry a valid HMAC signature; the HTTP thread does
+nothing but verify the signature and enqueue a trigger, and the scheduler
+thread (the sole writer to UniFi/state/audit) drains it and runs a full
+reconcile, coalescing bursts.
+
+.. code-block:: toml
+
+   [webhook]
+   enabled = true
+   host = "127.0.0.1"        # loopback only; cloudflared connects locally
+   port = 8787
+   max_body_bytes = 65536    # reject larger request bodies (fail-closed)
+   max_skew_seconds = 300    # reject signed requests outside this clock skew (replay guard)
+   debounce_seconds = 2.0    # settle window: a burst of webhooks collapses into one reconcile
+
+The shared secret is read from the ``WEBHOOK_HMAC_SECRET`` env variable and
+must match the ``ORIGIN_HMAC_SECRET`` configured on the ``door-webhook``
+Cloudflare Worker. Requests are signed as
+``HMAC_SHA256(secret, f"{timestamp}." + raw_body)`` with headers
+``X-Door-Sync-Timestamp`` and ``X-Door-Sync-Signature: sha256=<hex>``.
+
+See ``deploy/cloudflared.service`` and ``deploy/cloudflared-config.yml`` for the
+tunnel setup, and front the hostname with a Cloudflare Access service-token
+policy so only the Worker can reach it.
+
+
 Complete Example
 ----------------
 

pyproject.toml

@@ -8,7 +8,9 @@ authors = [
 ]
 requires-python = ">=3.11"
 dependencies = [
+    "flask>=3.1.3",
     "httpx>=0.28.1",
+    "waitress>=3.0.2",
 ]
 
 [project.scripts]

src/door_sync/__main__.py

@@ -16,16 +16,17 @@
 
 import argparse
 import logging
+import queue
 import sys
 from pathlib import Path
 
-from door_sync import cli, orchestrator, reconciler, scheduler, tier_mapping
+from door_sync import cli, orchestrator, reconciler, scheduler, tier_mapping, webhook
 from door_sync import config as config_mod
 from door_sync.civicrm.client import CivicrmClient
 from door_sync.unifi.client import UnifiClient
 
 # Expose config_mod so tests can monkeypatch it via main_mod.config_mod.
-__all__ = ["config_mod", "CivicrmClient", "UnifiClient", "scheduler", "main"]
+__all__ = ["config_mod", "CivicrmClient", "UnifiClient", "scheduler", "webhook", "main"]
 
 _logger = logging.getLogger("door_sync")
 
@@ -123,7 +124,15 @@ def cmd_run(args: argparse.Namespace) -> int:
         return 1
 
     if not args.once:
-        return scheduler.run_forever(config, dry_run=args.dry_run)
+        work_queue: queue.Queue[object] = queue.Queue()
+        server = (
+            webhook.start(config.webhook, work_queue=work_queue) if config.webhook.enabled else None
+        )
+        try:
+            return scheduler.run_forever(config, dry_run=args.dry_run, work_queue=work_queue)
+        finally:
+            if server is not None:
+                server.stop()
 
     try:
         result = orchestrator.reconcile(config, dry_run=args.dry_run)

src/door_sync/config.py

@@ -137,6 +137,47 @@ class OpsPaths:
     alert_flag: Path
 
 
+@dataclass(frozen=True)
+class WebhookConfig:
+    """Embedded webhook-receiver settings. Secret: hmac_secret (from env).
+
+    The receiver is an optional Flask+waitress server run in a second thread of
+    the daemon (see webhook.py). It binds loopback only; a Cloudflare Tunnel
+    connects locally, so the Pi never exposes a public port.
+
+    Parameters:
+        enabled: Whether daemon mode starts the local webhook HTTP server.
+        host: Bind address. Loopback only in production.
+        port: TCP port for the local webhook server.
+        hmac_secret: Shared secret for HMAC-SHA256 request signing (env
+            WEBHOOK_HMAC_SECRET). Required only when enabled.
+        max_body_bytes: Reject request bodies larger than this (fail-closed).
+        max_skew_seconds: Reject signed requests whose timestamp is more than
+            this many seconds from now (replay protection).
+        debounce_seconds: Settle window before a triggered reconcile runs, so a
+            burst of webhooks collapses into one cycle.
+    """
+
+    enabled: bool
+    host: str
+    port: int
+    hmac_secret: str
+    max_body_bytes: int
+    max_skew_seconds: int
+    debounce_seconds: float
+
+
+_DEFAULT_WEBHOOK_CONFIG = WebhookConfig(
+    enabled=False,
+    host="127.0.0.1",
+    port=8787,
+    hmac_secret="",
+    max_body_bytes=65536,
+    max_skew_seconds=300,
+    debounce_seconds=2.0,
+)
+
+
 @dataclass(frozen=True)
 class Config:
     """Top-level configuration assembled from TOML + env by `load()`.
@@ -149,6 +190,7 @@ class Config:
         tier_mapping: Rules mapping membership types to access policies.
         ops_paths: File paths for audit log, state, and alert flag.
         alert: Alert transport configuration.
+        webhook: Embedded webhook-receiver settings (disabled by default).
     """
 
     cadence_seconds: int
@@ -158,6 +200,7 @@ class Config:
     tier_mapping: TierMapping
     ops_paths: OpsPaths
     alert: AlertConfig
+    webhook: WebhookConfig = _DEFAULT_WEBHOOK_CONFIG
 
 
 @dataclass(frozen=True)
@@ -285,6 +328,7 @@ def env_get(name: str) -> str | None:
     tier_mapping = _validate_tier_mapping(data, issues)
     ops_paths = _validate_ops(data, issues)
     alert_config = _validate_alert(data, issues, env_get)
+    webhook = _validate_webhook(data, issues, env_get)
 
     if issues:
         raise ConfigError(issues)
@@ -297,6 +341,100 @@ def env_get(name: str) -> str | None:
         tier_mapping=tier_mapping,
         ops_paths=ops_paths,
         alert=alert_config,
+        webhook=webhook,
+    )
+
+
+def _validate_webhook(
+    data: dict[str, Any],
+    issues: list[ConfigIssue],
+    env_get: EnvGetter,
+) -> WebhookConfig:
+    """Validate the optional [webhook] table.
+
+    When absent or disabled, returns the disabled default and requires no
+    secret (mirrors _validate_alert returning the flag-file default). When
+    enabled, the HMAC secret comes from env WEBHOOK_HMAC_SECRET and is required.
+    """
+    section = data.get("webhook", {})
+    if not isinstance(section, dict):
+        issues.append(ConfigIssue(path="webhook", message="must be a table"))
+        return _DEFAULT_WEBHOOK_CONFIG
+
+    enabled = section.get("enabled", False)
+    if not isinstance(enabled, bool):
+        issues.append(
+            ConfigIssue(
+                path="webhook.enabled",
+                message=f"must be bool, got {type(enabled).__name__}",
+            )
+        )
+        return _DEFAULT_WEBHOOK_CONFIG
+    if not enabled:
+        return _DEFAULT_WEBHOOK_CONFIG
+
+    host = section.get("host", "127.0.0.1")
+    if not isinstance(host, str) or not host:
+        issues.append(ConfigIssue(path="webhook.host", message="must be non-empty string"))
+        host = "127.0.0.1"
+
+    port = section.get("port", 8787)
+    if isinstance(port, bool) or not isinstance(port, int):
+        issues.append(
+            ConfigIssue(path="webhook.port", message=f"must be int, got {type(port).__name__}")
+        )
+        port = 8787
+    elif not (1 <= port <= 65535):
+        issues.append(
+            ConfigIssue(path="webhook.port", message=f"must be between 1 and 65535, got {port}")
+        )
+        port = 8787
+
+    max_body = section.get("max_body_bytes", 65536)
+    if isinstance(max_body, bool) or not isinstance(max_body, int) or max_body <= 0:
+        issues.append(ConfigIssue(path="webhook.max_body_bytes", message="must be a positive int"))
+        max_body = 65536
+
+    max_skew = section.get("max_skew_seconds", 300)
+    if isinstance(max_skew, bool) or not isinstance(max_skew, int) or max_skew < 0:
+        issues.append(ConfigIssue(path="webhook.max_skew_seconds", message="must be an int >= 0"))
+        max_skew = 300
+
+    debounce_raw = section.get("debounce_seconds", 2.0)
+    if (
+        isinstance(debounce_raw, bool)
+        or not isinstance(debounce_raw, (int, float))
+        or debounce_raw < 0
+    ):
+        issues.append(ConfigIssue(path="webhook.debounce_seconds", message="must be a number >= 0"))
+        debounce = 2.0
+    else:
+        debounce = float(debounce_raw)
+
+    hmac_secret = (env_get("WEBHOOK_HMAC_SECRET") or "").strip()
+    if not hmac_secret:
+        issues.append(
+            ConfigIssue(
+                path="WEBHOOK_HMAC_SECRET",
+                message="required env var is missing or empty when webhook.enabled is true",
+            )
+        )
+    elif len(hmac_secret) < 16:
+        issues.append(
+            ConfigIssue(
+                path="WEBHOOK_HMAC_SECRET",
+                message="must be at least 16 characters",
+            )
+        )
+
+    return WebhookConfig(
+        enabled=True,
+        host=host,
+        port=port,
+        hmac_secret=hmac_secret,
+        max_body_bytes=max_body,
+        max_skew_seconds=max_skew,
+        debounce_seconds=debounce,
     )
 
 

src/door_sync/models.py

@@ -116,6 +116,25 @@ class ReconcileResult:
     diff: Diff | None
 
 
+@dataclass(frozen=True)
+class ReconcileRequest:
+    """A queued webhook trigger asking the scheduler to run a reconcile soon.
+
+    Enqueued by the webhook receiver thread and drained by the scheduler thread.
+    The reconcile is always whole-population (the CiviCRM client has no
+    per-contact fetch); contact_id is carried for logging only.
+
+    Parameters:
+        reason: Short label for why the reconcile was requested (e.g. the
+            webhook event type). Logged, never a member name.
+        contact_id: CiviCRM contact ID from the triggering event, for logging
+            only, or None when the payload did not carry one.
+    """
+
+    reason: str
+    contact_id: int | None = None
+
+
 @dataclass(frozen=True)
 class TierRule:
     """A single tier-mapping rule from configuration.