Platform Notification System — Unified Dispatch for GoGrab Users

Session: S241 Date: 2026-03-19 Status: Design Proposed Owner: backend-vitto-api (primary), @libs-api/messaging-core (shared delivery), backend-api (bridge consumer) Predecessor: S210.4 (ProactiveDispatcherService), S211 (Rich Messaging), S217 (Cross-Channel Bridge), S218 (Whistle)

Executive Summary

The platform notification system handles all messages from the GoGrab platform to its users — space builders, executives, admins, and developers (L1 users). Today, ProactiveDispatcherService handles single-user dispatch with safety guards but lacks broadcast capability, a notification log, custom message support, and an internal bridge for other backends to trigger notifications.

This spec extends the existing infrastructure with four additions:

1. Broadcast fan-out — send to all users in a space, all admins, or all platform users with batching and throttling
2. Notification log — persistent audit trail for every dispatch attempt
3. Custom message support — HTML email templates, rich WhatsApp cards, custom payloads
4. Internal bridge endpoint — lets backend-api (billing, deployments, ecommerce) trigger notifications through Vitto's dispatch pipeline without duplicating delivery logic

Design principle: This is NOT a rewrite. It extends ProactiveDispatcherService and ChannelGatewayService which are battle-tested and already enforce quiet hours, frequency caps, dedup, and opt-out. Zero existing behavior changes.

Table of Contents

1. Architecture Overview
2. Broadcast Service
3. Notification Log
4. Custom Message Support
5. Internal Bridge Endpoint
6. Schemas
7. API Endpoints
8. Integration Points
9. Key Files to Create/Modify
10. Risks & Mitigations

1. Architecture Overview

┌──────────────────────────────────────────────────────────────┐
│                    NOTIFICATION SOURCES                        │
│                                                                │
│  backend-vitto-api (internal)          backend-api (bridge)    │
│  ┌───────────────────────┐    HTTP    ┌──────────────────────┐│
│  │ HeartbeatService      │◄───────────│ BillingController    ││
│  │ QualityDropTrigger    │    POST    │ DeploymentService    ││
│  │ DailyDigestService    │  /internal │ EcommerceService     ││
│  │ PerfAlertService      │  /notif/   │ SingularityEngine    ││
│  │ MilestoneService      │  dispatch  │ (future sources)     ││
│  │ WhistleOrchestrator   │            └──────────────────────┘│
│  └───────────┬───────────┘                                     │
│              │ all funnel through                               │
│              ▼                                                  │
│  ┌───────────────────────────────────────────────┐             │
│  │         ProactiveDispatcherService             │             │
│  │  (EXISTING — extended, not replaced)           │             │
│  │                                                │             │
│  │  dispatch()        → single user               │             │
│  │  dispatchBroadcast() → NEW: fan-out to group   │             │
│  │                                                │             │
│  │  Safety guards:                                │             │
│  │  ✓ Master toggle    ✓ Per-trigger opt-out      │             │
│  │  ✓ Quiet hours      ✓ Frequency cap            │             │
│  │  ✓ 1h dedup         ✓ Critical bypass          │             │
│  └──────────────┬────────────────────────────────┘             │
│                 │                                               │
│    ┌────────────┼────────────────┐                              │
│    ▼            ▼                ▼                              │
│  Web (SSE)  Channel Delivery  Notification Log                 │
│  VittoNar-  ChannelGateway-   NotificationLog-                 │
│  rative     Service           Service (NEW)                    │
│  Service    (EXISTING)                                         │
│             │                  Persists every                   │
│             ▼                  attempt + result                 │
│  @libs-api/messaging-core      to MongoDB                      │
│  TwilioService │ EmailService                                  │
│  SlackManager  │ TelegramGroup                                 │
└──────────────────────────────────────────────────────────────┘

2. Broadcast Service

Problem

ProactiveDispatcherService.dispatch() takes a single userId. There is no way to notify "all users in space X" or "all platform admins."

Design

Add dispatchBroadcast() to ProactiveDispatcherService (not a new service — keeps the single entry point invariant).

// Added to ProactiveDispatcherService

