import { randomUUID } from "node:crypto";
import {
  COMPLETE_ACK_TIMEOUT_MS,
  COORDINATION_CHANNELS,
  COORDINATION_PROTOCOL_VERSION,
  DISCOVERY_WINDOW_MS,
  PREPARE_ACK_TIMEOUT_MS,
  freezeEvent,
  parseCompleted,
  parseDiscovered,
  parsePrepared,
  isCompactionOutcome,
  type CompactionOutcome,
  type CoordinationEventBus,
} from "./coordination-protocol.ts";

interface CoordinationTimeouts {
  readonly discoveryWindowMs: number;
  readonly prepareAckTimeoutMs: number;
  readonly completeAckTimeoutMs: number;
}

const DEFAULT_TIMEOUTS: CoordinationTimeouts = Object.freeze({
  discoveryWindowMs: DISCOVERY_WINDOW_MS,
  prepareAckTimeoutMs: PREPARE_ACK_TIMEOUT_MS,
  completeAckTimeoutMs: COMPLETE_ACK_TIMEOUT_MS,
});

export interface CoordinationTransaction {
  readonly requestId: string;
  readonly participantIds: readonly string[];
  complete(outcome: CompactionOutcome, continuationExpected?: boolean): Promise<boolean>;
  /** 仅在 continuation 尚未可能提交，或压缩事务仍未完成时发送幂等 not_started。 */
  cancel(): Promise<void>;
}

export type PrepareResult =
  | { readonly prepared: true; readonly transaction: CoordinationTransaction }
  | { readonly prepared: false };

export class CoordinationClient {
  private readonly events: CoordinationEventBus;
  private readonly timeouts: CoordinationTimeouts;
  private readonly participants = new Map<string, boolean>();
  private readonly invalidParticipantIds = new Set<string>();
  private discoveryPromise: Promise<void> | undefined;
  private discoveryCompleted = false;
  private discoveryRequestId: string | undefined;
  private preparePromise: Promise<PrepareResult> | undefined;
  private readonly pendingCancellations = new Set<() => void>();
  private closed = false;

  constructor(events: CoordinationEventBus, timeouts: CoordinationTimeouts = DEFAULT_TIMEOUTS) {
    this.events = events;
    this.timeouts = timeouts;
  }

  discover(): Promise<void> {
    if (this.closed || this.discoveryCompleted) return Promise.resolve();
    if (this.discoveryPromise) return this.discoveryPromise;

    const requestId = randomUUID();
    this.discoveryRequestId = requestId;
    this.participants.clear();
    this.invalidParticipantIds.clear();
    this.discoveryPromise = new Promise<void>((resolve) => {
      let settled = false;
      let timer: NodeJS.Timeout | undefined;
      const finish = (): void => {
        if (settled) return;
        settled = true;
        if (timer !== undefined) clearTimeout(timer);
        unsubscribe();
        this.pendingCancellations.delete(finish);
        if (this.discoveryRequestId === requestId) this.discoveryRequestId = undefined;
        this.discoveryCompleted = true;
        resolve();
      };
      const unsubscribe = this.events.on(COORDINATION_CHANNELS.discovered, (data) => {
        const response = parseDiscovered(data);
        if (!response || response.requestId !== requestId) return;
        if (this.invalidParticipantIds.has(response.participantId)) return;
        const previous = this.participants.get(response.participantId);
        if (previous !== undefined && previous !== response.requiresBarrier) {
          this.participants.delete(response.participantId);
          this.invalidParticipantIds.add(response.participantId);
          return;
        }
        this.participants.set(response.participantId, response.requiresBarrier);
      });
      timer = setTimeout(finish, this.timeouts.discoveryWindowMs);
      this.pendingCancellations.add(finish);
      try {
        this.events.emit(COORDINATION_CHANNELS.discover, freezeEvent({
          protocolVersion: COORDINATION_PROTOCOL_VERSION,
          requestId,
        }));
      } catch {
        finish();
      }
    }).finally(() => {
      this.discoveryPromise = undefined;
    });
    return this.discoveryPromise;
  }

  prepare(): Promise<PrepareResult> {
    if (this.closed) return Promise.resolve({ prepared: false });
    if (this.preparePromise !== undefined) return this.preparePromise;

    const operation = this.prepareOnce();
    const shared = operation.finally(() => {
      if (this.preparePromise === shared) this.preparePromise = undefined;
    });
    this.preparePromise = shared;
    return shared;
  }

  private async prepareOnce(): Promise<PrepareResult> {
    await this.refreshDiscovery();
    if (this.closed) return { prepared: false };

    const participantIds = [...this.participants]
      .filter(([, requiresBarrier]) => requiresBarrier)
      .map(([participantId]) => participantId)
      .sort();
    const requestId = randomUUID();
    if (participantIds.length === 0) {
      return {
        prepared: true,
        transaction: this.createTransaction(requestId, participantIds),
      };
    }

    const results = await Promise.all(participantIds.map(async (participantId) => {
      const response = await this.waitForPrepared(requestId, participantId);
      return response === true;
    }));
    const preparedParticipantIds = participantIds.filter((_, index) => results[index] === true);
    if (preparedParticipantIds.length === participantIds.length) {
      return {
        prepared: true,
        transaction: this.createTransaction(requestId, participantIds),
      };
    }

    // ACK 可能在对端建立屏障后丢失；向全部目标发送幂等 not_started 才能闭合不确定状态。
    await this.completeParticipants(requestId, participantIds, "not_started");
    return { prepared: false };
  }

