/**
 * This file is part of the NocoBase (R) project.
 * Copyright (c) 2020-2024 NocoBase Co., Ltd.
 * Authors: NocoBase Team.
 *
 * This project is dual-licensed under AGPL-3.0 and NocoBase Commercial License.
 * For more information, please refer to: https://www.nocobase.com/agreement.
 */

import { fn, literal, Model, Op, Transactionable, where } from '@nocobase/database';
import parser from 'cron-parser';
import type Plugin from '../../Plugin';
import type { WorkflowModel } from '../../types';
import { parseDateWithoutMs, SCHEDULE_MODE } from './utils';
import { parseCollectionName, SequelizeCollectionManager, SequelizeDataSource } from '@nocobase/data-source-manager';
import { pick } from 'lodash';
import { toJSON } from '../../utils';
import { EventOptions } from '../../Dispatcher';

export type ScheduleOnField = {
  field: string;
  // in seconds
  offset?: number;
  unit?: 1000 | 60000 | 3600000 | 86400000;
};

export interface ScheduleTriggerConfig {
  // trigger mode
  mode: number;
  // how to repeat
  repeat?: string | number | null;
  // limit of repeat times
  limit?: number;

  startsOn?: ScheduleOnField;
  endsOn?: string | ScheduleOnField;
}

function getOnTimestampWithOffset({ field, offset = 0, unit = 86400000 }: ScheduleOnField, now: Date) {
  if (!field) {
    return null;
  }
  const timestamp = now.getTime();
  // onDate + offset > now
  // onDate > now - offset
  return timestamp - offset * unit;
}

function getDataOptionTime(record, on, dir = 1) {
  if (!on) {
    return null;
  }
  switch (typeof on) {
    case 'string': {
      const time = parseDateWithoutMs(on);
      return time ? time : null;
    }
    case 'object': {
      const { field, offset = 0, unit = 86400000 } = on;
      if (!field || !record.get(field)) {
        return null;
      }
      const second = new Date(record.get(field));
      second.setMilliseconds(0);
      return second.getTime() + offset * unit * dir;
    }
    default:
      return null;
  }
}

const DialectTimestampFnMap: { [key: string]: (col: string) => string } = {
  postgres(col) {
    return `CAST(FLOOR(extract(epoch from ${col})) AS INTEGER)`;
  },
  mysql(col) {
    return `CAST(FLOOR(UNIX_TIMESTAMP(${col})) AS SIGNED INTEGER)`;
  },
  sqlite(col) {
    return `CAST(FLOOR(unixepoch(${col})) AS INTEGER)`;
  },
};
DialectTimestampFnMap.mariadb = DialectTimestampFnMap.mysql;

function getCronNextTime(cron, currentDate: Date): number {
  const interval = parser.parseExpression(cron, { currentDate });
  const next = interval.next();
  return next.getTime();
}

function matchCronNextTime(cron, currentDate: Date, range: number): boolean {
  return getCronNextTime(cron, currentDate) - currentDate.getTime() <= range;
}

function getHookId(workflow, type: string) {
  return `${type}#${workflow.id}`;
}

export default class DateFieldScheduleTrigger {
  events = new Map();

  private timer: NodeJS.Timeout | null = null;

  private cache: Map<string, any> = new Map();

  // caching workflows in range, default to 5min
  cacheCycle = 300_000;

  onAfterStart = () => {
    if (this.timer) {
      return;
    }

    this.timer = setInterval(() => this.reload(), this.cacheCycle);

    this.reload();
  };

  onBeforeStop = () => {
    if (this.timer) {
      clearInterval(this.timer);
    }

    for (const [key, timer] of this.cache.entries()) {
      clearTimeout(timer);
      this.cache.delete(key);
    }
  };

  constructor(public workflow: Plugin) {
    workflow.app.on('afterStart', this.onAfterStart);
    workflow.app.on('beforeStop', this.onBeforeStop);
  }

