Push Notification Infrastructure Codemap

┌─────────────────────────────────────────────────────────────────────┐
│                        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)     │
└─────────────────────────────────────────────────────────────────────┘

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

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({...}) });

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

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 }

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)

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)

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 }
      })

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

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',
    },
})

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

{
    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'
}

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, ... })

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 }

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)

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

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

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

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

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

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

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

@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>)

[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

[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()

[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: [...] }
       })

[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