import { Injectable, Inject, OnModuleInit, OnModuleDestroy } 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 { CatalogCacheService } from "@bff/modules/catalog/services/catalog-cache.service";

type PubSubCallback = (
  subscription: { topicName?: string },
  callbackType: string,
  data: unknown
) => void | Promise<void>;

interface PubSubClient {
  connect(): Promise<void>;
  subscribe(topic: string, cb: PubSubCallback, numRequested?: number): Promise<void>;
  close(): Promise<void>;
}

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

@Injectable()
export class CatalogCdcSubscriber implements OnModuleInit, OnModuleDestroy {
  private client: PubSubClient | null = null;
  private pubSubCtor: PubSubCtor | null = null;
  private productChannel: string | null = null;
  private pricebookChannel: string | null = null;
  private accountChannel: string | null = null;
  private readonly numRequested: number;

  constructor(
    private readonly config: ConfigService,
    private readonly sfConnection: SalesforceConnection,
    private readonly catalogCache: CatalogCacheService,
    @Inject(Logger) private readonly logger: Logger
  ) {
    this.numRequested = this.resolveNumRequested();
  }

  async onModuleInit(): Promise<void> {
    const productChannel =
      this.config.get<string>("SF_CATALOG_PRODUCT_CDC_CHANNEL")?.trim() ||
      "/data/Product2ChangeEvent";
    const pricebookChannel =
      this.config.get<string>("SF_CATALOG_PRICEBOOKENTRY_CDC_CHANNEL")?.trim() ||
      "/data/PricebookEntryChangeEvent";
    const accountChannel = this.config.get<string>("SF_ACCOUNT_ELIGIBILITY_CHANNEL")?.trim();

    try {
      const client = await this.ensureClient();

      this.productChannel = productChannel;
      await client.subscribe(
        productChannel,
        this.handleProductEvent.bind(this, productChannel),
        this.numRequested
      );
      this.logger.log("Subscribed to Product2 CDC channel", { productChannel });

      this.pricebookChannel = pricebookChannel;
      await client.subscribe(
        pricebookChannel,
        this.handlePricebookEvent.bind(this, pricebookChannel),
        this.numRequested
      );
      this.logger.log("Subscribed to PricebookEntry CDC channel", { pricebookChannel });

      if (accountChannel) {
        this.accountChannel = accountChannel;
        await client.subscribe(
          accountChannel,
          this.handleAccountEvent.bind(this, accountChannel),
          this.numRequested
        );
        this.logger.log("Subscribed to account eligibility channel", { accountChannel });
      }
    } catch (error) {
      this.logger.warn("Failed to initialize catalog CDC subscriber", {
        error: error instanceof Error ? error.message : String(error),
      });
    }
  }

  async onModuleDestroy(): Promise<void> {
    if (!this.client) return;
    try {
      await this.client.close();
    } catch (error) {
      this.logger.warn("Failed to close Salesforce CDC subscriber cleanly", {
        error: error instanceof Error ? error.message : String(error),
      });
    }
  }

  private async ensureClient(): Promise<PubSubClient> {
    if (this.client) {
      return this.client;
    }

    const ctor = await this.loadPubSubCtor();

    await this.sfConnection.connect();
    const accessToken = this.sfConnection.getAccessToken();
    const instanceUrl = this.sfConnection.getInstanceUrl();

    if (!accessToken || !instanceUrl) {
      throw new Error("Salesforce access token or instance URL missing for CDC subscriber");
    }

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

    const client = new ctor({
      authType: "user-supplied",
      accessToken,
      instanceUrl,
      pubSubEndpoint,
    });

    await client.connect();
    this.client = client;
    return client;
  }

  private loadPubSubCtor(): Promise<PubSubCtor> {
    if (!this.pubSubCtor) {
      const maybeCtor = (PubSubApiClientPkg as unknown as PubSubCtor) ?? null;
      const maybeDefault = (PubSubApiClientPkg as { default?: PubSubCtor }).default ?? null;
      const ctor = typeof maybeCtor === "function" ? maybeCtor : maybeDefault;
      if (!ctor) {
        throw new Error("Failed to load Salesforce Pub/Sub client constructor");
      }
      this.pubSubCtor = ctor;
    }
    return Promise.resolve(this.pubSubCtor);
  }

  private resolveNumRequested(): number {
    const raw = this.config.get<string>("SF_PUBSUB_NUM_REQUESTED") ?? "25";
    const parsed = Number.parseInt(raw, 10);
    if (!Number.isFinite(parsed) || parsed <= 0) {
      this.logger.warn("Invalid SF_PUBSUB_NUM_REQUESTED value; defaulting to 25", {
        rawValue: raw,
      });
      return 25;
    }
    return parsed;
  }

