Skip to main content

Push Notification Infrastructure Codemap

Architecture Overview

┌─────────────────────────────────────────────────────────────────────┐
│ Novu Framework (@novu/framework) │
│ Workflows registered via getWorkflowFactory() → workflow() │
│ Channels: step.push(), step.email(), step.inApp(), step.sms() │
│ Provider: OneSignal (via novu.subscribers.credentials.update) │
└─────────────────────────────────────────────────────────────────────┘

Packages involved: server/src/notification/, server/src/instantdb/, server/src/bullmq/, server/src/moderation/, server/src/templates/


1. Push Notification Type Definitions

File: server/src/notification/enums/push.enum.ts

42 push notification types across 6 categories:

CategoryTypes
Account & AuthWELCOME, EMAIL_VERIFICATION_REMINDER, PASSWORD_RESET_REQUEST, PASSWORD_CHANGED
Course & ConsultationCOURSE_PURCHASE_CONFIRMATION, COURSE_AVAILABLE, UPCOMING_CONSULTATION_REMINDER, CONSULTATION_COMPLETED_FEEDBACK_REQUEST, CONSULTATION_CANCELLED, COURSE_COMPLETION_BADGE, NEW_COURSE_IN_CATEGORY
Withdrawals & EarningsWITHDRAWAL_REQUEST_RECEIVED, WITHDRAWAL_APPROVED, WITHDRAWAL_DENIED, EARNINGS_UPDATE
EngagementINACTIVE_USER_REENGAGEMENT, COURSE_DISCOUNT_ALERT, INSTRUCTOR_MESSAGE_NOTIFICATION
Admin & SupportNEW_COURSE_SUBMITTED, COURSE_APPROVED, COURSE_REJECTED, USER_REPORTED_ALERT
Financial & SystemNEW_WITHDRAWAL_REQUEST, PAYOUT_PROCESSING_FAILURE, REVENUE_MILESTONE_ACHIEVEMENT, SUSPICIOUS_LOGIN_ATTEMPT, API_KEY_EXPIRY_WARNING, SERVICE_DOWNTIME_ALERT

2. Core Workflow Registration & Factory

File: server/src/notification/workflows.service.ts

Factory Pattern (workflows.service.ts:120-131)

private getWorkflowFactory()
├── NODE_ENV=TEST → returns mock { id, trigger: async => ({}) }
└── Production → const { workflow } = require('@novu/framework')
→ returns `workflow(id, steps, { payloadSchema })`

Returns Novu's workflow() function. All workflows follow:

workflow(Workflows.XXX, async ({ step, payload }) => {
await step.push('step-name', async () => {
// data retrieval happens INSIDE step callbacks
// renders templates, constructs { subject, body, data }
});
}, { payloadSchema: z.object({...}) });

Workflow Registration (workflows.service.ts:1422-1449)

getWorkFlows() returns an array of 23 workflow instances:

#WorkflowLinePush?
1consultationReminderWorkflow()490✅ (conditional)
2consultationRescheduledWorkflow()592❌ (inApp only)
3consultationRescheduleApprovedWorkflow()629❌ (inApp only)
4consultationCancelledWorkflow()668❌ (email + inApp)
5consultationEnrollmentWorkflow()738❌ (email + inApp)
6bookingConfirmationWorkflow()818❌ (email + inApp)
7orderNotificationWorkflow()911❌ (email only)
8productEnrollmentWorkflow()967❌ (email only)
9productGiftReceivedWorkflow()999❌ (email only)
10productGiftSentWorkflow()1028❌ (email only)
11pushNotificationWorkflow()132✅ Primary push
12inAppNotificationWorkflow()165❌ (inApp only)
13emailWorkflow()201❌ (email only)
14productUpdateWorkflow()1059❌ (email only)
15sendPaymentNotificationWorkflow()1089❌ (email only)
16sendUserActionNotificationWorkflow()1120❌ (email only)
17otpWorkflow()232❌ (sms + email)
18withdrawalMethodVerifiedWorkflow()1258❌ (email only)
19withdrawalTransactionWorkflow()1180❌ (email only)
20sendCourseCompletionNotificationWorkflow()1150❌ (email only)
21multistepWorkflow()294✅ Push + email + inApp + sms
22adminSystemAlertWorkflow()459❌ (inApp only)
23chatNotificationWorkflow()1381✅ Chat push with digest