interface BroadcastRequest {
  /** Target audience */
  target:
    | { type: 'space'; spaceId: string; roles?: SpaceMemberRole[] }
    | { type: 'platform'; roles?: PlatformRole[] }
    | { type: 'userIds'; userIds: string[] };

  /** Notification content */
  triggerName: string;
  severity: 'info' | 'warning' | 'critical';
  title: string;
  message: string;

  /** Custom content (see Section 4) */
  customContent?: CustomNotificationContent;

  /** Delivery control */
  metadata?: Record<string, unknown>;
  bypassGuards?: boolean;

  /** Throttle: max concurrent deliveries (default: 50) */
  concurrency?: number;

  /** Scheduling */
  scheduledAt?: Date; // Future-dated delivery
}

interface BroadcastResult {
  broadcastId: string;        // UUID for tracking
  totalRecipients: number;
  dispatched: number;
  suppressed: number;         // By safety guards
  failed: number;
  channels: Record<string, number>; // { whatsapp: 12, web: 45, email: 3 }
}

Resolution logic

async resolveRecipients(target: BroadcastRequest['target']): Promise<string[]> {
  switch (target.type) {
    case 'space': {
      // Call backend-api internal endpoint
      const members = await this.backendApiClient.getSpaceMembers(target.spaceId);
      return target.roles
        ? members.filter(m => target.roles.includes(m.role)).map(m => m.userId)
        : members.map(m => m.userId);
    }
    case 'platform': {
      // Call backend-api internal endpoint
      return this.backendApiClient.getUsersByRoles(target.roles || []);
    }
    case 'userIds':
      return target.userIds;
  }
}

Batching and throttling

• Uses p-limit (or manual semaphore) to cap concurrent dispatches at concurrency (default 50)
• Twilio rate limit is ~80 msg/sec for WhatsApp — batch delay of 50ms between sends ensures compliance
• For broadcasts >500 recipients, automatically delegates to ScheduledMessageQueueService for resilient delivery with retry
• Each individual dispatch goes through dispatch() — all safety guards apply per-user

Scheduled broadcasts

If scheduledAt is provided, dispatchBroadcast() resolves recipients immediately, then queues individual messages via ScheduledMessageQueueService.schedule() with the target timestamp. The existing cron (every 1 minute) picks them up.

3. Notification Log

Problem

No audit trail exists for dispatched notifications. You can't answer "what did we send to user X last week?" or build a notification inbox UI.

Design

A new NotificationLogService in backend-vitto-api that persists every dispatch() and dispatchBroadcast() result.

@Injectable()
export class NotificationLogService {
  constructor(
    @InjectModel(NotificationLogDoc.name)
    private readonly logModel: Model<NotificationLogDoc>,
  ) {}

  async log(entry: NotificationLogEntry): Promise<void> {
    await this.logModel.create(entry);
  }

  async getForUser(
    userId: string,
    opts?: { limit?: number; offset?: number; since?: Date; unreadOnly?: boolean },
  ): Promise<NotificationLogDoc[]> { ... }

  async markRead(userId: string, notificationIds: string[]): Promise<void> { ... }

  async getUnreadCount(userId: string): Promise<number> { ... }
}

Integration point

ProactiveDispatcherService.dispatch() calls this.notificationLogService.log() after delivery (fire-and-forget). Zero impact on delivery latency.

4. Custom Message Support

Problem

Current dispatcher sends plain text + optional rich card. No support for custom HTML emails, branded WhatsApp templates, or structured payloads.

Design

Add a customContent field to DispatchRequest and BroadcastRequest:

interface CustomNotificationContent {
  /** HTML email body (rendered server-side). Overrides plain text for email channel. */
  emailHtml?: string;

  /** Email subject line. Defaults to title if not provided. */
  emailSubject?: string;

  /** SendGrid dynamic template ID. If provided, emailHtml is ignored. */
  sendgridTemplateId?: string;

  /** SendGrid dynamic template variables */
  sendgridTemplateData?: Record<string, unknown>;

  /** WhatsApp Content API template SID (pre-approved). Required for business-initiated messages outside 24h window. */
  whatsappContentSid?: string;

