Architecture
What the system is. What it must do is docs/specifications/,
where each statement carries an identifier and a test that names it.
Event Flow
The nudging pipeline processes events through sequential stages:
| Stage | Component | Responsibility |
|---|---|---|
| 1. Ingestion | POST /admin/ingest-event |
Validates the incoming event payload |
| 2. Rule Evaluation | src/celine/nudging/engine/engine_service.py |
Matches event against rules, runs per-rule Python evaluators, renders messages |
| 3. Orchestration | src/celine/nudging/orchestrator/orchestrator.py |
Applies suppression, dedup, frequency limits, user preferences |
| 4. Delivery | src/celine/nudging/publishers/web/worker.py, src/celine/nudging/publishers/email/worker.py |
Sends web push or email |
Scheduled events follow the same pipeline but are triggered by the background scheduler instead of direct ingestion.
Components
Engine
The engine receives an event and:
- Resolves matching rules by
rule_idor event scenario - Loads each rule's custom Python evaluator from the seed directory
- Evaluates whether the rule should fire given the event payload
- Applies per-community rule overrides if configured
- Renders one Jinja2 title and body for the resolved language — not per channel: web push and email send the same rendered strings
Orchestrator
The orchestrator decides whether and how to deliver a notification:
- Checks per-user notification preferences (enabled, channels, per-kind opt-in/opt-out)
- Applies deduplication using rule_id:user_id:community_id:scope keys
- Enforces frequency limits (max_per_day)
- Emits delivery jobs for each applicable channel
Publishers
- Web push (
src/celine/nudging/publishers/web/worker.py) — sends VAPID-authenticated push via pywebpush - Email (
src/celine/nudging/publishers/email/worker.py) — sends via SMTP with TLS/SSL support
Scheduler
scheduler.py runs as a background task, polling scheduled_events every SCHEDULER_POLL_SECONDS. Due events are processed through the engine pipeline in batches.
Database Models
PostgreSQL (async via SQLAlchemy + asyncpg):
| Table | Purpose |
|---|---|
rules |
Rule definitions: id, kind, nudge_type, severity, definition (JSONB) |
rule_overrides |
Per-community overrides for rules |
templates |
Jinja2 title and body per rule and language, unique on (rule, lang) |
user_preferences |
Per-user preferences: enabled, channels, language, per-kind settings, max_per_day |
nudges_log |
Event processing log |
notifications |
Delivered notifications with read/deleted status |
delivery_log |
Per-channel delivery attempt records |
web_push_subscriptions |
Browser push subscription endpoints per user/community |
scheduled_events |
Future events to be processed at trigger_at, idempotent on external_key |
Authorization
Rego, evaluated in process through celine.sdk.policies (regorus) —
policies/celine/nudging/authz.rego. No OPA server is involved.
is_ingest— thenudging.ingestscope, or any administratoris_admin— thenudging.adminscope, or membership of theadmingroup (realm-level or organisation-level)- User endpoints — ownership in SQL, matching the token's
subor itspreferred_username. The bundle also publishes afiltersrule, and nothing reads it.
The bundle fails closed: the service will not start without it, and an evaluation that
does not return an explicit true denies. See
REQ-0003 – REQ-0010.
Stack
- Python >= 3.12, FastAPI, uvicorn
- SQLAlchemy 2 (async) + asyncpg
- Alembic for migrations
- pywebpush for VAPID web push
- Jinja2 for template rendering
- Pydantic settings