3. Push Notification Workflow (Single Channel)

File: server/src/notification/workflows.service.ts:132-163

Invocation Path

pushNotificationWorkflow().trigger({
to: subscriberId,
payload: {
userId: string,
template: PushNotification, // enum key
data: { TITLE?, MESSAGE? }
}
})

Step Flow

step.push('send-push-notification')
├── userService.getUser(payload.userId)
├── templateService.processTemplate(
│ payload.template, // PushNotification enum
│ NotificationChannel.PUSH, // 'push'
│ payload.data, // interpolation variables
│ user // user context for template
│ )
├── Validates: subject && body !== null
└── Returns: { subject, body, data }

Typical Caller Pattern

Triggered mostly via triggerWithCircuitBreaker() which wraps the trigger in:

  • CircuitBreakerStrategy: 60% error threshold, 8 req volume, 15s sleep, 90s window
  • RetryStrategy: 3 max retries, 15s max delay, exponential backoff (1.5s base)

4. Chat Message Push Notification Flow

Message Creation & Queuing

ModerationService (moderation.service.ts:270-303)
└── After moderation pass:
└── notificationService.addNotificationJob({
messageId, roomId, senderId, senderName,
messageType, content, timestamp, roomName,
roomType, entityType
})
└── notificationQueue.add('PROCESS_NOTIFICATION', data)
(queue: NOTIFICATION)

BullMQ Notification Processor

File: server/src/notification/processors/notification.processor.ts

NotificationProcessor.process(job)
└── job.name === 'PROCESS_NOTIFICATION'
└── job.data.messageId exists?
├── YES → processChatNotification(job)
└── NO → handle other types

processChatNotification(job):
├── 1. getMessageDetails(messageId)
│ → instantdbService.getMessage()
├── 2. Check message age threshold
│ → settingsService.getSettingValue('CHAT_NOTIFICATION_MAX_AGE_MINUTES')
│ → Default: 2 minutes
│ → If too old: markMessageNotificationProcessed() + return
├── 3. Route by room type:

│ PRIVATE:
│ ├── getNotificationRecipients(roomId, senderId, timestamp)
│ │ ├── getRoomById() + getRoomParticipants()
│ │ ├── Filter: skip sender
│ │ └── Filter: shouldNotifyUser() per recipient
│ │ ├── isUserInRoomView(userId, roomId) → ephemeral presence
│ │ └── getUserLastSeen(userId, roomId) → persistent read receipt
│ └── triggerChatNotificationWorkflow(recipientId, messageData)
│ → triggerTarget = { subscriberId: recipientId }

│ GROUP:
│ ├── notificationService.ensureRoomTopic(roomId, roomName)
│ │ → deriveRoomTopicKey(roomId) → `room-${roomId}`
│ │ → Check if topic exists via novu.topics.get()
│ │ → If not: novu.topics.create({ key, name: `Room: ${roomName}` })
│ └── triggerChatNotificationWorkflow(null, messageData)
│ → triggerTarget = { type: "Topic", topicKey: `room-${roomId}` }

└── 4. Mark message as notification processed
→ instantdbService.markMessageNotificationProcessed(messageId)

Chat Notification Workflow Definition

File: server/src/notification/workflows.service.ts:1381-1420

chatNotificationWorkflow():
├── step.digest('batch-chat-messages', () => {
│ amount: digestWindow || 5 seconds
│ digestKey: subscriberId || `room-${roomId}`
│ })
└── step.push('send-chat-notification', () => {
subject: payload.sender // sender name for private, room name for group
body: payload.content // formatted + truncated content
data: { messageId, roomId, messageType, timestamp }
})

Message Formatting

File: server/src/notification/services/message-formatter.service.ts

Message TypeNotification Content
TEXTTruncated to CHAT_NOTIFICATION_CONTENT_MAX_LENGTH (default: 100) + ellipsis
FILE / IMAGE / VIDEO / AUDIO / DOCUMENT"You received a file"

