import { KafkaSerializer, KafkaService } from "./kafka.service";
import MetricService from "./metric.service";
import { config } from "../utils/config";

import logger from "../utils/logger";
import { ChoiceRequest, RequestInventorySource, RequestType as RequestTypeModel } from "../models";
import { FeatureConfigService } from "./FeatureConfig.service";
import IDS from "../protos/kafka-schema-ids/DeliveryMonitoring.json";
import { ogury } from "../protos/protos";
import IDeliveryMonitoringMessage = ogury.delivery.monitoring.IDeliveryMonitoringMessage;
import Service = ogury.delivery.monitoring.DeliveryMonitoringMessage.Service;
import DeliveryMonitoringMessage = ogury.delivery.monitoring.DeliveryMonitoringMessage;
import InventorySource = ogury.delivery.monitoring.DeliveryMonitoringMessage.InventorySource;
import SourceType = ogury.delivery.monitoring.DeliveryMonitoringMessage.SourceType;
import RequestType = ogury.delivery.monitoring.DeliveryMonitoringMessage.RequestType;

type SchemaIds = { [key: string]: { [key: string]: { [key: string]: number } } };

const SCHEMA_ID = getSchemaId();

const MESSAGE_COUNTER = "delivery_monitoring_kafka_total";

class DeliveryMonitoringService {
  private static instance: DeliveryMonitoringService;
  private kafkaConnector: KafkaService<IDeliveryMonitoringMessage>;
  private metricService: MetricService;

  constructor() {
    this.kafkaConnector = new KafkaService(
      config.BROKER_LIST_CLUSTER_B,
      "DeliveryMonitoring",
      deliveryProtoSerializer
    );
    this.metricService = MetricService.getInstance();
    this.metricService.addCounter(MESSAGE_COUNTER, "number of DeliveryMonitoring message sent", [
      "service",
      "error",
    ]);

    if (SCHEMA_ID) {
      logger.info(`[DELIVERY MONITORING] schema ID = ${SCHEMA_ID}`);
    } else {
      logger.error(new Error(`[DELIVERY MONITORING] failed to get schema ID`));
    }
  }

  async health(): Promise<boolean> {
    return this.kafkaConnector.health();
  }

  public static getInstance(): DeliveryMonitoringService {
    return this.instance || (this.instance = new DeliveryMonitoringService());
  }

  public sendMessage(request: ChoiceRequest): void {
    let error = false;
    try {
      const deliveryMonitoringMessage: IDeliveryMonitoringMessage = {
        request_id: request.request_id ?? "",
        request_time: new Date().getTime(),
        request_type: getRequestType(request),
        ssp: request.context?.ssp ?? "",
        app_bundle_id: request.app?.bundle_id ?? "",
        ad_unit_id: request.app?.ad_unit_id ?? "",
        geoip_code: request.context?.country ?? "",
        service: Service.MS_CHOICES,
        inventory_source: getInventorySource(request),
        source_type: getSourceType(request),
        steps_statuses: request.stepsStatuses,
      };

      logger.debug(() => ({
        message: "DeliveryMonitoring topic message",
        deliveryMonitoringMessage,
      }));
      this.kafkaConnector.send(deliveryMonitoringMessage, deliveryMonitoringMessage.request_id);
    } catch (e: any) {
      error = true;
      logger.debug(() => `[KAFKA] failed to create Delivery kafka log : ${e.message}`);
    } finally {
      this.metricService.incCounter(MESSAGE_COUNTER, 1, {
        service: "ms-choices",
        error: error.toString(),
      });
    }
  }
}

function getInventorySource(request: ChoiceRequest): InventorySource {
  switch (request.inventory_source) {
    case RequestInventorySource.INTERNAL:
      return InventorySource.INTERNAL;
    case RequestInventorySource.HEADER_BIDDING:
      return InventorySource.HEADER_BIDDING;
    case RequestInventorySource.EXTERNAL:
      return InventorySource.EXTERNAL;
  }
}

function getSourceType(request: ChoiceRequest): SourceType {
  if (request.environment_type === "web") {
    return SourceType.WEB;
  }
  return SourceType.IN_APP;
}

function getRequestType(request: ChoiceRequest): RequestType {
  if (request.type === RequestTypeModel.AFTER_AD) {
    return RequestType.AFTER_AD;
  }
  return RequestType.AD;
}

export function getSchemaId(): any {
  const ids: SchemaIds = IDS;
  const env = config.OGURY_ENV;
  const region = config.AWS_REGION;
  const az = region === "eu-west-1" ? config.AWS_AZ : "a";
  return ids[env]?.[region]?.[az] ?? 0;
}

const deliveryProtoSerializer: KafkaSerializer<IDeliveryMonitoringMessage> = {
  // https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html#wire-format
  to_kafka: (deliveryMessage: IDeliveryMonitoringMessage) => {
    const magicByte = Buffer.alloc(1);
    const schemaId = Buffer.alloc(4);
    schemaId.writeInt32BE(SCHEMA_ID, 0);
    const messageIndex = Buffer.alloc(1);
    const payload = Buffer.from(DeliveryMonitoringMessage.encode(deliveryMessage).finish());
    return Buffer.concat([magicByte, schemaId, messageIndex, payload]);
  },
};

export default DeliveryMonitoringService;