  /** WhatsApp Content API template variables */
  whatsappContentVariables?: Record<string, string>;

  /** Slack Block Kit blocks (overrides plain text for Slack channel) */
  slackBlocks?: unknown[];

  /** Arbitrary structured payload (forwarded to channel adapters as metadata) */
  payload?: Record<string, unknown>;
}

Delivery logic change in dispatch()

// In ProactiveDispatcherService.dispatch(), after resolveChannels():

for (const channel of channels) {
  if (channel.type === 'email' && customContent?.emailHtml) {
    await this.channelGateway.sendCustomEmail({
      to: channel.id,
      subject: customContent.emailSubject || title,
      html: customContent.emailHtml,
      spaceId: metadata?.spaceId,  // For VittoIdentity BYOT resolution
    });
  } else if (channel.type === 'email' && customContent?.sendgridTemplateId) {
    await this.channelGateway.sendTemplatedEmail({
      to: channel.id,
      templateId: customContent.sendgridTemplateId,
      dynamicTemplateData: customContent.sendgridTemplateData,
      spaceId: metadata?.spaceId,
    });
  } else if (channel.type === 'whatsapp' && customContent?.whatsappContentSid) {
    await this.channelGateway.sendWhatsAppTemplate({
      to: channel.id,
      contentSid: customContent.whatsappContentSid,
      contentVariables: customContent.whatsappContentVariables,
      spaceId: metadata?.spaceId,
    });
  } else if (channel.type === 'slack' && customContent?.slackBlocks) {
    await this.channelGateway.sendSlackBlocks({
      channelId: channel.id,
      blocks: customContent.slackBlocks,
    });
  } else {
    // Existing path: plain text + optional rich card adapter
    ...existing code...
  }
}

Email HTML rendering

For emailHtml, the ChannelGatewayService.sendCustomEmail() method calls EmailService.sendRaw() (new method on the existing service):

// Added to EmailService in @libs-api/messaging-core
async sendRaw(opts: {
  to: string;
  subject: string;
  html: string;
  from?: string;
  fromName?: string;
  replyTo?: string;
}): Promise<boolean> {
  // Uses SendGrid v3 API with custom HTML body
  // Resolves credentials via VittoIdentityService if spaceId provided
}

Security: emailHtml is server-provided (from backend code or admin APIs), never from end-user input. No XSS vector. If future admin UI allows HTML editing, sanitize with DOMPurify on the backend before storage.

SendGrid Dynamic Templates

For businesses that use SendGrid dynamic templates (version-controlled, designed in SendGrid UI), sendgridTemplateId bypasses local HTML entirely. The dynamicTemplateData object is passed as-is to SendGrid's API, which handles Handlebars substitution server-side.

5. Internal Bridge Endpoint

Problem

backend-api has 3 commented-out billing notification stubs and no way to call Vitto's dispatch pipeline. Today, backend-api → backend-vitto-api HTTP calls don't exist (only the reverse).

Design

On backend-vitto-api: New internal controller

// APIs/backend-vitto-api/src/app/engine/controllers/internal-notification.controller.ts

@Controller('vitto-api/engine/internal/notifications')
export class InternalNotificationController {
  constructor(
    private readonly dispatcher: ProactiveDispatcherService,
  ) {}

  @Post('dispatch')
  @UseGuards(ServiceOnlyGuard)  // Existing guard — validates BACKEND_API_INTERNAL_TOKEN
  async dispatch(@Body() request: DispatchRequest): Promise<DispatchResult> {
    return this.dispatcher.dispatch(request);
  }

  @Post('broadcast')
  @UseGuards(ServiceOnlyGuard)
  async broadcast(@Body() request: BroadcastRequest): Promise<BroadcastResult> {
    return this.dispatcher.dispatchBroadcast(request);
  }
}

On @libs-api/messaging-core: Bridge client

// libs/libs-api/messaging-core/src/lib/services/notification-bridge-client.service.ts

@Injectable()
export class NotificationBridgeClientService {
  private readonly vittoApiUrl: string;
  private readonly internalToken: string;

