import S3Service from "../s3.service";
import MetricService, { BucketSet } from "../metric.service";
import { Options } from "csv-parser";
import { sleep } from "../../utils/helper";
import { ChoiceRequest } from "../../models";
import logger from "../../utils/logger";
import { config, isTest } from "../../utils/config";
import { BucketMapping, parseBucketMapping } from "./abstractScoreHashMap.service";
import { CampaignCreative } from "../../steps/ordering/ordering.model";
import constants from "../../utils/constants";

const LOG_ID = "[AUDIENCE_HASHMAP_SERVICE]";
const APPLICATIONS_KEY_S3 = "apps";
const CAMPAIGNS_KEY_S3 = "campaigns";
const SIMILARITIES_KEY_S3 = "similarities";
const SPARK_SUCCESS_FILE = "_SUCCESS";
const CSV_OPTIONS: Options = {
  separator: ",",
  strict: false,
  mapValues: ({ header, value }) => {
    switch (header) {
      case "app_bundle_id_index":
        return parseInt(value);
      case "similarity":
        return parseFloat(value);
      default:
        return value;
    }
  },
};

class Metrics {
  private static readonly prefix = "audience_hashmap_";
  static readonly ADD_SCORES = Metrics.prefix + "add_scores";
  static readonly GET_SCORE = Metrics.prefix + "get_score";
  static readonly REFRESH = Metrics.prefix + "refresh";
  static readonly SIMILARITIES_SIZE = Metrics.prefix + "similarities_size";
  static readonly USAGE = Metrics.prefix + "usage";

  static initMetrics(metricService: MetricService): void {
    metricService.addHistogram(
      Metrics.ADD_SCORES,
      "Duration of getting all scores for a request",
      undefined,
      BucketSet.zeroTo1ms
    );
    metricService.addHistogram(
      Metrics.GET_SCORE,
      "Duration of getting a score for a campaign",
      undefined,
      BucketSet.zeroTo100us
    );
    metricService.addHistogram(
      Metrics.REFRESH,
      "Duration of refreshing hashmap",
      undefined,
      BucketSet.zeroTo60s
    );
    metricService.addGauge(
      Metrics.SIMILARITIES_SIZE,
      "Number of similarities stored in the hashmap"
    );
    metricService.addCounter(Metrics.USAGE, "Number of hashmap usages for each usage type", [
      "app_type",
      "score_type",
    ]);
  }
}

export interface AudienceScoreInfo {
  score: number; // actual score value
  bucketsMap: BucketMapping[]; // buckets mapping for score uniformization
}

export interface AudienceScores {
  campaigns: Map<string, string>; // campaign_id -> brief_id
  apps: Map<string, number>; // app_bundle_id -> app_bundle_id_index
  similarities: Map<string, Map<number, AudienceScoreInfo>>; // brief_id -> { app_bundle_id_index -> score_info }
}

// Audience metadata for a given campaign
export interface AudienceMetadata {
  brief: string; // brief corresponding to the campaign
  score?: number; // audience score returned by the hashmap
  prior?: number; // average audience score of this campaign
  bucketsMap?: BucketMapping[]; // buckets mapping for score uniformization
  fallback: boolean; // whether the fallback mechanism has been used
}

export default class AudienceHashMap {
  private static instance: AudienceHashMap;
  protected audienceScores: AudienceScores;
  private readonly metricService: MetricService;
  private readonly s3Service: S3Service;
  protected s3Bucket: string;
  protected s3Prefix: string;
  protected maxRetries: number;
  private readonly retryDelay: number;
  private readonly refreshInterval: number;
  private lastRefreshTime: Date;

  constructor() {
    this.audienceScores = {
      campaigns: new Map(),
      apps: new Map(),
      similarities: new Map(),
    };
    this.s3Service = new S3Service();
    this.s3Bucket = config.AUDIENCE_S3_BUCKET;
    this.s3Prefix = config.AUDIENCE_S3_PREFIX;
    this.maxRetries = parseInt(config.AUDIENCE_MAX_RETRIES);
    this.retryDelay = parseInt(config.AUDIENCE_RETRY_DELAY);
    this.refreshInterval = parseInt(config.AUDIENCE_REFRESH_INTERVAL);
    this.metricService = MetricService.getInstance();
    Metrics.initMetrics(this.metricService);
    this.lastRefreshTime = new Date(0);

    if (!isTest()) {
      logger.info(`${LOG_ID} starting service`);
      this.refresh();
      setInterval(() => this.refresh(), this.refreshInterval);
    }
  }

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