  close(): void {
    if (this.closed) return;
    this.closed = true;
    for (const cancel of [...this.pendingCancellations]) cancel();
    this.pendingCancellations.clear();
    this.discoveryRequestId = undefined;
    this.participants.clear();
    this.invalidParticipantIds.clear();
  }

  private async refreshDiscovery(): Promise<void> {
    const pendingDiscovery = this.discoveryPromise;
    if (pendingDiscovery !== undefined) await pendingDiscovery;
    if (this.closed) return;

    this.discoveryCompleted = false;
    await this.discover();
  }

  private createTransaction(requestId: string, participantIds: readonly string[]): CoordinationTransaction {
    let completion: Promise<boolean> | undefined;
    let compensation: Promise<boolean> | undefined;
    let firstDecision: Readonly<{
      outcome: CompactionOutcome;
      continuationExpected: boolean;
    }> | undefined;
    let cancellation: Promise<void> | undefined;
    const compensate = (): Promise<boolean> => {
      compensation ??= this.completeParticipants(requestId, participantIds, "not_started", false);
      return compensation;
    };
    const complete = (
      outcome: CompactionOutcome,
      continuationExpected = false,
    ): Promise<boolean> => {
      if (
        !isCompactionOutcome(outcome)
        || typeof continuationExpected !== "boolean"
        || (continuationExpected && outcome !== "succeeded")
      ) return Promise.resolve(false);
      if (completion !== undefined) return completion;
      firstDecision = Object.freeze({ outcome, continuationExpected });
      completion = this.completeTransaction(
        requestId,
        participantIds,
        firstDecision.outcome,
        firstDecision.continuationExpected,
        compensate,
      );
      return completion;
    };
    return Object.freeze({
      requestId,
      participantIds: Object.freeze([...participantIds]),
      complete,
      cancel: (): Promise<void> => {
        cancellation ??= (async () => {
          if (completion === undefined) {
            await complete("not_started", false);
            return;
          }
          await completion;
          if (firstDecision?.outcome !== "not_started") await compensate();
        })();
        return cancellation;
      },
    });
  }

  private async completeTransaction(
    requestId: string,
    participantIds: readonly string[],
    outcome: CompactionOutcome,
    continuationExpected: boolean,
    compensate: () => Promise<boolean>,
  ): Promise<boolean> {
    const completed = await this.completeParticipants(
      requestId,
      participantIds,
      outcome,
      continuationExpected,
    );
    if (completed || outcome === "not_started") return completed;

    // 任一固定成员未确认时，已成功成员也可能正在等待 continuation；全体补偿才能闭合事务。
    await compensate();
    return false;
  }

  private async waitForPrepared(requestId: string, participantId: string): Promise<boolean | undefined> {
    return await this.waitForAck(
      COORDINATION_CHANNELS.prepared,
      this.timeouts.prepareAckTimeoutMs,
      (data) => {
        const response = parsePrepared(data);
        if (!response || response.requestId !== requestId || response.participantId !== participantId) return undefined;
        return response.prepared;
      },
      () => this.events.emit(COORDINATION_CHANNELS.prepare, freezeEvent({
        protocolVersion: COORDINATION_PROTOCOL_VERSION,
        requestId,
        participantId,
      })),
    );
  }

  private async completeParticipants(
    requestId: string,
    participantIds: readonly string[],
    outcome: CompactionOutcome,
    continuationExpected = false,
  ): Promise<boolean> {
    const effectiveContinuationExpected = outcome === "succeeded" && continuationExpected;
    if (this.closed) {
      if (outcome !== "not_started") return participantIds.length === 0;
      for (const participantId of participantIds) {
        try {
          this.events.emit(COORDINATION_CHANNELS.complete, freezeEvent({
            protocolVersion: COORDINATION_PROTOCOL_VERSION,
            requestId,
            participantId,
            outcome,
            continuationExpected: effectiveContinuationExpected,
          }));
        } catch {
          // close 后没有 ACK 窗口；同步投递失败只能保持未确认。
        }
      }
      return participantIds.length === 0;
    }

    const results = await Promise.all(participantIds.map(async (participantId) => {
      const response = await this.waitForAck(
        COORDINATION_CHANNELS.completed,
        this.timeouts.completeAckTimeoutMs,
        (data) => {
          const completed = parseCompleted(data);
          if (!completed || completed.requestId !== requestId || completed.participantId !== participantId) {
            return undefined;
          }
          return completed.completed;
        },
        () => this.events.emit(COORDINATION_CHANNELS.complete, freezeEvent({
          protocolVersion: COORDINATION_PROTOCOL_VERSION,
          requestId,
          participantId,
          outcome,
          continuationExpected: effectiveContinuationExpected,
        })),
      );
      return response === true;
    }));
    return results.every(Boolean);
  }

  private async waitForAck(
    channel: string,
    timeoutMs: number,
    parse: (data: unknown) => boolean | undefined,
    send: () => void,
  ): Promise<boolean | undefined> {
    if (this.closed) return undefined;
    return await new Promise<boolean | undefined>((resolve) => {
      let settled = false;
      let timer: NodeJS.Timeout | undefined;
      const finish = (value: boolean | undefined): void => {
        if (settled) return;
        settled = true;
        if (timer !== undefined) clearTimeout(timer);
        unsubscribe();
        this.pendingCancellations.delete(cancel);
        resolve(value);
      };
      const cancel = (): void => finish(undefined);
      const unsubscribe = this.events.on(channel, (data) => {
        const value = parse(data);
        if (value !== undefined) finish(value);
      });
      timer = setTimeout(cancel, timeoutMs);
      this.pendingCancellations.add(cancel);
      try {
        send();
      } catch {
        finish(undefined);
      }
    });
  }
}