import {
  Inject,
  Injectable,
  type OnModuleDestroy,
  type OnModuleInit,
} from "@nestjs/common";
import { and, eq, isNull, lte, or, type SQL } from "drizzle-orm";
import type { OutboxRow, OutboxTable } from "./schema.js";
import {
  OUTBOX_CONFIG,
  OUTBOX_DB,
  OUTBOX_KAFKA_PRODUCER,
  OUTBOX_LOGGER,
  OUTBOX_TABLE,
  type OutboxConfig,
  type OutboxDbClient,
  type OutboxKafkaProducer,
  type OutboxLogger,
} from "./types.js";

/**
 * OutboxPublisher —— 轮询 pending 记录并投递到 Kafka。
 *
 * 实现 at-least-once 投递语义：
 * 1. 周期性（pollIntervalMs）批量拉取 status=pending 且到期的记录
 * 2. 逐条投递到 Kafka（message key = event id，保证同聚合有序）
 * 3. 投递成功 → 标记 published
 * 4. 投递失败 → 指数退避更新 nextRetryAt；重试耗尽 → 标记 failed
 *
 * 幂等性：
 * - producer 应由调用方配置为 idempotent（`kafka.producer({ idempotent: true, transactionalId })`），
 *   避免生产端重试产生重复消息
 * - message key = event id（cuid2），消费端可基于 event_id 去重（Redis SETNX 或 DB 唯一索引）
 */
@Injectable()
export class OutboxPublisher implements OnModuleInit, OnModuleDestroy {
  private intervalId: ReturnType<typeof setInterval> | null = null;
  private isPolling = false;

  constructor(
    @Inject(OUTBOX_DB) private readonly db: OutboxDbClient,
    @Inject(OUTBOX_TABLE) private readonly table: OutboxTable,
    @Inject(OUTBOX_KAFKA_PRODUCER)
    private readonly producer: OutboxKafkaProducer,
    @Inject(OUTBOX_CONFIG) private readonly config: OutboxConfig,
    @Inject(OUTBOX_LOGGER) private readonly logger: OutboxLogger,
  ) {}

  async onModuleInit(): Promise<void> {
    await this.start();
  }

  async onModuleDestroy(): Promise<void> {
    await this.stop();
  }

  async start(): Promise<void> {
    if (this.intervalId) return;
    this.logger.info(
      {
        topic: this.config.kafkaTopic,
        pollIntervalMs: this.config.pollIntervalMs,
        batchSize: this.config.batchSize,
      },
      "OutboxPublisher started",
    );
    this.intervalId = setInterval(() => {
      void this.poll();
    }, this.config.pollIntervalMs);
  }

  async stop(): Promise<void> {
    if (this.intervalId) {
      clearInterval(this.intervalId);
      this.intervalId = null;
    }
    this.logger.info("OutboxPublisher stopped");
  }

  /**
   * 拉取一批到期 pending 记录并逐条投递。
   * 通过 isPolling 标志保证不会并发轮询。
   */
  private async poll(): Promise<void> {
    if (this.isPolling) return;
    this.isPolling = true;
    try {
      const now = new Date();
      const where: SQL | undefined = and(
        eq(this.table.status, "pending"),
        or(isNull(this.table.nextRetryAt), lte(this.table.nextRetryAt, now)),
      );
      const messages: OutboxRow[] = await this.db
        .select()
        .from(this.table)
        .where(where)
        .limit(this.config.batchSize);

      for (const message of messages) {
        await this.dispatch(message);
      }
    } catch (error) {
      this.logger.error({ error }, "Outbox poll failed");
    } finally {
      this.isPolling = false;
    }
  }

  /**
   * 投递单条记录到 Kafka 并更新状态。
   */
  private async dispatch(message: OutboxRow): Promise<void> {
    try {
      await this.producer.send({
        topic: this.config.kafkaTopic,
        messages: [
          {
            key: message.id,
            value: message.payload,
            headers: this.buildHeaders(message),
          },
        ],
      });

      await this.db
        .update(this.table)
        .set({ status: "published", publishedAt: new Date(), lastError: null })
        .where(eq(this.table.id, message.id));

      this.logger.info(
        {
          id: message.id,
          eventType: message.eventType,
          topic: this.config.kafkaTopic,
        },
        "Outbox message published",
      );
    } catch (error) {
      await this.handleFailure(message, this.toErrorMessage(error));
    }
  }

  /**
   * 失败处理：指数退避更新 nextRetryAt，重试耗尽则标记 failed。
   */
  private async handleFailure(
    message: OutboxRow,
    errorMessage: string,
  ): Promise<void> {
    const newRetryCount = message.retryCount + 1;
    this.logger.error(
      {
        id: message.id,
        eventType: message.eventType,
        retryCount: newRetryCount,
        error: errorMessage,
      },
      "Outbox publish failed",
    );

    if (newRetryCount >= message.maxRetryCount) {
      await this.db
        .update(this.table)
        .set({
          status: "failed",
          retryCount: newRetryCount,
          lastError: errorMessage,
        })
        .where(eq(this.table.id, message.id));
      return;
    }

    const backoffMs = this.config.retryBackoffMs * 2 ** newRetryCount;
    const nextRetryAt = new Date(Date.now() + backoffMs);
    await this.db
      .update(this.table)
      .set({
        retryCount: newRetryCount,
        nextRetryAt,
        lastError: errorMessage,
      })
      .where(eq(this.table.id, message.id));
  }

  /**
   * 构建 Kafka 消息头：eventId / eventType / aggregateId + 调用方 metadata。
   */
  private buildHeaders(message: OutboxRow): Record<string, string> {
    const headers: Record<string, string> = {
      eventId: message.id,
      eventType: message.eventType,
      aggregateId: message.aggregateId,
    };
    if (message.metadata) {
      for (const [key, value] of Object.entries(message.metadata)) {
        headers[key] = value;
      }
    }
    return headers;
  }

  private toErrorMessage(error: unknown): string {
    return error instanceof Error ? error.message : String(error);
  }
}