  /**
   * For each campaign, get the corresponding score and add it to the campaign's metadata.
   * @param request Choice request.
   * @param campaignCreatives
   * @param appBundleId Id of the application.
   * @param webPage Url of the page for web-mobile request.
   */
  public addScores(
    request: ChoiceRequest,
    campaignCreatives: CampaignCreative[],
    appBundleId: string | null | undefined,
    webPage?: string
  ): void {
    const endAddScoresTimer = this.metricService.startTimer(Metrics.ADD_SCORES);
    const { appIndex, appType } = this.getAppIndex(appBundleId, webPage);
    this.metricService.incCounterWithReq(request, Metrics.USAGE, 1, {
      app_type: appType,
    });
    for (const campaignCreative of campaignCreatives) {
      const endGetScoreTimer = this.metricService.startTimer(Metrics.GET_SCORE);
      const audienceMetadata = this.getScore(campaignCreative.campaign.id, appIndex);

      let scoreType = "NO_BRIEF";
      if (audienceMetadata.brief !== constants.FALLBACK_STR) {
        scoreType = audienceMetadata.fallback ? "FALLBACK" : "SCORE";
      }
      this.metricService.incCounterWithReq(request, Metrics.USAGE, 1, { score_type: scoreType });

      campaignCreative.metadataTmp.audienceMetadata = audienceMetadata;
      campaignCreative.metadata.audience_score = audienceMetadata.score;
      endGetScoreTimer();
    }

    endAddScoresTimer();
  }

  private getScore(campaignId: string, appIndex: number): AudienceMetadata {
    let isFallback = appIndex === constants.FALLBACK_NUM;

    // Get brief corresponding to campaign
    let brief = this.audienceScores.campaigns.get(campaignId);
    if (brief === undefined) {
      brief = constants.FALLBACK_STR;
      isFallback = true;
    }
    // Get score-per-app hashmap corresponding to brief
    let appScores = this.audienceScores.similarities.get(brief);
    if (appScores === undefined) {
      appScores = this.audienceScores.similarities.get(constants.FALLBACK_STR);
      isFallback = true;
    }
    // Get score infos corresponding to app
    let scoreInfos = appScores?.get(appIndex);
    if (scoreInfos === undefined) {
      scoreInfos = appScores?.get(constants.FALLBACK_NUM);
      isFallback = true;
    }
    // Get prior score-per-app + bucketsMap corresponding to brief
    const priorScoreInfos = appScores?.get(constants.FALLBACK_NUM);

    return {
      brief: brief,
      score: scoreInfos?.score,
      prior: priorScoreInfos?.score,
      bucketsMap: priorScoreInfos?.bucketsMap,
      fallback: isFallback,
    };
  }

  private getAppIndex(
    appBundleId: string | null | undefined,
    webPage?: string
  ): { appIndex: number; appType: "PAGE_URL" | "BUNDLE_ID" | "NO_APP" } {
    let appIndex = webPage ? this.audienceScores.apps.get(webPage) : null;
    if (appIndex != null) {
      return { appIndex: appIndex, appType: "PAGE_URL" };
    }
    appIndex = this.audienceScores.apps.get(appBundleId ?? "");
    if (appIndex != null) {
      return { appIndex: appIndex, appType: "BUNDLE_ID" };
    }
    return { appIndex: constants.FALLBACK_NUM, appType: "NO_APP" };
  }

