import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import type { OutboxEvent } from "../../src/shared/outbox/outbox.schema.js";

// ============================================================
// Mock 外部依赖 —— 使用 vi.hoisted 确保 mock 变量在 hoisted 的 vi.mock 中可用
// ============================================================

const mocks = vi.hoisted(() => {
  const mockProducer = {
    send: vi.fn(),
  };
  const mockFindPending = vi.fn();
  const mockMarkPublished = vi.fn();
  const mockMarkFailed = vi.fn();
  const mockIncrementRetry = vi.fn();
  return {
    mockProducer,
    mockFindPending,
    mockMarkPublished,
    mockMarkFailed,
    mockIncrementRetry,
  };
});

// Mock logger
vi.mock("../../src/shared/observability/logger.js", () => ({
  logger: {
    info: vi.fn(),
    warn: vi.fn(),
    error: vi.fn(),
    debug: vi.fn(),
  },
}));

// Mock Kafka client
vi.mock("../../src/shared/kafka/kafka.client.js", () => ({
  getProducer: () => mocks.mockProducer,
  isKafkaHealthy: vi.fn(() => true),
}));

// Mock topic-map
vi.mock("../../src/shared/kafka/topic-map.js", () => ({
  resolveTopic: vi.fn((eventType: string) => {
    const map: Record<string, string> = {
      "notification.sent": "edu.notification.sent",
      "notification.read": "edu.notification.read",
      "notification.recalled": "edu.notification.recalled",
      "notification.failed": "edu.notification.failed",
    };
    return map[eventType] ?? "edu.notification.events";
  }),
}));

// Mock outbox repository
vi.mock("../../src/shared/outbox/outbox.repository.js", () => ({
  findPending: mocks.mockFindPending,
  markPublished: mocks.mockMarkPublished,
  markFailed: mocks.mockMarkFailed,
  incrementRetry: mocks.mockIncrementRetry,
}));

// 导入被测模块（在 mock 之后）
import { outboxPublisher } from "../../src/shared/outbox/outbox.publisher.js";

// ============================================================
// 辅助
// ============================================================

const {
  mockProducer,
  mockFindPending,
  mockMarkPublished,
  mockMarkFailed,
  mockIncrementRetry,
} = mocks;

function createMessage(overrides: Partial<OutboxEvent> = {}): OutboxEvent {
  return {
    eventId: "evt-1",
    aggregateType: "Notification",
    aggregateId: "agg-1",
    eventType: "notification.sent",
    topic: "edu.notification.sent",
    payload: { notificationId: "n1", userId: "u1" },
    status: "pending",
    retryCount: 0,
    maxRetryCount: 5,
    createdAt: new Date("2026-01-01T00:00:00Z"),
    publishedAt: null,
    nextRetryAt: null,
    lastError: null,
    metadata: null,
    ...overrides,
  } as OutboxEvent;
}

// 访问私有方法
function poll(): Promise<void> {
  return (outboxPublisher as unknown as { poll: () => Promise<void> }).poll();
}

function dispatch(message: OutboxEvent): Promise<void> {
  return (
    outboxPublisher as unknown as {
      dispatch: (m: OutboxEvent) => Promise<void>;
    }
  ).dispatch(message);
}

// ============================================================
// Tests
// ============================================================