Digest windows:

  • PRIVATE rooms: CHAT_NOTIFICATION_DIGEST_WINDOW_DM (1s)
  • GROUP rooms: CHAT_NOTIFICATION_DIGEST_WINDOW_GROUP (10s)
  • CATEGORY rooms: CHAT_NOTIFICATION_DIGEST_WINDOW_CATEGORY (10s)

5. Event-Driven Push Notification Flow

File: server/src/notification/listeners/notification.listener.ts

Trigger Pattern

All event handlers follow:

@OnEvent(AppEvents.XXX)
async handleXxxEvent(payload) {
await this.workflowsService.triggerWithCircuitBreaker(() =>
this.workflowsService.someWorkflow().trigger({ to, payload })
);
}

Events → Workflow Mapping (with push)

EventWorkflowPush?
SPACE_SESSION_LIVEmultistepWorkflow()push: { data: { TITLE, MESSAGE } }
SPACE_SESSION_REMINDER_24HRmultistepWorkflow()❌ (email + inApp)
SPACE_SESSION_REMINDER_1HRmultistepWorkflow()❌ (email + inApp)
SPACE_REGISTRATION_COMPLETEDmultistepWorkflow()❌ (email + inApp)
SPACE_REGISTRATION_CANCELLEDmultistepWorkflow()❌ (email + inApp)
COURSE_APPROVEDproductUpdateWorkflow()❌ (email only)
COURSE_REJECTIONproductUpdateWorkflow()❌ (email only)
CONSULTATION_APPROVEDproductUpdateWorkflow()❌ (email only)
CONSULTATION_REJECTEDproductUpdateWorkflow()❌ (email only)
PRODUCT_ENROLLMENTproductEnrollmentWorkflow()❌ (email only)

The SPACE_SESSION_LIVE handler (notification.listener.ts:419-450) is the only event-driven path that triggers push notifications:

await this.workflowsService.multistepWorkflow().trigger({
to: payload.recipientUserId,
payload: {
userId: payload.recipientUserId,
push: { data: { TITLE: 'Space is live now', MESSAGE: 'The session has started. Join now.' } },
inapp: { data: { ... } },
data: { spaceId, sessionId },
priority: 'high',
},
})

6. Multistep Workflow (Multi-Channel)

File: server/src/notification/workflows.service.ts:294-457

Single trigger → up to 4 channel steps, each with conditional skip:

multistepWorkflow():
├── step.email('send-multistep-email')
│ ├── IF email.key → processTemplate() with Enum key
│ ├── IF email.data → use TITLE/MESSAGE directly
│ └── skip: !payload.email

├── step.push('send-multistep-push')
│ ├── IF push.key → processTemplate() with PushNotification enum
│ ├── IF push.data → use TITLE/MESSAGE directly
│ └── skip: !payload.push

├── step.inApp('send-multistep-inapp')
│ ├── IF inapp.key → processTemplate() with InAppNotification enum
│ ├── IF inapp.data → use TITLE/MESSAGE directly
│ └── skip: !payload.inapp

└── step.sms('send-multistep-sms')
├── IF sms.key → processTemplate() with SmsNotification enum
├── IF sms.data → use MESSAGE directly
└── skip: !payload.sms

Payload schema:

{
userId: string,
eventId?: AppEvents,
email?: { key?: EmailNotification, data?: { TITLE, MESSAGE } },
push?: { key?: PushNotification, data?: { TITLE, MESSAGE } },
inapp?: { key?: InAppNotification, data?: { TITLE, MESSAGE } },
sms?: { key?: SmsNotification, data?: { MESSAGE } },
data?: Record<string, any>,
priority?: 'low' | 'medium' | 'high' | 'urgent'
}

7. Consultation Reminder Push

Scheduling (BullMQ → ReminderProcessor)

File: server/src/bullmq/processors/reminder.processor.ts

ReminderProcessor.process(job)
├── job.name === 'CONSULTATION_REMINDER_24HR'
│ → reminderType = TWENTY_FOUR_HOURS
├── job.name === 'CONSULTATION_REMINDER_30MIN'
│ → reminderType = THIRTY_MINUTES