  public async refresh(retryCount = 0): Promise<void> {
    function filterKeysBySubPrefix(s3Keys: string[], subPrefix: string) {
      return s3Keys.filter((k) => k.split("/")[1] === subPrefix);
    }
    logger.info(`${LOG_ID} beginning hashmap refresh (attempt ${retryCount})`);
    try {
      const endRefreshTimer = this.metricService.startTimer(Metrics.REFRESH);
      const keys = await this.s3Service.listKeys(this.s3Bucket, this.s3Prefix, [
        CAMPAIGNS_KEY_S3 + "/" + SPARK_SUCCESS_FILE,
        APPLICATIONS_KEY_S3 + "/" + SPARK_SUCCESS_FILE,
        SIMILARITIES_KEY_S3 + "/" + SPARK_SUCCESS_FILE,
        // Here we assert that for each folder "SPARK_SUCCESS_FILE" exists, because if it's not
        // the case, it means that the files are currently being modified by the DAG which generates
        // them periodically. In this case, we want the refresh to fail so that we can retry later.
      ]);
      const [campaigns, apps, similarities] = await Promise.all([
        this.loadCampaigns(filterKeysBySubPrefix(keys, CAMPAIGNS_KEY_S3)),
        this.loadApps(filterKeysBySubPrefix(keys, APPLICATIONS_KEY_S3)),
        this.loadSimilarities(filterKeysBySubPrefix(keys, SIMILARITIES_KEY_S3)),
      ]);
      this.onRefreshSuccess(campaigns, apps, similarities);
      endRefreshTimer();
    } catch (err: any) {
      await this.onRefreshError(err, retryCount);
    }
  }

  private onRefreshSuccess(
    campaigns: Map<string, string>,
    apps: Map<string, number>,
    similarities: Map<string, Map<number, AudienceScoreInfo>>
  ): void {
    this.audienceScores = {
      campaigns: campaigns,
      apps: apps,
      similarities: similarities,
    };
    this.metricService.setGauge(
      Metrics.SIMILARITIES_SIZE,
      Array.from(similarities.values())
        .map((m) => m.size)
        .reduce((s, n) => s + n, 0)
    );
    this.lastRefreshTime = new Date();
    logger.info(`${LOG_ID} hashmap refresh succeeded`);
  }

  private async onRefreshError(err: any, retryCount: number): Promise<void> {
    logger.error(err, `${LOG_ID} hashmap refresh failed`);
    if (retryCount >= this.maxRetries) {
      logger.error(err, `${LOG_ID} ${this.maxRetries} maximum retries reached`);
    } else {
      logger.info(`${LOG_ID} retrying refresh in ${this.retryDelay} ms`);
      await sleep(this.retryDelay);
      await this.refresh(retryCount + 1);
    }
  }

  private async loadCampaigns(s3Keys: string[]): Promise<Map<string, string>> {
    return this.loadHashMap<string, string>(s3Keys, CSV_OPTIONS, (hashmap, line) => {
      hashmap.set(line.campaign_id, line.brief_id);
    });
  }

  private async loadApps(s3Keys: string[]): Promise<Map<string, number>> {
    return this.loadHashMap<string, number>(s3Keys, CSV_OPTIONS, (hashmap, line) => {
      hashmap.set(line.app_bundle_id, line.app_bundle_id_index);
    });
  }

  private async loadSimilarities(
    s3Keys: string[]
  ): Promise<Map<string, Map<number, AudienceScoreInfo>>> {
    return this.loadHashMap<string, Map<number, AudienceScoreInfo>>(
      s3Keys,
      CSV_OPTIONS,
      (hashmap, line) => {
        let briefMap = hashmap.get(line.brief_id);
        if (briefMap === undefined) {
          briefMap = new Map();
          hashmap.set(line.brief_id, briefMap);
        }
        briefMap.set(line.app_bundle_id_index, {
          score: line.similarity,
          bucketsMap: parseBucketMapping(line.cdf, ";"),
        });
      }
    );
  }

  private async loadHashMap<K, V>(
    s3Keys: string[],
    options: Options,
    processLineFn: (hashmap: Map<K, V>, line: any) => void
  ): Promise<Map<K, V>> {
    const hashmap = new Map<K, V>();
    await Promise.all(
      s3Keys.map(async (k) => {
        const reader = this.s3Service.readLinesCsv(this.s3Bucket, k, options);
        for await (const line of reader) processLineFn(hashmap, line);
      })
    );
    return hashmap;
  }

  /**
   * return hashmap size and last update epoch
   */
  getStatus() {
    return {
      audienceScoresSizes: {
        campaigns: this.audienceScores.campaigns.size,
        apps: this.audienceScores.apps.size,
        similarities: this.audienceScores.similarities.size,
      },
      lastRefreshTimestamp: this.lastRefreshTime.getTime(),
    };
  }
}