import RedisClient, { RedisClientOptions } from ".";
import RedisCommandsQueue from "./commands-queue";
import { isIP } from "net";
import { lookup } from "dns/promises";
import assert from "node:assert";
import { setTimeout } from "node:timers/promises";
import { RedisTcpSocketOptions } from "./socket";
import diagnostics_channel from "node:diagnostics_channel";
import { RedisArgument } from "../RESP/types";
import { publish, CHANNELS } from "./tracing";

type RedisType = RedisClient<any, any, any, any, any>;

export const SMIGRATED_EVENT = "__SMIGRATED";

interface Address {
  host: string;
  port: number;
}

interface Destination {
  addr: Address;
  slots: (number | [number, number])[];
}

interface SMigratedEntry {
    source: Address;
    destinations: Destination[]
}

export interface SMigratedEvent {
  seqId: number,
  entries: SMigratedEntry[]
}

export const MAINTENANCE_EVENTS = {
  PAUSE_WRITING: "pause-writing",
  RESUME_WRITING: "resume-writing",
  TIMEOUTS_UPDATE: "timeouts-update",
} as const;

const PN = {
  MOVING: "MOVING",
  MIGRATING: "MIGRATING",
  MIGRATED: "MIGRATED",
  FAILING_OVER: "FAILING_OVER",
  FAILED_OVER: "FAILED_OVER",
  SMIGRATING: "SMIGRATING",
  SMIGRATED: "SMIGRATED",
};

export type DiagnosticsEvent = {
  type: string;
  timestamp: number;
  data?: Object;
};

export const dbgMaintenance = (...args: any[]) => {
  if (!process.env.REDIS_DEBUG_MAINTENANCE) return;
  return console.log(new Date().toISOString().slice(11, 23), "[MNT]", ...args);
};

export const emitDiagnostics = (event: DiagnosticsEvent) => {
  if (!process.env.REDIS_EMIT_DIAGNOSTICS) return;

  const channel = diagnostics_channel.channel("redis.maintenance");
  channel.publish(event);
};

export interface MaintenanceUpdate {
  relaxedCommandTimeout?: number;
  relaxedSocketTimeout?: number;
}

export default class EnterpriseMaintenanceManager {
  #commandsQueue: RedisCommandsQueue;
  #options: RedisClientOptions;
  #isMaintenance = 0;
  #client: RedisType;

  static setupDefaultMaintOptions(options: RedisClientOptions) {
    if (options.maintNotifications === undefined) {
      options.maintNotifications =
        options?.RESP === 3 ? "auto" : "disabled";
    }
    if (options.maintEndpointType === undefined) {
      options.maintEndpointType = "auto";
    }
    if (options.maintRelaxedSocketTimeout === undefined) {
      options.maintRelaxedSocketTimeout = 10000;
    }
    if (options.maintRelaxedCommandTimeout === undefined) {
      options.maintRelaxedCommandTimeout = 10000;
    }
  }

  static async getHandshakeCommand(
    options: RedisClientOptions,
    clientId: string,
  ): Promise<
    | { cmd: Array<RedisArgument>; errorHandler: (error: Error) => void }
    | undefined
  > {
    if (options.maintNotifications === "disabled") return;

    const host = options.url
      ? new URL(options.url).hostname
      : (options.socket as RedisTcpSocketOptions | undefined)?.host;

    if (!host) return;

    const tls = options.socket?.tls ?? false;

    const movingEndpointType = await determineEndpoint(tls, host, options);
    return {
      cmd: [
        "CLIENT",
        "MAINT_NOTIFICATIONS",
        "ON",
        "moving-endpoint-type",
        movingEndpointType,
      ],
      errorHandler: (error: Error) => {
        dbgMaintenance("handshake failed:", error);

        publish(CHANNELS.ERROR, () => ({
          error,
          origin: 'client',
          internal: true,
          clientId,
        }));

        if (options.maintNotifications === "enabled") {
          throw error;
        }

      },
    };
  }