├── Validate session exists & status is SCHEDULED/UPCOMING
├── Fetch student + consultant
├── Trigger for student: workflow.trigger({ to: student.id, ... })
└── Trigger for consultant: workflow.trigger({ to: consultant.id, ... })

Reminder Workflow Definition

File: server/src/notification/workflows.service.ts:490-590

consultationReminderWorkflow():
├── Data retrieval helper: getConsultationReminderData()
│ ├── sessionsService.findOne(sessionId, [relations...])
│ ├── Load: consultation → consultant → user
│ └── Determine userForEmail + otherParty

├── step.email('send-consultation-reminder-email') ← ALWAYS
│ └── template: EmailNotification.UPCOMING_CONSULTATION_REMINDER
│ with MJML template CONSULTATION

└── Conditional: reminderType === THIRTY_MINUTES
├── step.push('send-30min-push-reminder')
│ → subject: 'Session Starting Soon'
│ → body: `Your consultation with ${otherParty.firstName} is in 30 minutes.`
│ → data: { sessionId, meetingLink }
└── step.inApp('send-30min-in-app-reminder')
→ subject: 'Session Starting Soon'
→ body: same as push
→ data: { sessionId, meetingLink }

8. Device Token Registration & Novu Sync

File: server/src/notification/notification.service.ts

Client → Server Flow

Client sends: POST /notification/device-token (or similar)
→ NotificationController (unimplemented in controller — called from other services)

saveDeviceToken(data: DeviceTokenDto, user?: User) → notification.service.ts:166

├── 1. Check existing: deviceTokenRepo.findOne({ deviceToken, fingerprint })

├── 2. Token already exists:
│ ├── Update lastActive
│ ├── If no user attached but user provided: attach user + sync credentials
│ └── Return 'success'

├── 3. New token:
│ ├── deviceTokenRepo.save({ deviceToken, fingerprint, user, lastActive })
│ └── If user provided: fetch all user tokens → sync to Novu
│ └── setCredential(user.id, allDeviceTokens)

└── setCredential(subscriberId, deviceTokens) → notification.service.ts:297
├── Resolve provider ID:
│ - Config: ONESIGNAL_PROVIDER_ID
│ - Normalize: 'onesignal' → 'one-signal'
│ - Fallback: NovuProviderId.OneSignal = 'one-signal'
└── novu.subscribers.credentials.update({
providerId,
credentials: { deviceTokens }
}, subscriberId)

Device Token Entity

File: server/src/notification/entities/device-tokens.entity.ts

ColumnTypeNotes
idUUID (PK)Auto-generated
deviceTokenstringPush device token
fingerprintstringDevice fingerprint
userRelation → UserNullable (attached on first login)
lastActivetimestampUpdated on token reuse
createdAt / updatedAttimestampsAuto

9. Topic-Based Group Push

Topic Lifecycle

Room Creation (instantdb.service.ts:365-419):

getOrCreateRoom() or createRoom()
└── IF type === GROUP:
└── notificationService.ensureRoomTopic(roomId, roomName)

Topic Initialization (notification.service.ts:338-359):

ensureRoomTopic(roomId, roomName):
├── deriveRoomTopicKey(roomId) → `room-${roomId}`
├── Try: novu.topics.get(topicKey)
│ ├── Found: log + return
│ └── Not found: novu.topics.create({ key, name: `Room: ${roomName}` })
└── Return topicKey

Participant Subscription (instantdb.service.ts:137-188):

createRoomParticipant({ entityType, entityId, userId, role })
├── Create participant in InstantDB
└── IF room.type === GROUP:
└── notificationService.addSubscriberToTopic([userId], topicKey)

Participant Unsubscription (instantdb.service.ts:310-332):

leaveRoom(roomId, userId)
├── Delete participant from InstantDB
└── IF room.type === GROUP:
└── notificationService.removeSubscriberFromTopic(userId, topicKey)

Topic Notification Delivery

From Chat Processor (notification.processor.ts:287-296):

