- Engagement scoring listener: 11 event subscriptions, weighted scoring (donation=50, subscription=40, shift=20, canvass=15, email=10, video=3), Redis sorted set leaderboard, per-contact score + last-activity tracking - Homepage stats listener: 12 subscriptions, incremental Redis counters (emails, signups, donations, responses, canvass, videos), capped recent activity lists (last 20 per type), cache invalidation on data changes - GET /api/homepage/live-stats — public real-time counters + recent activity - GET /api/observability/engagement-leaderboard — admin top-N contacts - Total: 8 listeners, 70 subscriptions across all modules Bunker Admin
7.5 KiB
7.5 KiB
Service Integrations — EventBus Architecture
Tracking document for the platform-wide EventBus and service integration work.
Started: 2026-03-30 Branch: v2
Architecture Overview
Changemaker Lite has 30+ services but most operate as isolated tools. The EventBus provides a centralized, typed, in-process pub/sub system that decouples event producers from consumers.
Service Handler (shift created, donation completed, etc.)
|
v
eventBus.publish('shift.created', payload)
|
+-- ListmonkListener (newsletter sync)
+-- RocketChatListener (team notifications)
+-- CrmActivityListener (contact timeline)
+-- CalendarSyncListener (unified calendar)
+-- N8nWebhookListener (external automation)
+-- GancioSyncListener (public event calendar)
Why In-Process EventEmitter (not Redis PubSub)
- Single Express process — no distributed coordination needed
- Zero serialization overhead (pass JS objects directly)
- Data already persisted in DB — events are ephemeral notifications
- Matches the existing fire-and-forget pattern used by Listmonk/RC services
- Can be swapped to Redis PubSub later if we go multi-process
Key Files
| File | Purpose |
|---|---|
api/src/types/events.ts |
Typed event catalog (all event names + payloads) |
api/src/services/event-bus.service.ts |
Core EventBus (publish/subscribe/stats) |
api/src/services/event-listeners/listmonk.listener.ts |
Listmonk newsletter sync |
api/src/services/event-listeners/rocketchat.listener.ts |
Rocket.Chat notifications |
api/src/services/event-listeners/crm-activity.listener.ts |
CRM ContactActivity writer |
api/src/services/event-listeners/calendar-sync.listener.ts |
Calendar unification |
api/src/services/event-listeners/n8n-webhook.listener.ts |
n8n automation bridge |
api/src/services/event-listeners/gancio.listener.ts |
Gancio event sync (shifts + ticketed events) |
api/src/services/event-listeners/engagement-scoring.listener.ts |
Contact engagement scores (Redis ZSET) |
api/src/services/event-listeners/homepage-stats.listener.ts |
Homepage real-time counters + cache invalidation |
Progress Tracker
Phase 1: Core Infrastructure
- Explore existing event patterns (Listmonk, RC, Gancio, provisioning)
- Design EventBus architecture
- Implement EventBus service (
api/src/services/event-bus.service.ts) - Define typed event catalog (
api/src/types/events.ts— 46 events across 14 modules) - Register EventBus in server.ts startup
- Add EventBus stats endpoint (
GET /api/observability/event-bus)
Phase 2: Migrate Existing Integrations
- Listmonk event sync → EventBus listener (9 event subscriptions)
- Rocket.Chat webhook service → EventBus listener (4 event subscriptions)
- Gancio shift/event sync → EventBus listener (3 event subscriptions)
Phase 3: New Listeners
- CRM Activity auto-generation listener (11 event subscriptions)
- Calendar sync listener (8 event subscriptions)
- n8n webhook emitter listener (wildcard subscription, forwards all events)
- Listmonk webhook receiver (inbound: open, click, bounce, unsubscribe → EventBus)
Phase 4: Wire Up Publishers (migrated from inline calls)
- Shift CRUD + signup (shift.created/updated/deleted, shift.signup.created/cancelled)
- Canvass session complete + visits (canvass.session.completed, contact.address.updated)
- Response submit (response.submitted)
- Campaign email sent (campaign.email.sent)
- Payment/donation/subscription events (3 event types)
- Contact tag changes (contact.tags.changed — 3 call sites)
- Reengagement sent (reengagement.sent)
- Campaign CRUD + publish + moderation (campaign.created/updated/deleted/published/status.changed)
- User create/update/delete/approve (user.created/updated/deleted/approved)
- SMS campaign start/complete + message send/receive (4 event types)
- Media video publish/unpublish/view (3 event types)
- Ticketed event publish/cancel (EventBus publishes alongside existing Gancio calls)
- Impact story publish (social.impact-story.published)
- Meeting create/delete (jitsi.routes.ts — meeting.created, meeting.deleted)
Phase 4b: Extended Listeners (2026-03-31)
- RC listener: +7 subscriptions (campaign.published, donations, subscriptions, SMS escalation, user.approved, video.published, ticketed-event.published)
- CRM listener: +2 subscriptions (subscription activated, email bounced)
- RC webhook service: +7 new formatter methods
- Prisma migration: SHIFT, MEETING, TICKETED_EVENT added to CalendarItemSource enum
- Calendar sync listener: uses proper source types (SHIFT, MEETING, TICKETED_EVENT)
Phase 4c: New Data Listeners (2026-03-31)
- Engagement scoring listener (11 subscriptions, Redis ZSET leaderboard)
- Homepage stats listener (12 subscriptions, Redis counters + recent activity)
- GET /api/homepage/live-stats endpoint (public, real-time counters + recent)
- GET /api/observability/engagement-leaderboard endpoint (admin, top contacts)
Phase 5: Future
- Migrate meeting-planner Gancio calls to EventBus (blocked: synchronous return value needed)
- Homepage service: swap COUNT queries for Redis counters in getStats()
- Engagement score materialization: periodic job to denormalize scores to Contact model
Event Catalog
Currently Wired (11 event points, 3 consumers)
| Event | Listmonk | Rocket.Chat | Gancio |
|---|---|---|---|
| shift.signup | yes | yes | - |
| shift.signup.cancelled | - | yes | - |
| shift.created | - | - | yes |
| shift.updated | - | - | yes |
| shift.deleted | - | - | yes |
| canvass.session.completed | yes | yes | - |
| canvass.address.updated | yes | - | - |
| campaign.email.sent | yes | - | - |
| response.submitted | - | yes | - |
| subscription.activated | yes | - | - |
| donation.completed | yes | - | - |
| product.purchased | yes | - | - |
| contact.tags.changed | yes | - | - |
| reengagement.sent | yes | - | - |
New Events (49+ handlers need publishers)
| Event | CRM Activity | Calendar | RC | n8n |
|---|---|---|---|---|
| campaign.created | - | - | - | yes |
| campaign.published | - | - | yes | yes |
| campaign.status.changed | - | - | yes | yes |
| user.approved | - | - | yes | yes |
| user.created | - | - | - | yes |
| video.published | - | - | yes | yes |
| video.viewed | yes | - | - | - |
| sms.message.received | yes | - | yes* | yes |
| sms.campaign.completed | - | - | yes | yes |
| ticketed-event.published | - | yes | - | yes |
| meeting.created | - | yes | - | - |
| impact-story.published | - | - | yes | yes |
| shift.created | - | yes | - | yes |
| donation.completed | yes | - | yes | yes |
| subscription.activated | yes | - | - | yes |
*SMS escalations (QUESTION/NEGATIVE sentiment) to relevant RC channel
Design Decisions
- Listeners self-guard: Each listener checks its own feature flag (ENABLE_CHAT, LISTMONK_SYNC_ENABLED, etc.) — the EventBus doesn't filter
- Error isolation: Each listener wraps its handler in try-catch; one listener failing doesn't affect others
- No persistence: Events are ephemeral — if the server restarts mid-event, it's lost (data is already in DB)
- Stats tracking: EventBus tracks per-event emission counts + per-listener execution counts for observability
- Wildcard subscriptions: Listeners can subscribe to
shift.*to catch all shift events