Assist_Design/apps/bff/src/integrations/salesforce/events/pubsub.subscriber.ts

import { Injectable, OnModuleInit, OnModuleDestroy, Inject } from "@nestjs/common";
import { ConfigService } from "@nestjs/config";
import { Logger } from "nestjs-pino";
import PubSubApiClientPkg from "salesforce-pubsub-api-client";
import { SalesforceConnection } from "../services/salesforce-connection.service";
import { ProvisioningQueueService } from "@bff/modules/orders/queue/provisioning.queue";
import { CacheService } from "@bff/infra/cache/cache.service";
import {
  replayKey as sfReplayKey,
  statusKey as sfStatusKey,
  latestSeenKey as sfLatestSeenKey,
} from "./event-keys.util";
import type {
  SalesforcePubSubEvent,
  SalesforcePubSubError,
  SalesforcePubSubSubscription,
  SalesforcePubSubCallbackType,
} from "../types/pubsub-events.types";

type SubscribeCallback = (
  subscription: SalesforcePubSubSubscription,
  callbackType: SalesforcePubSubCallbackType,
  data: SalesforcePubSubEvent | SalesforcePubSubError | unknown
) => void | Promise<void>;

interface PubSubClient {
  connect(): Promise<void>;
  subscribe(topic: string, cb: SubscribeCallback, numRequested?: number): Promise<void>;
  subscribeFromReplayId(
    topic: string,
    cb: SubscribeCallback,
    numRequested: number | null,
    replayId: number
  ): Promise<void>;
  subscribeFromEarliestEvent(
    topic: string,
    cb: SubscribeCallback,
    numRequested?: number
  ): Promise<void>;
  requestAdditionalEvents(topic: string, numRequested: number): Promise<void>;
  close(): Promise<void>;
}

type PubSubCtor = new (opts: {
  authType: string;
  accessToken: string;
  instanceUrl: string;
  pubSubEndpoint: string;
}) => PubSubClient;

@Injectable()
export class SalesforcePubSubSubscriber implements OnModuleInit, OnModuleDestroy {
  private client: PubSubClient | null = null;
  private channel!: string;
  private replayCorruptionRecovered = false;
  private subscribeCallback!: SubscribeCallback;

  constructor(
    private readonly config: ConfigService,
    private readonly sfConn: SalesforceConnection,
    private readonly provisioningQueue: ProvisioningQueueService,
    private readonly cache: CacheService,
    @Inject(Logger) private readonly logger: Logger
  ) {}