// GROUP room → topic trigger
triggerTarget = {
type: "Topic",
topicKey: `room-${messageData.roomId}`
}

await workflowsService.chatNotificationWorkflow().trigger({
to: triggerTarget,
payload: { ... }
})

Generic Topic Notification (notification.service.ts:271-286):

notifyTopic(topicKey: Topics | string, payload, workflowId?):
├── resolveTopicWorkflowId(topicKey)
│ ├── Topics.ADMIN → Workflows.ADMIN_SYSTEM_ALERT
│ └── defaultundefined (warns + returns null)
└── novu.trigger({ workflowId, to: { type: "Topic", topicKey }, payload })

10. Resilience & Error Handling

File: server/src/notification/workflows.service.ts:1293-1379

Circuit Breaker + Retry

@UseResilience(
CircuitBreakerStrategy({
errorThresholdPercentage: 60,
requestVolumeThreshold: 8,
sleepWindowInMilliseconds: 15000,
rollingWindowInMilliseconds: 90000,
timeoutInMilliseconds: 5000,
}),
RetryStrategy({
maxRetries: 3,
maxDelay: 15000,
backoff: new ExponentialBackoff({ baseDelay: 1500 }),
})
)
async triggerWithCircuitBreaker<T>(workflowTrigger: () => Promise<T>)

isRetryableError() classifies errors: timeouts, network errors, template processing errors, user lookup errors, Novu validation errors, rate limiting (429), service unavailable (502/503/504).

Chat Notification Failure Isolation

In NotificationProcessor (notification.processor.ts:136-142):

  • Errors are logged but NOT thrown (prevents job retry)
  • Chat message creation is decoupled from notification delivery
  • "Chat notification failed, but chat message creation succeeded"

11. Test Trigger Endpoint

File: server/src/notification/notification.controller.ts:99-138

POST /notification/test-trigger with { channel: 'push' | 'email' | 'inapp' | 'sms' | 'multistep' }:

ChannelWorkflowTrigger Payload
pushpushNotificationWorkflow(){ template: PushNotification.WELCOME, data: { TITLE, MESSAGE } }
emailemailWorkflow(){ template: EmailNotification.WELCOME, data: { TITLE, MESSAGE } }
inappinAppNotificationWorkflow(){ template: InAppNotification.CUSTOM, data: { TITLE, MESSAGE } }
smsotpWorkflow(){ template: SmsNotification.OTP_CODE, channel: SMS, data: { CODE, TITLE, MESSAGE } }
multistepmultistepWorkflow(){ email, push, inapp, sms all with data }

12. Novu Subscriber Management

File: server/src/notification/notification.service.ts

MethodLineDescription
createSubscriber(user)156Creates Novu subscriber with id, email, name
updateSubscriber(user)219Patches subscriber email + name
deactivateSubscriber(subscriberId)230Deletes subscriber from Novu
getSubscribers() / listSubscribers()195/203Lists subscribers (limit 20)
generateSubscriberHash(subscriberId)311HMAC-SHA256 hash for secure inbox
resubscribeUserToGroupRooms(userId)373Re-subscribes user to all GROUP room topics
resubscribeUserToRoomTopic(userId, roomId)409Re-subscribes user to single room topic

Topic Management

MethodLineDescription
addSubscriberToTopic(subscriberIds, topicKey)208Subscribe users to a topic
removeSubscriberFromTopic(subscriberId, topicKey)234Unsubscribe user from topic
createTopic(key, name)263Create a Novu topic
getTopic(key)258Get topic details
syncListToTopic(roles, topic)248Sync user roles to a topic
ensureRoomTopic(roomId, roomName)338Get-or-create room topic
deriveRoomTopicKey(roomId)325room-${roomId}
getNotificationTriggerTarget(roomType, recipientId, roomId)435Returns { subscriberId } or { type: "Topic", topicKey }

13. Key Enums & Constants

File: server/src/notification/enums/

