Push Notification System

interface Notification {
  id: string;               // UUID, idempotency key
  userId: string;
  channel: 'push' | 'email' | 'sms';
  templateId: string;
  templateVars: Record<string, string>;
  status: 'pending' | 'queued' | 'delivering' | 'delivered' | 'failed' | 'dead_lettered';
  priority: 'high' | 'normal' | 'low';
  retryCount: number;       // 0–5
  createdAt: Date;
  sentAt?: Date;
  deliveredAt?: Date;
  failureReason?: string;
  // Delivery-channel-specific
  deviceToken?: string;     // push
  toAddress?: string;       // email
  phoneNumber?: string;     // sms
}

interface UserPreferences {
  userId: string;
  deviceTokens: Array<{ token: string; platform: 'ios' | 'android'; updatedAt: Date }>;
  email: string;
  phoneNumber?: string;
  optOuts: {
    push: boolean;
    email: boolean;
    sms: boolean;
    categories: string[];   // e.g. ['marketing', 'promotions']
  };
  quietHours?: {
    start: string;          // '22:00'
    end: string;            // '08:00'
    timezone: string;       // 'America/New_York'
  };
  frequencyCaps: {
    push: { max: number; windowHours: number };
    email: { max: number; windowHours: number };
    sms: { max: number; windowHours: number };
  };
}

interface NotificationMessage {
  notificationId: string;   // idempotency key
  userId: string;
  channel: 'push' | 'email' | 'sms';
  templateId: string;
  templateVars: Record<string, string>;
  priority: 'high' | 'normal' | 'low';
  deviceToken?: string;
  toAddress?: string;
  phoneNumber?: string;
  enqueuedAt: number;       // unix ms
  attempt: number;          // 1-based retry count
}

{
  "userIds": ["u1", "u2", "u3"],
  "templateId": "order_confirmed",
  "templateVars": { "orderId": "ORD-9821", "total": "$49.99" },
  "channels": ["push", "email"],
  "priority": "high",
  "idempotencyKey": "order-9821-confirmation"
}

{
  "batchId": "batch_abc123",
  "queued": 6,
  "skipped": 1,
  "message": "1 user opted out of push notifications"
}

{
  "id": "notif_xyz",
  "status": "delivered",
  "channel": "push",
  "deliveredAt": "2024-06-27T10:42:13Z"
}

interface NotificationService {
  send(req: SendNotificationRequest): Promise<BatchResult>
  fanOut(notificationId: string, userIds: string[], channels: Channel[]): Promise<void>
  checkPreferences(userId: string): Promise<UserPreferences>
  isQuietHours(prefs: UserPreferences): boolean
  isFrequencyCapped(userId: string, channel: Channel): Promise<boolean>
}

interface ChannelAdapter {
  deliver(message: NotificationMessage): Promise<DeliveryResult>
}

interface DeliveryResult {
  success: boolean;
  provider: string;
  providerId?: string;     // FCM message ID, SES message ID, etc.
  error?: string;
  invalidToken?: boolean;  // for push: clean up device token
}

// --- Domain Types ---

type Channel = 'push' | 'email' | 'sms';
type NotifStatus = 'pending' | 'queued' | 'delivering' | 'delivered' | 'failed' | 'dead_lettered';

interface SendNotificationRequest {
  userIds: string[];
  templateId: string;
  templateVars: Record<string, string>;
  channels: Channel[];
  priority: 'high' | 'normal' | 'low';
  idempotencyKey: string;
}

interface BatchResult {
  batchId: string;
  queued: number;
  skipped: number;
  reasons: Array<{ userId: string; reason: string }>;
}

// --- NotificationService ---

class NotificationService {
  constructor(
    private readonly prefsDb: UserPrefsRepository,
    private readonly notifDb: NotificationRepository,
    private readonly kafka: KafkaProducer,
    private readonly dedup: DeduplicationStore,
  ) {}