  async onModuleInit(): Promise<void> {
    const enabled = this.config.get("SF_EVENTS_ENABLED", "false") === "true";
    if (!enabled) {
      this.logger.log("Salesforce Pub/Sub subscriber disabled", { enabled });
      return;
    }

    this.channel = this.config.get<string>(
      "SF_PROVISION_EVENT_CHANNEL",
      "/event/Order_Fulfilment_Requested__e"
    );

    try {
      await this.sfConn.connect();
      const accessToken = this.sfConn.getAccessToken();
      const instanceUrl = this.sfConn.getInstanceUrl();
      if (!accessToken || !instanceUrl) {
        throw new Error("Salesforce access token || instance URL unavailable");
      }

      const endpoint = this.config.get<string>(
        "SF_PUBSUB_ENDPOINT",
        "api.pubsub.salesforce.com:7443"
      );

      const maybeCtor: unknown =
        (PubSubApiClientPkg as { default?: unknown })?.default ?? (PubSubApiClientPkg as unknown);
      const Ctor = maybeCtor as PubSubCtor;
      this.client = new Ctor({
        authType: "user-supplied",
        accessToken,
        instanceUrl,
        pubSubEndpoint: endpoint,
      });
      await this.client.connect();
      if (!this.client) throw new Error("Pub/Sub client not initialized after connect");
      const client = this.client;

      const _replayKey = sfReplayKey(this.channel);
      const _replayMode = this.config.get<string>("SF_EVENTS_REPLAY", "LATEST");
      const numRequested = Number(this.config.get("SF_PUBSUB_NUM_REQUESTED", "50")) || 50;
      const maxQueue = Number(this.config.get("SF_PUBSUB_QUEUE_MAX", "100")) || 100;

      await this.cache.set(sfStatusKey(this.channel), {
        status: "connecting",
        since: Date.now(),
      });

      this.subscribeCallback = async (subscription, callbackType, data) => {
        try {
          const argTypes = [typeof subscription, typeof callbackType, typeof data];
          const type = callbackType;
          const typeNorm = String(type || "").toLowerCase();
          const topic = subscription.topicName || this.channel;

          if (typeNorm === "data" || typeNorm === "event") {
            const event = data as SalesforcePubSubEvent;
            // Basic breadcrumb to confirm we are handling data callbacks
            this.logger.debug("SF Pub/Sub data callback received", {
              topic,
              argTypes,
              hasPayload: Boolean(event?.payload),
            });
            const payload = event?.payload;

            // Only check parsed payload
            const orderIdVal = payload?.["OrderId__c"] ?? payload?.["OrderId"];
            const orderId = typeof orderIdVal === "string" ? orderIdVal : undefined;
            if (!orderId) {
              this.logger.warn("Pub/Sub event missing OrderId__c; skipping", {
                argTypes,
                topic,
                payloadKeys: payload ? Object.keys(payload) : [],
              });
              const depth = await this.provisioningQueue.depth();
              if (depth < maxQueue) {
                await client.requestAdditionalEvents(topic, 1);
              }
              return;
            }

            const replayVal = (event as { replayId?: unknown })?.replayId;
            const idempotencyKey =
              typeof replayVal === "number" || typeof replayVal === "string"
                ? String(replayVal)
                : String(Date.now());
            const pubsubReplayId = typeof replayVal === "number" ? replayVal : undefined;

            await this.provisioningQueue.enqueue({
              sfOrderId: orderId,
              idempotencyKey,
              pubsubReplayId,
            });
            this.logger.log("Enqueued provisioning job from SF event", {
              sfOrderId: orderId,
              replayId: pubsubReplayId,
              topic,
            });

            // Do not request more here; rely on 'lastevent' to top-up
          } else if (typeNorm === "lastevent") {
            const depth = await this.provisioningQueue.depth();
            const available = Math.max(0, maxQueue - depth);
            const desired = Math.max(0, Math.min(numRequested, available));
            if (desired > 0) {
              await client.requestAdditionalEvents(topic, desired);
            }
          } else if (typeNorm === "grpckeepalive") {
            const latestVal = (data as { latestReplayId?: unknown })?.latestReplayId;
            const latest = typeof latestVal === "number" ? latestVal : undefined;
            if (typeof latest === "number") {
              await this.cache.set(sfLatestSeenKey(this.channel), {
                id: String(latest),
                at: Date.now(),
              });
            }
          } else if (typeNorm === "grpcstatus" || typeNorm === "end") {
            // No-op; informational
          } else if (typeNorm === "error") {
            this.logger.warn("SF Pub/Sub stream error", { topic, data });
            try {
              // Detect replay id corruption and auto-recover once by clearing the cursor and resubscribing
              const errorData = data as SalesforcePubSubError;
              const details = errorData.details || "";
              const metadata = errorData.metadata || {};
              const errorCodes = Array.isArray(metadata["error-code"])
                ? metadata["error-code"]
                : [];
              const hasCorruptionCode = errorCodes.some(code =>
                String(code).includes("replayid.corrupted")
              );
              const mentionsReplayValidation = /Replay ID validation failed/i.test(details);

              if (
                (hasCorruptionCode || mentionsReplayValidation) &&
                !this.replayCorruptionRecovered
              ) {
                this.replayCorruptionRecovered = true;
                const key = sfReplayKey(this.channel);
                await this.cache.del(key);
                this.logger.warn(
                  "Cleared invalid Salesforce Pub/Sub replay cursor; retrying subscription",
                  {
                    channel: this.channel,
                    key,
                  }
                );
                await this.cache.set(sfStatusKey(this.channel), {
                  status: "reconnecting",
                  since: Date.now(),
                });
                // Try re-subscribing without the invalid cursor
                await this.subscribeWithPolicy();
              }
            } catch (recoveryErr) {
              this.logger.warn("SF Pub/Sub replay corruption auto-recovery failed", {
                error: recoveryErr instanceof Error ? recoveryErr.message : String(recoveryErr),
              });
            }
          } else {
            // Unknown callback type: log once with minimal context
            const maybeEvent = data as SalesforcePubSubEvent | undefined;
            const hasPayload = Boolean(maybeEvent?.payload);
            this.logger.debug("SF Pub/Sub callback ignored (unknown type)", {
              type,
              topic,
              argTypes,
              hasPayload,
            });
          }
        } catch (err) {
          this.logger.error("Pub/Sub subscribe callback failed", {
            error: err instanceof Error ? err.message : String(err),
          });
        }
      };

      await this.subscribeWithPolicy();
    } catch (error) {
      this.logger.error("Salesforce Pub/Sub subscription failed", {
        error: error instanceof Error ? error.message : String(error),
      });
      try {
        await this.cache.set(sfStatusKey(this.channel || "/event/OrderProvisionRequested__e"), {
          status: "disconnected",
          since: Date.now(),
        });
      } catch (cacheErr) {
        this.logger.warn("Failed to set SF Pub/Sub disconnected status", {
          error: cacheErr instanceof Error ? cacheErr.message : String(cacheErr),
        });
      }
    }
  }

  async onModuleDestroy(): Promise<void> {
    try {
      if (this.client) {
        await this.client.close();
        this.client = null;
      }
      await this.cache.set(sfStatusKey(this.channel), {
        status: "disconnected",
        since: Date.now(),
      });
    } catch (error) {
      this.logger.warn("Error closing Salesforce Pub/Sub client", {
        error: error instanceof Error ? error.message : String(error),
      });
    }
  }

  private async subscribeWithPolicy(): Promise<void> {
    if (!this.client) throw new Error("Pub/Sub client not initialized");
    if (!this.subscribeCallback) throw new Error("Subscribe callback not initialized");

    const _replayMode = this.config.get<string>("SF_EVENTS_REPLAY", "LATEST");
    const numRequested = Number(this.config.get("SF_PUBSUB_NUM_REQUESTED", "50")) || 50;
    const _replayKey = sfReplayKey(this.channel);
    const storedReplay = _replayMode !== "ALL" ? await this.cache.get<string>(_replayKey) : null;

    if (storedReplay && _replayMode !== "ALL") {
      await this.client.subscribeFromReplayId(
        this.channel,
        this.subscribeCallback,
        numRequested,
        Number(storedReplay)
      );
    } else if (_replayMode === "ALL") {
      await this.client.subscribeFromEarliestEvent(
        this.channel,
        this.subscribeCallback,
        numRequested
      );
    } else {
      await this.client.subscribe(this.channel, this.subscribeCallback, numRequested);
    }

    await this.cache.set(sfStatusKey(this.channel), {
      status: "connected",
      since: Date.now(),
    });
    this.logger.log("Salesforce Pub/Sub subscription active", { channel: this.channel });
  }

  // keys moved to shared util
}