  reload() {
    for (const [key, timer] of this.cache.entries()) {
      clearTimeout(timer);
      this.cache.delete(key);
    }

    const workflows = Array.from(this.workflow.enabledCache.values()).filter(
      (item) => item.type === 'schedule' && item.config.mode === SCHEDULE_MODE.DATE_FIELD,
    );

    workflows.forEach((workflow) => {
      this.inspect(workflow);
    });
  }

  async inspect(workflow: WorkflowModel) {
    const now = new Date();
    const records = await this.loadRecordsToSchedule(workflow, now);
    this.workflow.getLogger(workflow.id).info(`[Schedule on date field] ${records.length} records to schedule`);
    records.forEach((record) => {
      const nextTime = this.getRecordNextTime(workflow, record);
      this.schedule(workflow, record, nextTime, Boolean(nextTime));
    });
  }

  // 1. startsOn in range -> yes
  // 2. startsOn before now, has no repeat -> no
  // 3. startsOn before now, and has repeat:
  //   a. repeat out of range -> no
  //   b. repeat in range (number or cron):
  //     i. endsOn after now -> yes
  //     ii. endsOn before now -> no
  async loadRecordsToSchedule(
    { id, config: { collection, limit, startsOn, repeat, endsOn }, stats }: WorkflowModel,
    currentDate: Date,
  ) {
    const { dataSourceManager } = this.workflow.app;
    if (limit && stats.executed >= limit) {
      this.workflow.getLogger(id).warn(`[Schedule on date field] limit reached (all executed ${stats.executed})`);
      return [];
    }
    if (!startsOn) {
      this.workflow.getLogger(id).warn(`[Schedule on date field] "startsOn" is not configured`);
      return [];
    }
    const timestamp = currentDate.getTime();

    const startTimestamp = getOnTimestampWithOffset(startsOn, currentDate);
    if (!startTimestamp) {
      this.workflow.getLogger(id).warn(`[Schedule on date field] "startsOn.field" is not configured`);
      return [];
    }

    const [dataSourceName, collectionName] = parseCollectionName(collection);
    const { collectionManager } = dataSourceManager.get(dataSourceName);
    if (!(collectionManager instanceof SequelizeCollectionManager)) {
      return [];
    }
    const { db } = collectionManager;
    const { model } = collectionManager.getCollection(collectionName);

    const range = this.cacheCycle * 2;

    const conditions: any[] = [
      {
        [startsOn.field]: {
          // cache next 2 cycles
          [Op.lt]: new Date(startTimestamp + range),
        },
      },
    ];

    if (repeat) {
      // when repeat is number, means repeat after startsOn
      if (typeof repeat === 'number') {
        const tsFn = DialectTimestampFnMap[db.options.dialect];
        if (repeat > range && tsFn) {
          const offsetSeconds = Math.round(((startsOn.offset || 0) * (startsOn.unit || 1000)) / 1000);
          const repeatSeconds = Math.round(repeat / 1000);
          const nowSeconds = Math.round(timestamp / 1000);
          const { field } = model.getAttributes()[startsOn.field];
          const modExp = literal(
            `MOD(MOD(${tsFn(
              db.sequelize.getQueryInterface().quoteIdentifiers(field),
            )} + ${offsetSeconds} - ${nowSeconds}, ${repeatSeconds}) + ${repeatSeconds}, ${repeatSeconds})`,
          );
          conditions.push(where(modExp, { [Op.lt]: Math.round(range / 1000) }));
        }
      } else if (typeof repeat === 'string') {
        if (!matchCronNextTime(repeat, currentDate, range)) {
          return [];
        }
      }

      if (endsOn) {
        if (typeof endsOn === 'string') {
          if (parseDateWithoutMs(endsOn) <= timestamp) {
            return [];
          }
        } else {
          const endTimestamp = getOnTimestampWithOffset(endsOn, currentDate);
          if (endTimestamp) {
            conditions.push({
              [endsOn.field]: {
                [Op.gte]: new Date(endTimestamp),
              },
            });
          }
        }
      }
    } else {
      conditions.push({
        [startsOn.field]: {
          [Op.gte]: new Date(startTimestamp),
        },
      });
    }
    this.workflow.getLogger(id).debug(`[Schedule on date field] conditions: `, { conditions });

    return model.findAll({
      where: {
        [Op.and]: conditions,
      },
    });
  }