  private async handleProductEvent(
    channel: string,
    subscription: { topicName?: string },
    callbackType: string,
    data: unknown
  ): Promise<void> {
    if (!this.isDataCallback(callbackType)) return;
    const payload = this.extractPayload(data);
    const productIds = this.extractRecordIds(payload);

    this.logger.log("Product2 CDC event received", {
      channel,
      topicName: subscription.topicName,
      productIds,
    });

    const invalidated = await this.catalogCache.invalidateProducts(productIds);

    if (!invalidated) {
      this.logger.debug(
        "No catalog cache entries were linked to product IDs; falling back to full invalidation",
        {
          channel,
          productIds,
        }
      );
      await this.invalidateAllCatalogs();
    }
  }

  private async handlePricebookEvent(
    channel: string,
    subscription: { topicName?: string },
    callbackType: string,
    data: unknown
  ): Promise<void> {
    if (!this.isDataCallback(callbackType)) return;
    const payload = this.extractPayload(data);
    const pricebookId = this.extractStringField(payload, ["PricebookId", "Pricebook2Id"]);

    const portalPricebookId = this.config.get<string>("PORTAL_PRICEBOOK_ID");
    if (portalPricebookId && pricebookId && pricebookId !== portalPricebookId) {
      this.logger.debug("Ignoring pricebook event for non-portal pricebook", {
        channel,
        pricebookId,
      });
      return;
    }

    const productId = this.extractStringField(payload, ["Product2Id", "ProductId"]);

    this.logger.log("PricebookEntry CDC event received", {
      channel,
      pricebookId,
      productId,
    });

    const invalidated = await this.catalogCache.invalidateProducts(productId ? [productId] : []);

    if (!invalidated) {
      this.logger.debug(
        "No catalog cache entries mapped to product from pricebook event; performing full invalidation",
        {
          channel,
          pricebookId,
          productId,
        }
      );
      await this.invalidateAllCatalogs();
    }
  }

  private async handleAccountEvent(
    channel: string,
    subscription: { topicName?: string },
    callbackType: string,
    data: unknown
  ): Promise<void> {
    if (!this.isDataCallback(callbackType)) return;
    const payload = this.extractPayload(data);
    const accountId = this.extractStringField(payload, ["AccountId__c", "AccountId"]);
    const eligibility = this.extractStringField(payload, [
      "Internet_Eligibility__c",
      "InternetEligibility__c",
    ]);

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

    this.logger.log("Account eligibility event received", {
      channel,
      accountId,
      eligibility,
    });

    await this.catalogCache.invalidateEligibility(accountId);
    await this.catalogCache.setEligibilityValue(accountId, eligibility ?? null);
  }

  private async invalidateAllCatalogs(): Promise<void> {
    try {
      await this.catalogCache.invalidateAllCatalogs();
    } catch (error) {
      this.logger.warn("Failed to invalidate catalog caches", {
        error: error instanceof Error ? error.message : String(error),
      });
    }
  }

  private isDataCallback(callbackType: string): boolean {
    const normalized = String(callbackType || "").toLowerCase();
    return normalized === "data" || normalized === "event";
  }

  private extractPayload(data: unknown): Record<string, unknown> | undefined {
    if (!data || typeof data !== "object") {
      return undefined;
    }

    const candidate = data as { payload?: unknown };
    if (candidate.payload && typeof candidate.payload === "object") {
      return candidate.payload as Record<string, unknown>;
    }

    return data as Record<string, unknown>;
  }

  private extractStringField(
    payload: Record<string, unknown> | undefined,
    fieldNames: string[]
  ): string | undefined {
    if (!payload) return undefined;
    for (const field of fieldNames) {
      const value = payload[field];
      if (typeof value === "string") {
        const trimmed = value.trim();
        if (trimmed.length > 0) {
          return trimmed;
        }
      }
    }
    return undefined;
  }

  private extractRecordIds(payload: Record<string, unknown> | undefined): string[] {
    if (!payload) {
      return [];
    }

    const header = this.extractChangeEventHeader(payload);
    const ids = header?.recordIds ?? [];
    if (Array.isArray(ids)) {
      return ids.filter((id): id is string => typeof id === "string" && id.trim().length > 0);
    }

    return [];
  }

  private extractChangeEventHeader(
    payload: Record<string, unknown>
  ): { recordIds?: unknown; changedFields?: unknown } | undefined {
    const header = payload["ChangeEventHeader"];
    if (header && typeof header === "object") {
      return header as { recordIds?: unknown; changedFields?: unknown };
    }
    return undefined;
  }
}