  async send(req: SendNotificationRequest): Promise<BatchResult> {
    // Idempotency check
    if (await this.dedup.exists(req.idempotencyKey)) {
      return { batchId: req.idempotencyKey, queued: 0, skipped: req.userIds.length, reasons: [] };
    }

    const result: BatchResult = { batchId: generateId(), queued: 0, skipped: 0, reasons: [] };

    for (const userId of req.userIds) {
      const prefs = await this.prefsDb.findByUserId(userId);
      const allowedChannels = req.channels.filter(ch => this.isChannelAllowed(ch, prefs, userId));

      for (const channel of allowedChannels) {
        const notifId = generateId();
        await this.notifDb.insert({ id: notifId, userId, channel, status: 'queued', ...req });
        await this.kafka.produce(`notif.${channel}`, { notificationId: notifId, userId, channel, ...req, attempt: 1 });
        result.queued++;
      }

      const skippedChannels = req.channels.length - allowedChannels.length;
      if (skippedChannels > 0) {
        result.skipped += skippedChannels;
        result.reasons.push({ userId, reason: 'opted_out_or_quiet_hours' });
      }
    }

    await this.dedup.set(req.idempotencyKey, 86400);
    return result;
  }

  private isChannelAllowed(channel: Channel, prefs: UserPreferences, userId: string): boolean {
    if (prefs.optOuts[channel]) return false;
    if (prefs.quietHours && this.isQuietHours(prefs.quietHours)) return false;
    return true;
  }

  private isQuietHours(quietHours: UserPreferences['quietHours']): boolean {
    // Compare current time in user's timezone against quiet window
    const now = new Date().toLocaleTimeString('en-US', { hour12: false, timeZone: quietHours!.timezone });
    return now >= quietHours!.start || now <= quietHours!.end;
  }
}

// --- Channel Adapter Interface ---

interface ChannelAdapter {
  channel: Channel;
  deliver(message: NotificationMessage): Promise<DeliveryResult>;
}

// --- Push Adapter ---

class PushAdapter implements ChannelAdapter {
  channel: Channel = 'push';

  constructor(
    private readonly fcmClient: FCMClient,
    private readonly apnsClient: APNsClient,
    private readonly prefsDb: UserPrefsRepository,
  ) {}

  async deliver(msg: NotificationMessage): Promise<DeliveryResult> {
    const { deviceToken, templateId, templateVars } = msg;
    if (!deviceToken) return { success: false, provider: 'none', error: 'no_device_token' };

    const platform = deviceToken.startsWith('apns:') ? 'ios' : 'android';
    const client = platform === 'ios' ? this.apnsClient : this.fcmClient;

    try {
      const result = await client.send({ token: deviceToken, ...templateVars });
      return { success: true, provider: platform, providerId: result.messageId };
    } catch (err: any) {
      const isInvalidToken = err.code === 'InvalidRegistration' || err.code === 'Unregistered';
      if (isInvalidToken) {
        // Clean up stale token
        await this.prefsDb.removeDeviceToken(msg.userId, deviceToken);
      }
      return { success: false, provider: platform, error: err.code, invalidToken: isInvalidToken };
    }
  }
}

// --- Generic Worker (wraps any ChannelAdapter) ---

class NotificationWorker {
  constructor(
    private readonly adapter: ChannelAdapter,
    private readonly dedup: DeduplicationStore,
    private readonly notifDb: NotificationRepository,
    private readonly dlqProducer: KafkaProducer,
    private readonly MAX_RETRIES = 5,
  ) {}

  async process(msg: NotificationMessage): Promise<void> {
    const dedupKey = `notif:${msg.notificationId}:${msg.channel}:delivered`;
    if (await this.dedup.exists(dedupKey)) return; // already delivered

    await this.notifDb.updateStatus(msg.notificationId, 'delivering');
    const result = await this.adapter.deliver(msg);

    if (result.success) {
      await this.notifDb.updateStatus(msg.notificationId, 'delivered', { deliveredAt: new Date(), providerId: result.providerId });
      await this.dedup.set(dedupKey, 86400);
    } else if (result.invalidToken) {
      await this.notifDb.updateStatus(msg.notificationId, 'failed', { failureReason: 'invalid_token' });
      // ACK — no retry for invalid tokens
    } else if (msg.attempt >= this.MAX_RETRIES) {
      await this.notifDb.updateStatus(msg.notificationId, 'dead_lettered');
      await this.dlqProducer.produce('notif.dlq', { ...msg, finalError: result.error });
    } else {
      await this.notifDb.updateStatus(msg.notificationId, 'failed', { retryCount: msg.attempt });
      throw new Error(`delivery_failed:${result.error}`); // NACK → Kafka retry
    }
  }
}