  getRecordNextTime(workflow: WorkflowModel, record: Model, nextSecond = false) {
    const {
      config: { startsOn, endsOn, repeat, limit },
      stats,
    } = workflow;
    if (limit && stats.executed >= limit) {
      return null;
    }
    const logger = this.workflow.getLogger(workflow.id);
    const range = this.cacheCycle;
    const now = new Date();
    now.setMilliseconds(nextSecond ? 1000 : 0);
    const timestamp = now.getTime();
    const startTime = getDataOptionTime(record, startsOn);
    const endTime = getDataOptionTime(record, endsOn);
    let nextTime = null;
    if (!startTime) {
      logger.debug(`[Schedule on date field] getNextTime: startsOn not configured`);
      return null;
    }
    if (startTime > timestamp + range) {
      logger.debug(`[Schedule on date field] getNextTime: startsOn is out of caching window`);
      return null;
    }
    if (startTime >= timestamp) {
      if (!endTime || startTime <= endTime) {
        return startTime;
      }
      logger.debug(`[Schedule on date field] getNextTime: endsOn is before startsOn or out of caching window`);
      return null;
    } else {
      if (!repeat) {
        logger.debug(
          `[Schedule on date field] getNextTime: startsOn is before current time and repeat is not configured`,
        );
        return null;
      }
    }
    if (typeof repeat === 'number') {
      nextTime = timestamp + repeat - ((timestamp - startTime) % repeat);
      if (nextTime - timestamp > range) {
        logger.debug(`[Schedule on date field] getNextTime: nextTime (${nextTime}) is out of caching window`);
        return null;
      }
      if (endTime && endTime < nextTime) {
        logger.debug(`[Schedule on date field] getNextTime: nextTime is after endsOn`);
        return null;
      }
    } else if (typeof repeat === 'string') {
      nextTime = getCronNextTime(repeat, now);
      if (nextTime - timestamp > range) {
        logger.debug(`[Schedule on date field] getNextTime: nextTime (${nextTime}) is out of caching window`);
        return null;
      }
      if (endTime && endTime < nextTime) {
        logger.debug(`[Schedule on date field] getNextTime: nextTime is after endsOn`);
        return null;
      }
    }
    if (endTime && endTime <= timestamp) {
      logger.debug(`[Schedule on date field] getNextTime: nextTime is after endsOn`);
      return null;
    }
    return nextTime;
  }

  schedule(workflow: WorkflowModel, record: Model, nextTime: number, toggle = true, options = {}) {
    const [dataSourceName, collectionName] = parseCollectionName(workflow.config.collection);
    const { filterTargetKey } = this.workflow.app.dataSourceManager.dataSources
      .get(dataSourceName)
      .collectionManager.getCollection(collectionName);
    const recordPk = record.get(filterTargetKey);
    if (toggle) {
      const nextInterval = Math.max(0, nextTime - Date.now());
      const key = `${workflow.id}:${recordPk}@${nextTime}`;

      if (!this.cache.has(key)) {
        if (nextInterval) {
          this.cache.set(key, setTimeout(this.trigger.bind(this, workflow, record, nextTime), nextInterval));
        } else {
          return this.trigger(workflow, record, nextTime, options);
        }
      }
    } else {
      for (const [key, timer] of this.cache.entries()) {
        if (key.startsWith(`${workflow.id}:${recordPk}@`)) {
          clearTimeout(timer);
          this.cache.delete(key);
        }
      }
    }
  }

