Assist_Design/apps/bff/src/integrations/salesforce/events/account-events.subscriber.ts

/**
 * Salesforce Account Platform Events Subscriber
 *
 * Handles real-time Account Platform Events for:
 * - Eligibility status changes
 * - Verification status changes
 *
 * When events are received:
 * 1. Invalidate Redis caches (eligibility + verification)
 * 2. Send SSE to connected portal clients
 * 3. Create in-app notifications (for final status changes)
 *
 * @see docs/integrations/salesforce/platform-events.md
 */

import { Injectable, Inject, Optional } from "@nestjs/common";
import type { OnModuleInit } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import { Logger } from "nestjs-pino";
import {
  PubSubClientService,
  isDataCallback,
  extractPayload,
  extractStringField,
} from "./shared/index.js";
import { ServicesCacheService } from "@bff/modules/services/application/services-cache.service.js";
import { RealtimeService } from "@bff/infra/realtime/realtime.service.js";
import { AccountNotificationHandler } from "@bff/modules/notifications/account-cdc-listener.service.js";

@Injectable()
export class AccountEventsSubscriber implements OnModuleInit {
  constructor(
    private readonly config: ConfigService,
    private readonly pubSubClient: PubSubClientService,
    private readonly servicesCache: ServicesCacheService,
    private readonly realtime: RealtimeService,
    @Inject(Logger) private readonly logger: Logger,
    @Optional() private readonly accountNotificationHandler?: AccountNotificationHandler
  ) {}

  async onModuleInit(): Promise<void> {
    const channel = this.config.get<string>("SF_ACCOUNT_EVENT_CHANNEL")?.trim();

    if (!channel) {
      this.logger.warn("SF_ACCOUNT_EVENT_CHANNEL not configured; skipping account events");
      return;
    }

    try {
      await this.pubSubClient.subscribe(
        channel,
        this.handleAccountEvent.bind(this, channel),
        "account-platform-event"
      );
    } catch (error) {
      this.logger.warn("Failed to subscribe to Account Platform Events", {
        channel,
        error: error instanceof Error ? error.message : String(error),
      });
    }
  }

  /**
   * Handle Account Platform Events (eligibility + verification updates)
   *
   * Salesforce Flow fires this event when these fields change:
   * - Internet_Eligibility__c, Internet_Eligibility_Status__c
   * - Id_Verification_Status__c, Id_Verification_Rejection_Message__c
   *
   * Platform Event fields:
   * - Account_Id__c (required)
   * - Eligibility_Status__c (only if changed)
   * - Verification_Status__c (only if changed)
   * - Rejection_Message__c (when rejected)
   *
   * Actions:
   * - ALWAYS invalidate both caches
   * - ALWAYS send SSE to portal
   * - Create notification only for final states (eligible/verified/rejected)
   */
  private async handleAccountEvent(
    channel: string,
    subscription: { topicName?: string },
    callbackType: string,
    data: unknown
  ): Promise<void> {
    if (!isDataCallback(callbackType)) return;

    const payload = extractPayload(data);
    const accountId = extractStringField(payload, ["Account_Id__c"]);

    if (!accountId) {
      this.logger.warn("Account event missing Account_Id__c", { channel });
      return;
    }

    const eligibilityStatus = extractStringField(payload, ["Eligibility_Status__c"]);
    const verificationStatus = extractStringField(payload, ["Verification_Status__c"]);
    const rejectionMessage = extractStringField(payload, ["Rejection_Message__c"]);

    this.logger.log("Account platform event received", {
      channel,
      accountIdTail: accountId.slice(-4),
      hasEligibilityStatus: eligibilityStatus !== undefined,
      hasVerificationStatus: verificationStatus !== undefined,
    });

    // ALWAYS invalidate caches
    await this.servicesCache.invalidateEligibility(accountId);
    await this.servicesCache.invalidateVerification(accountId);

    // ALWAYS notify portal to refetch
    this.realtime.publish(`account:sf:${accountId}`, "account.updated", {
      timestamp: new Date().toISOString(),
    });

    // Create notifications for status changes (handler filters to final states)
    if (this.accountNotificationHandler && (eligibilityStatus || verificationStatus)) {
      void this.accountNotificationHandler.processAccountEvent({
        accountId,
        eligibilityStatus,
        eligibilityValue: undefined,
        verificationStatus,
        verificationRejectionMessage: rejectionMessage,
      });
    }
  }
}