  constructor(private readonly configService: ConfigService) {
    this.vittoApiUrl = this.configService.get('VITTO_API_BASE_URL', 'http://localhost:3003');
    this.internalToken = this.configService.get('BACKEND_API_INTERNAL_TOKEN', '');
  }

  async dispatch(request: DispatchRequest): Promise<DispatchResult> {
    return this.post('/vitto-api/engine/internal/notifications/dispatch', request);
  }

  async broadcast(request: BroadcastRequest): Promise<BroadcastResult> {
    return this.post('/vitto-api/engine/internal/notifications/broadcast', request);
  }

  private async post<T>(path: string, body: unknown): Promise<T> {
    const response = await fetch(`${this.vittoApiUrl}${path}`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${this.internalToken}`,
      },
      body: JSON.stringify(body),
    });

    if (!response.ok) {
      const text = await response.text().catch(() => 'Unknown error');
      throw new Error(`Notification bridge failed (${response.status}): ${text}`);
    }

    return response.json() as T;
  }
}

This lives in messaging-core so any backend can import it. backend-api already imports MessagingCoreModule.

On backend-api: Usage example (billing)

// APIs/backend-api/src/app/billing/mcp-billing.controller.ts

constructor(
  private readonly notificationBridge: NotificationBridgeClientService,
) {}

// Replace the 3 commented-out stubs:
await this.notificationBridge.dispatch({
  userId,
  triggerName: 'billing.overage_charged',
  severity: 'warning',
  title: 'Overage Charges Applied',
  message: `You've been charged $${amount} for ${calls} API calls over your plan limit.`,
  customContent: {
    emailSubject: 'GoGrab Billing: Overage Charges',
    emailHtml: renderBillingEmail({ amount, calls, planName }),
  },
});

6. Schemas

NotificationLogDoc (NEW)

// APIs/backend-vitto-api/src/app/proactive/schemas/notification-log.schema.ts

@Schema({ timestamps: true, collection: 'notification-log' })
export class NotificationLogDoc extends Document {
  @Prop({ required: true, index: true })
  userId: string;

  @Prop({ index: true })
  broadcastId?: string;  // Groups logs from same broadcast

  @Prop({ required: true })
  triggerName: string;

  @Prop({ required: true, type: String })
  severity: 'info' | 'warning' | 'critical';

  @Prop({ required: true })
  title: string;

  @Prop({ required: true })
  message: string;

  @Prop({ type: [String] })
  channelsAttempted: string[];  // ['whatsapp', 'web']

  @Prop({ type: [String] })
  channelsDelivered: string[];  // ['whatsapp']  (web always succeeds)

  @Prop({ type: String, default: 'delivered' })
  status: 'delivered' | 'suppressed' | 'failed';

  @Prop()
  suppressionReason?: string;  // 'quiet_hours', 'frequency_cap', 'dedup', 'opt_out'

  @Prop({ default: false })
  readByUser: boolean;

  @Prop()
  readAt?: Date;

  @Prop({ type: Object })
  metadata?: Record<string, unknown>;

  @Prop({ type: Object })
  customContent?: Record<string, unknown>;  // Subset of what was sent (for audit, not full HTML)

  @Prop()
  source?: string;  // 'vitto-api' | 'backend-api:billing' | 'backend-api:deployment'