describe("OutboxPublisher", () => {
  beforeEach(async () => {
    vi.clearAllMocks();
    vi.useFakeTimers();
    // 重置 singleton 状态：确保 intervalId 被清除
    await outboxPublisher.stop();
    mockProducer.send.mockResolvedValue({} as never);
    mockMarkPublished.mockResolvedValue(undefined);
    mockMarkFailed.mockResolvedValue(undefined);
    mockIncrementRetry.mockResolvedValue(undefined);
    mockFindPending.mockResolvedValue([]);
  });

  afterEach(() => {
    vi.useRealTimers();
  });

  // ----------------------------------------------------------
  // start / stop
  // ----------------------------------------------------------
  describe("start / stop", () => {
    it("start 应设置定时轮询", async () => {
      await outboxPublisher.start();

      // 推进定时器触发 poll
      await vi.advanceTimersByTimeAsync(5000);

      expect(mockFindPending).toHaveBeenCalled();
    });

    it("重复 start 不应创建多个定时器", async () => {
      await outboxPublisher.start();
      await outboxPublisher.start();

      // 推进 5 秒，应只 poll 一次（第二个 start 是 no-op）
      await vi.advanceTimersByTimeAsync(5000);

      // 只有一个 interval，poll 应只被调用一次
      expect(mockFindPending).toHaveBeenCalledTimes(1);
    });

    it("stop 应清除定时器", async () => {
      await outboxPublisher.start();
      await outboxPublisher.stop();

      mockFindPending.mockClear();
      await vi.advanceTimersByTimeAsync(10000);

      expect(mockFindPending).not.toHaveBeenCalled();
    });

    it("无定时器时 stop 不应报错", async () => {
      await outboxPublisher.stop();
      // 不抛出即可
    });
  });

  // ----------------------------------------------------------
  // poll
  // ----------------------------------------------------------
  describe("poll", () => {
    it("findPending 返回消息时应逐条 dispatch", async () => {
      const messages = [
        createMessage({ eventId: "evt-1" }),
        createMessage({ eventId: "evt-2" }),
      ];
      mockFindPending.mockResolvedValue(messages);

      await poll();

      expect(mockFindPending).toHaveBeenCalledWith(100); // BATCH_SIZE
      expect(mockProducer.send).toHaveBeenCalledTimes(2);
      expect(mockMarkPublished).toHaveBeenCalledTimes(2);
      expect(mockMarkPublished).toHaveBeenCalledWith("evt-1");
      expect(mockMarkPublished).toHaveBeenCalledWith("evt-2");
    });

    it("findPending 返回空数组时不应 dispatch", async () => {
      mockFindPending.mockResolvedValue([]);

      await poll();

      expect(mockProducer.send).not.toHaveBeenCalled();
      expect(mockMarkPublished).not.toHaveBeenCalled();
    });

    it("findPending 抛出异常时应记录日志不中断", async () => {
      mockFindPending.mockRejectedValue(new Error("DB connection lost"));

      // 不应抛出
      await poll();

      // logger.error 应被调用（由 mock 拦截）
    });

    it("并发 poll 保护：正在 poll 时不应重复执行", async () => {
      const messages = [createMessage()];
      // 让 producer.send 返回一个未完成的 Promise
      let resolveSend: () => void;
      mockProducer.send.mockReturnValue(
        new Promise((resolve) => {
          resolveSend = resolve as () => void;
        }),
      );
      mockFindPending.mockResolvedValue(messages);

      // 启动第一次 poll（未完成）
      const firstPoll = poll();

      // 尝试第二次 poll（应被跳过）
      await poll();

      // 只有第一次的 findPending 被调用了一次
      // （第二次 poll 因为 isPolling=true 直接返回）
      expect(mockFindPending).toHaveBeenCalledTimes(1);

      // 完成
      resolveSend!();
      await firstPoll;
    });
  });

  // ----------------------------------------------------------
  // dispatch —— 成功路径
  // ----------------------------------------------------------
  describe("dispatch 成功", () => {
    it("应发送到正确的 topic 并标记 published", async () => {
      const message = createMessage({
        eventType: "notification.read",
        aggregateId: "notif-1",
      });

      await dispatch(message);

      expect(mockProducer.send).toHaveBeenCalledWith({
        topic: "edu.notification.read",
        messages: [
          {
            key: "notif-1",
            value: JSON.stringify({ notificationId: "n1", userId: "u1" }),
            headers: expect.objectContaining({
              eventId: "evt-1",
              eventType: "notification.read",
              aggregateType: "Notification",
              aggregateId: "notif-1",
            }),
          },
        ],
      });
      expect(mockMarkPublished).toHaveBeenCalledWith("evt-1");
    });

    it("payload 为字符串时应直接使用", async () => {
      const message = createMessage({
        payload: "raw-string-payload" as unknown,
      } as OutboxEvent);

      await dispatch(message);

      const sendArg = mockProducer.send.mock.calls[0][0];
      expect(sendArg.messages[0].value).toBe("raw-string-payload");
    });

    it("payload 为对象时应 JSON 序列化", async () => {
      const payload = { key: "value", num: 42 };
      const message = createMessage({ payload });

      await dispatch(message);

      const sendArg = mockProducer.send.mock.calls[0][0];
      expect(sendArg.messages[0].value).toBe(JSON.stringify(payload));
    });

    it("metadata 应合并到 headers", async () => {
      const message = createMessage({
        metadata: { userId: "u1", source: "test" },
      });

      await dispatch(message);

      const sendArg = mockProducer.send.mock.calls[0][0];
      expect(sendArg.messages[0].headers).toEqual(
        expect.objectContaining({
          eventId: "evt-1",
          eventType: "notification.sent",
          aggregateType: "Notification",
          aggregateId: "agg-1",
          userId: "u1",
          source: "test",
        }),
      );
    });

    it("metadata 为 null 时 headers 只含基础字段", async () => {
      const message = createMessage({ metadata: null });

      await dispatch(message);

      const sendArg = mockProducer.send.mock.calls[0][0];
      const headers = sendArg.messages[0].headers;
      expect(Object.keys(headers)).toEqual(
        expect.arrayContaining([
          "eventId",
          "eventType",
          "aggregateType",
          "aggregateId",
        ]),
      );
      expect(Object.keys(headers)).toHaveLength(4);
    });
  });

  // ----------------------------------------------------------
  // dispatch —— 重试逻辑
  // ----------------------------------------------------------
  describe("dispatch 重试逻辑", () => {
    it("失败且 retryCount < MAX_RETRY → incrementRetry", async () => {
      const message = createMessage({ retryCount: 2 });
      mockProducer.send.mockRejectedValue(new Error("Kafka timeout"));

      await dispatch(message);

      // retryCount=2, +1=3 < MAX_RETRY(5)
      expect(mockIncrementRetry).toHaveBeenCalledTimes(1);
      expect(mockIncrementRetry).toHaveBeenCalledWith(
        "evt-1",
        "Kafka timeout",
        2000 * 2 ** 2, // RETRY_BACKOFF_MS * 2^retryCount = 2000 * 4 = 8000
      );
      expect(mockMarkFailed).not.toHaveBeenCalled();
    });

    it("失败且 retryCount+1 >= MAX_RETRY → markFailed", async () => {
      const message = createMessage({ retryCount: 4 }); // 4+1=5 >= 5
      mockProducer.send.mockRejectedValue(new Error("Kafka down"));

      await dispatch(message);

      expect(mockMarkFailed).toHaveBeenCalledTimes(1);
      expect(mockMarkFailed).toHaveBeenCalledWith("evt-1", "Kafka down");
      expect(mockIncrementRetry).not.toHaveBeenCalled();
    });

    it("失败且 retryCount=5（超过 MAX_RETRY）→ markFailed", async () => {
      const message = createMessage({ retryCount: 5 });
      mockProducer.send.mockRejectedValue(new Error("Still failing"));

      await dispatch(message);

      expect(mockMarkFailed).toHaveBeenCalledWith("evt-1", "Still failing");
    });

    it("非 Error 类型的异常应转字符串", async () => {
      const message = createMessage({ retryCount: 0 });
      mockProducer.send.mockRejectedValue("string error");

      await dispatch(message);

      expect(mockIncrementRetry).toHaveBeenCalledWith(
        "evt-1",
        "string error",
        2000, // 2000 * 2^0 = 2000
      );
    });

    it("指数退避应正确计算：retryCount=0 → 2000ms, retryCount=1 → 4000ms", async () => {
      mockProducer.send.mockRejectedValue(new Error("fail"));

      // retryCount=0 → 2000ms
      await dispatch(createMessage({ retryCount: 0 }));
      expect(mockIncrementRetry).toHaveBeenLastCalledWith(
        "evt-1",
        "fail",
        2000,
      );

      // retryCount=1 → 4000ms
      await dispatch(createMessage({ retryCount: 1 }));
      expect(mockIncrementRetry).toHaveBeenLastCalledWith(
        "evt-1",
        "fail",
        4000,
      );
    });
  });

  // ----------------------------------------------------------
  // 完整 poll → dispatch → markPublished 流程
  // ----------------------------------------------------------
  describe("完整流程", () => {
    it("poll → dispatch 多条 → 全部 markPublished", async () => {
      const messages = [
        createMessage({ eventId: "evt-1", eventType: "notification.sent" }),
        createMessage({ eventId: "evt-2", eventType: "notification.read" }),
        createMessage({
          eventId: "evt-3",
          eventType: "notification.recalled",
        }),
      ];
      mockFindPending.mockResolvedValue(messages);

      await poll();

      expect(mockProducer.send).toHaveBeenCalledTimes(3);
      expect(mockMarkPublished).toHaveBeenCalledTimes(3);

      // 验证每条消息发送到正确 topic
      const topics = mockProducer.send.mock.calls.map(
        (call) => (call[0] as { topic: string }).topic,
      );
      expect(topics).toEqual([
        "edu.notification.sent",
        "edu.notification.read",
        "edu.notification.recalled",
      ]);
    });

    it("poll 中部分 dispatch 失败不应中断后续消息", async () => {
      const messages = [
        createMessage({ eventId: "evt-1" }),
        createMessage({ eventId: "evt-2" }),
        createMessage({ eventId: "evt-3" }),
      ];
      mockFindPending.mockResolvedValue(messages);
      // 第二条失败
      mockProducer.send
        .mockResolvedValueOnce({} as never)
        .mockRejectedValueOnce(new Error("Kafka error"))
        .mockResolvedValueOnce({} as never);

      await poll();

      // 第一条和第三条成功 markPublished，第二条 incrementRetry
      expect(mockMarkPublished).toHaveBeenCalledTimes(2);
      expect(mockMarkPublished).toHaveBeenCalledWith("evt-1");
      expect(mockMarkPublished).toHaveBeenCalledWith("evt-3");
      expect(mockIncrementRetry).toHaveBeenCalledTimes(1);
      expect(mockIncrementRetry).toHaveBeenCalledWith(
        "evt-2",
        "Kafka error",
        2000,
      );
    });
  });
});