EnumFilePurpose
PushNotificationpush.enum.ts42 push template keys
Workflowsworkflows.enum.ts40+ workflow ID strings
NotificationChannelnotification-channel.enum.tsEMAIL, PUSH, IN_APP, SMS
Topicstopics.enum.tsADMIN, USERS, SYSTEM
NovuProviderIdnovu.enums.tsOneSignal = 'one-signal'
NovuRecipientTypenovu.enums.tsTopic
AppEventscommon/enums/app-events.enum.ts170+ domain events

14. Data Flow Diagrams

Push Notification (Single Channel)

[Service/Controller]
→ workflowsService.pushNotificationWorkflow().trigger({
to: subscriberId,
payload: { userId, template, data }
})
→ Novu Framework → step.push('send-push-notification')
→ userService.getUser(userId)
→ templateService.processTemplate(template, 'push', data, user)
→ Returns { subject, body, data }
→ Novu dispatches to OneSignal
→ Device receives push

Chat Message Push

[User sends message]
→ ModerationService processes message
→ notificationService.addNotificationJob({ messageId, ... })
→ BullMQ Queue (NOTIFICATION)
→ NotificationProcessor.processChatNotification()
→ getMessageDetails()
→ Check message age (CHAT_NOTIFICATION_MAX_AGE_MINUTES)
→ Route: PRIVATE → per-recipient with shouldNotifyUser() filtering
GROUP → topic-based single trigger
→ MessageFormatterService.formatNotificationContent()
→ workflowsService.chatNotificationWorkflow().trigger()
→ step.digest() for batching
→ step.push('send-chat-notification') → { subject, body, data }
→ markMessageNotificationProcessed()

Device Token Registration

[Client registers device]
→ notificationService.saveDeviceToken({ token, fingerprint }, user)
→ Upsert in device-tokens table
→ Fetch all tokens for user
→ setCredential(userId, tokens)
→ novu.subscribers.credentials.update({
providerId: 'one-signal',
credentials: { deviceTokens: [...] }
})

Event-Driven Push (Space Session Live)

[Space goes live]
→ EventEmitter emits AppEvents.SPACE_SESSION_LIVE
→ NotificationListener.handleSpaceSessionLiveEvent()
→ triggerWithCircuitBreaker()
→ CircuitBreakerStrategy + RetryStrategy
→ multistepWorkflow().trigger({
to: recipientUserId,
payload: { push: { data: { TITLE, MESSAGE } }, inapp: {...}, ... }
})
→ Novu Framework
→ step.push('send-multistep-push') → subject, body, data
→ step.inApp('send-multistep-inapp') → subject, body, data

15. File Index

FilePathKey Lines
Workflows Serviceserver/src/notification/workflows.service.ts132, 294, 490, 1059, 1309, 1381, 1422
Notification Serviceserver/src/notification/notification.service.ts166, 208, 271, 297, 338, 373, 435
Notification Processorserver/src/notification/processors/notification.processor.ts70, 149, 183, 270, 350
Notification Listenerserver/src/notification/listeners/notification.listener.ts59, 201, 329, 374, 419, 452
Notification Controllerserver/src/notification/notification.controller.ts99
Message Formatterserver/src/notification/services/message-formatter.service.ts22, 92
Reminder Processorserver/src/bullmq/processors/reminder.processor.ts32
Push Enumserver/src/notification/enums/push.enum.ts1-43
Workflows Enumserver/src/notification/enums/workflows.enum.ts1-102
Notification Channel Enumserver/src/notification/enums/notification-channel.enum.ts1-6
Topics Enumserver/src/notification/enums/topics.enum.ts1-5
Novu Enumsserver/src/notification/enums/novu.enums.ts1-7
Device Token DTOserver/src/notification/dto/device-token.dto.ts1-11
Device Token Entityserver/src/notification/entities/device-tokens.entity.ts1-26
InstantDB Serviceserver/src/instantdb/instantdb.service.ts137, 247, 285, 365, 758, 905, 927
Instant Listenerserver/src/instantdb/listeners/instant.listener.ts51, 98, 219
Moderation Serviceserver/src/moderation/moderation.service.ts287
App Eventsserver/src/common/enums/app-events.enum.ts1-171
Template Serviceserver/src/templates/template.service.ts(template processing)