  // TTL: auto-delete logs older than 90 days
  @Prop({ type: Date, default: () => new Date(), index: { expires: '90d' } })
  expiresAt: Date;
}

export const NotificationLogSchema = SchemaFactory.createForClass(NotificationLogDoc);

NotificationLogSchema.index({ userId: 1, createdAt: -1 });
NotificationLogSchema.index({ broadcastId: 1 });
NotificationLogSchema.index({ userId: 1, readByUser: 1 });

UserNotificationPrefsDoc — Existing, minimal additions

// Add to existing schema (mind/schemas/user-notification-prefs.schema.ts)

/** S241: Multiple channel preferences (not just one preferred) */
@Prop({ type: [{ channel: String, channelId: String, enabled: { type: Boolean, default: true } }] })
channelPreferences?: Array<{
  channel: 'web' | 'slack' | 'discord' | 'whatsapp' | 'telegram' | 'email';
  channelId: string;
  enabled: boolean;
}>;

/** S241: Per-trigger channel override. Key = triggerName, value = channel type */
@Prop({ type: Object })
triggerChannelOverrides?: Record<string, string>;

Note: The existing preferredChannel + channelId fields remain as the primary default. The new channelPreferences array is optional and only used by resolveChannels() when present — backward compatible.

7. API Endpoints

Internal (service-to-service, @ServiceOnly guard)

External (user-facing, JWT auth)

Admin (executive/admin role guard)

8. Integration Points

backend-api billing stubs (immediate unblock)

Replace 3 commented-out notificationService.send() calls in mcp-billing.controller.ts with NotificationBridgeClientService.dispatch().

backend-api deployment completions (future)

BuildPipelineService can call notificationBridge.dispatch() when a deployment succeeds/fails.

Singularity flow engine (future)

LogicBlockExecutorService.executeNotificationAction() (currently a stub at line 583) calls NotificationBridgeClientService.dispatch() instead of logging. This gives every Singularity flow the ability to trigger real notifications.

backend-api internal endpoints (required)

Add two internal endpoints on backend-api for Vitto's resolveRecipients() to call:

// APIs/backend-api/src/app/internal/internal-spaces.controller.ts (extend existing)

@Get('spaces/:id/members')
@UseGuards(ServiceOnlyGuard)
async getSpaceMembers(@Param('id') spaceId: string) { ... }

// APIs/backend-api/src/app/internal/internal-users.controller.ts (extend existing)

@Get('users/by-roles')
@UseGuards(ServiceOnlyGuard)
async getUsersByRoles(@Query('roles') roles: string) { ... }

9. Key Files to Create/Modify

New files

Modified files

10. Risks & Mitigations

Critical Issues

C1. In-memory dedup/rate-limit state lost on restart (Confidence: Certain)

• dailyCounts and dedupCache in ProactiveDispatcherService are Map instances — not persisted.
• Risk: Duplicate notifications after process restart; frequency cap resets.
• Mitigation: For broadcast, deduplicate at the NotificationLogDoc level (query before dispatch). For single dispatch, accept the edge case — process restarts are rare and the 1h dedup window is short.
• Future: Move to Redis with TTL keys if multi-instance deployment happens.

C2. Broadcast to large audiences could overwhelm Twilio (Confidence: Certain)

• Twilio WhatsApp rate limit: ~80 msg/sec. A 10K-user broadcast at full speed = 125 seconds.
• Mitigation: concurrency cap (default 50) + 50ms delay between sends. For >500 recipients, auto-queue via ScheduledMessageQueueService. Log warns if broadcast exceeds 1000 recipients.

C3. Internal token security (Confidence: Certain)

• BACKEND_API_INTERNAL_TOKEN is a shared secret. If compromised, any service can dispatch as Vitto.
• Mitigation: Already the established pattern across the codebase (used by BackendAiClientService, BackendApiClientService). Accept for now. Future: mTLS or JWT-signed service tokens.

Non-Critical Improvements

N1. SSE-based notification inbox — The GET /notifications endpoint returns paginated results from MongoDB. For real-time badge updates, emit a conscience.notification.new event on the existing SSE stream. Frontend already subscribes to conscience.* events.

N2. Notification categories — The triggerName field currently stores arbitrary strings. Consider adding a category enum (billing, deployment, security, system, social) for UI filtering. Low priority — can be added when the frontend notification panel is built.

N3. Email unsubscribe link — For non-critical email notifications, include a one-click unsubscribe link per CAN-SPAM / GDPR requirements. The link calls PUT /notifications/preferences with the trigger disabled.

Assumptions to Verify

• A1: ServiceOnlyGuard validates Authorization: Bearer ${BACKEND_API_INTERNAL_TOKEN}. Verify the exact implementation before wiring the bridge.
• A2: internal-spaces.controller.ts and internal-users.controller.ts already exist and accept new endpoints. Verified: they do.
• A3: ScheduledMessageQueueService.schedule() can accept >500 items without performance issues. The collection has no index on scheduledAt — add one if needed.