  async trigger(workflow: WorkflowModel, record: Model, nextTime: number, { transaction }: Transactionable = {}) {
    const [dataSourceName, collectionName] = parseCollectionName(workflow.config.collection);
    const { repository, filterTargetKey } = this.workflow.app.dataSourceManager.dataSources
      .get(dataSourceName)
      .collectionManager.getCollection(collectionName);
    const recordPk = record.get(filterTargetKey);
    const data = (await repository.findOne({
      filterByTk: recordPk,
      appends: workflow.config.appends,
      transaction,
    })) as Model;
    if (!data) {
      this.workflow
        .getLogger(workflow.id)
        .warn(`[Schedule on date field] record (${recordPk}) not exists, will not trigger`);
      return;
    }
    const eventKey = `${workflow.id}:${recordPk}@${nextTime}`;
    this.cache.delete(eventKey);
    // NOTE: data.toJSON() will cause erorr

    const json = toJSON(data);
    const args: [WorkflowModel, object, EventOptions] = [
      workflow,
      {
        data: json,
        date: new Date(nextTime),
      },
      {
        eventKey,
      },
    ];
    if (transaction) {
      transaction.afterCommit(() => {
        this.workflow.trigger(...args);
      });
    } else {
      this.workflow.trigger(...args);
    }

    if (!workflow.config.repeat || (workflow.config.limit && workflow.stats.executed >= workflow.config.limit - 1)) {
      return;
    }

    const n = this.getRecordNextTime(workflow, data, true);
    if (n) {
      this.schedule(workflow, data, n, true);
    }
  }

  on(workflow: WorkflowModel) {
    this.inspect(workflow);

    const { collection } = workflow.config;
    const [dataSourceName, collectionName] = parseCollectionName(collection);
    const event = `${collectionName}.afterSaveWithAssociations`;
    const name = getHookId(workflow, event);
    if (this.events.has(name)) {
      return;
    }

    const listener = async (data: Model, { transaction }) => {
      const nextTime = this.getRecordNextTime(workflow, data);
      this.workflow.getLogger().debug(`[Schedule on date field] record saved, nextTime: ${nextTime}`);
      return this.schedule(workflow, data, nextTime, Boolean(nextTime), { transaction });
    };

    this.events.set(name, listener);
    const dataSource = this.workflow.app.dataSourceManager.dataSources.get(dataSourceName) as SequelizeDataSource;
    if (!dataSource) {
      this.workflow.getLogger().warn(`[Schedule on date field] data source not exists: ${dataSourceName}`);
      return;
    }
    const { db } = dataSource.collectionManager as SequelizeCollectionManager;
    db.on(event, listener);
  }

  off(workflow: WorkflowModel) {
    for (const [key, timer] of this.cache.entries()) {
      if (key.startsWith(`${workflow.id}:`)) {
        clearTimeout(timer);
        this.cache.delete(key);
      }
    }

    const { collection } = workflow.config;
    const [dataSourceName, collectionName] = parseCollectionName(collection);
    const event = `${collectionName}.afterSaveWithAssociations`;
    const name = getHookId(workflow, event);
    const listener = this.events.get(name);
    if (listener) {
      const dataSource = this.workflow.app.dataSourceManager.dataSources.get(dataSourceName) as SequelizeDataSource;
      if (!dataSource) {
        this.workflow.getLogger().warn(`[Schedule on date field] data source not exists: ${dataSourceName}`);
        return;
      }
      const { db } = dataSource.collectionManager as SequelizeCollectionManager;
      db.off(event, listener);
      this.events.delete(name);
    }
  }

  async execute(workflow, values, options) {
    const [dataSourceName, collectionName] = parseCollectionName(workflow.config.collection);
    const { collectionManager } = this.workflow.app.dataSourceManager.dataSources.get(dataSourceName);
    const { filterTargetKey, repository } = collectionManager.getCollection(collectionName);

    let { data } = values;
    let filterByTk;
    let loadNeeded = false;
    if (data && typeof data === 'object') {
      filterByTk = Array.isArray(filterTargetKey)
        ? pick(
            data,
            filterTargetKey.sort((a, b) => a.localeCompare(b)),
          )
        : data[filterTargetKey];
    } else {
      filterByTk = data;
      loadNeeded = true;
    }
    if (loadNeeded || workflow.config.appends?.length) {
      data = await repository.findOne({
        filterByTk,
        appends: workflow.config.appends,
      });
    }

    return this.workflow.trigger(workflow, { ...values, data, date: values?.date ?? new Date() }, options);
  }

  validateContext(values) {
    if (!values?.data) {
      return {
        data: 'Data is required',
      };
    }
    return null;
  }
}