  constructor(
    commandsQueue: RedisCommandsQueue,
    client: RedisType,
    options: RedisClientOptions,
  ) {
    this.#commandsQueue = commandsQueue;
    this.#options = options;
    this.#client = client;

    this.#commandsQueue.addPushHandler(this.#onPush);
  }

  #onPush = (push: Array<any>): boolean => {
    dbgMaintenance("ONPUSH:", push.map(String));

    if (!Array.isArray(push) || !Object.values(PN).includes(String(push[0]))) {
      return false;
    }

    const type = String(push[0]);

    publish(CHANNELS.MAINTENANCE, () => ({
      notification: type,
      clientId: this.#client._clientId,
    }));

    emitDiagnostics({
      type,
      timestamp: Date.now(),
      data: {
        push: push.map(String),
      },
    });
    switch (type) {
      case PN.MOVING: {
        // [ 'MOVING', '17', '15', '54.78.247.156:12075' ]
        //             ^seq   ^after    ^new ip
        const afterSeconds = push[2];
        const url: string | null = push[3] ? String(push[3]) : null;
        dbgMaintenance("Received MOVING:", afterSeconds, url);
        this.#onMoving(afterSeconds, url);
        return true;
      }
      case PN.MIGRATING:
      case PN.SMIGRATING:
      case PN.FAILING_OVER: {
        dbgMaintenance("Received MIGRATING|SMIGRATING|FAILING_OVER");
        this.#onMigrating();
        return true;
      }
      case PN.MIGRATED:
      case PN.FAILED_OVER: {
        dbgMaintenance("Received MIGRATED|FAILED_OVER");
        this.#onMigrated();
        return true;
      }
      case PN.SMIGRATED: {
        dbgMaintenance("Received SMIGRATED");
        this.#onSMigrated(push);
        this.#onMigrated();  // Un-relax timeouts after slot migration completes
        return true;
      }
    }
    return false;
  };

  //  Queue:
  //     toWrite [ C D E ]
  //     waitingForReply [ A B ]   - aka In-flight commands
  //
  //  time: ---1-2---3-4-5-6---------------------------
  //
  //  1. [EVENT] MOVING PN received
  //  2. [ACTION] Pause writing ( we need to wait for new socket to connect and for all in-flight commands to complete )
  //  3. [EVENT] New socket connected
  //  4. [EVENT] In-flight commands completed
  //  5. [ACTION] Destroy old socket
  //  6. [ACTION] Resume writing -> we are going to write to the new socket from now on
  #onMoving = async (
    afterSeconds: number,
    url: string | null,
  ): Promise<void> => {
    // 1 [EVENT] MOVING PN received
    this.#onMigrating();

    let host: string;
    let port: number;

    // The special value `none` indicates that the `MOVING` message doesn’t need
    // to contain an endpoint. Instead it contains the value `null` then. In
    // such a corner case, the client is expected to schedule a graceful
    // reconnect to its currently configured endpoint after half of the grace
    // period that was communicated by the server is over.
    if (url === null) {
      assert(this.#options.maintEndpointType === "none");
      const { host: h, port: p } = this.#getAddress()
      host = h;
      port = p;
      const waitTime = (afterSeconds * 1000) / 2;
      dbgMaintenance(`Wait for ${waitTime}ms`);
      await setTimeout(waitTime);
    } else {
      const split = url.split(":");
      host = split[0];
      port = Number(split[1]);
    }

    // 2 [ACTION] Pause writing
    dbgMaintenance("Pausing writing of new commands to old socket");
    this.#client._pause();

    dbgMaintenance("Creating new tmp client");
    let start = performance.now();

    // If the URL is provided, it takes precedense
    // the options object could just be mutated
    if (this.#options.url) {
      const u = new URL(this.#options.url);
      u.hostname = host;
      u.port = String(port);
      this.#options.url = u.toString();
    } else {
      this.#options.socket = {
        ...this.#options.socket,
        host,
        port,
      };
    }
    const tmpClient = this.#client.duplicate();
    tmpClient.on("error", (error: unknown) => {
      //We dont know how to handle tmp client errors
      dbgMaintenance(`[ERR]`, error);
    });
    dbgMaintenance(
      `Tmp client created in ${(performance.now() - start).toFixed(2)}ms`,
    );
    dbgMaintenance(
      `Set timeout for tmp client to ${this.#options.maintRelaxedSocketTimeout}`,
    );
    tmpClient._maintenanceUpdate({
      relaxedCommandTimeout: this.#options.maintRelaxedCommandTimeout,
      relaxedSocketTimeout: this.#options.maintRelaxedSocketTimeout,
    });
    dbgMaintenance(`Connecting tmp client: ${host}:${port}`);
    start = performance.now();
    await tmpClient.connect();
    dbgMaintenance(
      `Connected to tmp client in ${(performance.now() - start).toFixed(2)}ms`,
    );
    // 3 [EVENT] New socket connected

    dbgMaintenance(`Wait for all in-flight commands to complete`);
    await this.#commandsQueue.waitForInflightCommandsToComplete();
    dbgMaintenance(`In-flight commands completed`);
    // 4 [EVENT] In-flight commands completed

    dbgMaintenance("Swap client sockets...");
    const oldSocket = this.#client._ejectSocket();
    const newSocket = tmpClient._ejectSocket();
    this.#client._insertSocket(newSocket);
    tmpClient._insertSocket(oldSocket);
    tmpClient.destroy();
    dbgMaintenance("Swap client sockets done.");
    // 5 + 6
    dbgMaintenance("Resume writing");
    this.#client._unpause();
    this.#onMigrated();
    publish(CHANNELS.CONNECTION_HANDOFF, () => ({ clientId: this.#client._clientId }));
  };

  #onMigrating = () => {
    this.#isMaintenance++;
    if (this.#isMaintenance > 1) {
      dbgMaintenance(`Timeout relaxation already done`);
      return;
    }

    const update: MaintenanceUpdate = {
      relaxedCommandTimeout: this.#options.maintRelaxedCommandTimeout,
      relaxedSocketTimeout: this.#options.maintRelaxedSocketTimeout,
    };

    this.#client._maintenanceUpdate(update);
  };

  #onMigrated = () => {
    //ensure that #isMaintenance doesnt go under 0
    this.#isMaintenance = Math.max(this.#isMaintenance - 1, 0);
    if (this.#isMaintenance > 0) {
      dbgMaintenance(`Not ready to unrelax timeouts yet`);
      return;
    }

    const update: MaintenanceUpdate = {
      relaxedCommandTimeout: undefined,
      relaxedSocketTimeout: undefined,
    };

    this.#client._maintenanceUpdate(update);
  };

  #onSMigrated = (push: any[]) => {
    const smigratedEvent = EnterpriseMaintenanceManager.parseSMigratedPush(push);
    dbgMaintenance(`emit smigratedEvent`, smigratedEvent);
    this.#client._handleSmigrated(smigratedEvent);
  }

  /**
   * Parses an SMIGRATED push message into a structured SMigratedEvent.
   *
   * SMIGRATED format:
   * - SMIGRATED, "seqid", followed by a list of N triplets:
   *     - source endpoint
   *     - target endpoint
   *     - comma separated list of slot ranges
   *
   * A source and a target endpoint may appear in multiple triplets.
   * There is no optimization of the source, dest, slot-range list in the SMIGRATED message.
   * The client code should read through the entire list of triplets in order to get a full
   * list of moved slots, or full list of sources and targets.
   *
   * Example:
   * [ 'SMIGRATED', 15, [ [ '127.0.0.1:6379', '127.0.0.2:6379', '123,456,789-1000' ], [ '127.0.0.3:6380', '127.0.0.4:6380', '124,457,300-500' ] ] ]
   *                ^seq     ^source1          ^destination1     ^slots                  ^source2          ^destination2    ^slots
   *
   * Result structure guarantees:
   * - Each source address appears in exactly one entry (entries are deduplicated by source)
   * - Within each entry, each destination address appears exactly once (destinations are deduplicated per source)
   * - Each destination contains the complete list of slots that moved from that source to that destination
   * - Note: The same destination address CAN appear under different sources (e.g., node X receives slots from both A and B)
   */
  static parseSMigratedPush(push: any[]): SMigratedEvent {
    const map = new Map<string, Destination[]>();

    for (const [src, destination, slots] of push[2]) {
      const source = String(src);
      const [dHost, dPort] = String(destination).split(':');
      // `slots` could be mix of single slots and ranges, for example: 123,456,789-1000
      const parsedSlots = String(slots).split(',').map((singleOrRange): number | [number, number] => {
        const separatorIndex = singleOrRange.indexOf('-');
        if (separatorIndex === -1) {
          // Its single slot
          return Number(singleOrRange);
        }
        // Its range
        return [Number(singleOrRange.substring(0, separatorIndex)), Number(singleOrRange.substring(separatorIndex + 1))];
      });

      const destinations: Destination[] = map.get(source) ?? [];
      const dest = destinations.find(d => d.addr.host === dHost && d.addr.port === Number(dPort));
      if (dest) {
        // destination already exists, just add the slots
        dest.slots = dest.slots.concat(parsedSlots);
      } else {
        destinations.push({
          addr: {
            host: dHost,
            port: Number(dPort)
          },
          slots: parsedSlots
        });
      }
      map.set(source, destinations);
    }

    const entries: SMigratedEntry[] = [];
    for (const [src, destinations] of map.entries()) {
      const [host, port] = src.split(":");
      entries.push({
        source: {
          host,
          port: Number(port)
        },
        destinations
      });
    }

    return {
      seqId: push[1],
      entries
    };
  }

  #getAddress(): { host: string, port: number } {
    assert(this.#options.socket !== undefined);
    assert("host" in this.#options.socket);
    assert(typeof this.#options.socket.host === "string");
    const host = this.#options.socket.host;
    assert(typeof this.#options.socket.port === "number");
    const port = this.#options.socket.port;
    return { host, port };
  }
}


export type MovingEndpointType =
  | "auto"
  | "internal-ip"
  | "internal-fqdn"
  | "external-ip"
  | "external-fqdn"
  | "none";

function isPrivateIP(ip: string): boolean {
  const version = isIP(ip);
  if (version === 4) {
    const octets = ip.split(".").map(Number);
    return (
      octets[0] === 10 ||
      (octets[0] === 172 && octets[1] >= 16 && octets[1] <= 31) ||
      (octets[0] === 192 && octets[1] === 168)
    );
  }
  if (version === 6) {
    return (
      ip.startsWith("fc") || // Unique local
      ip.startsWith("fd") || // Unique local
      ip === "::1" || // Loopback
      ip.startsWith("fe80") // Link-local unicast
    );
  }
  return false;
}

async function determineEndpoint(
  tlsEnabled: boolean,
  host: string,
  options: RedisClientOptions,
): Promise<MovingEndpointType> {
  assert(options.maintEndpointType !== undefined);
  if (options.maintEndpointType !== "auto") {
    dbgMaintenance(`Determine endpoint type: ${options.maintEndpointType}`);
    return options.maintEndpointType;
  }

  const ip = isIP(host) ? host : (await lookup(host, { family: 0 })).address;

  const isPrivate = isPrivateIP(ip);

  let result: MovingEndpointType;
  if (tlsEnabled) {
    result = isPrivate ? "internal-fqdn" : "external-fqdn";
  } else {
    result = isPrivate ? "internal-ip" : "external-ip";
  }

  dbgMaintenance(`Determine endpoint type: ${result}`);
  return result;
}
