import { Inject, Injectable } from "@nestjs/common";
import { EventEmitter } from "node:events";
import { v4 as uuid } from "uuid";
import { EVENTS_STREAM, EVENTS_STREAM_MAXLEN } from "../constants.js";
import { logger } from "./logger.js";
import { RedisService } from "./redis.service.js";

export interface DomainEventPayload {
  workspaceId?: string;
  [key: string]: unknown;
}

/**
 * Domain event bus.
 *
 * `publish(event, payload)` appends to the Redis stream `factory:events`
 * (XADD, MAXLEN ~ 10000) and mirrors it on an in-process EventEmitter so
 * local consumers and tests can listen without a live Redis.
 *
 * Workers (BullMQ side) consume the stream for search indexing, webhooks,
 * automation runs and notification fan-out. Publishing must never fail a
 * request: Redis outages log a warning and drop the stream write.
 */
@Injectable()
export class EventBus {
  private readonly local = new EventEmitter();
  static readonly localStream = "local";

  constructor(@Inject(RedisService) private readonly redis: RedisService) {
    this.local.setMaxListeners(200);
  }

  async publish(event: string, payload: DomainEventPayload = {}): Promise<void> {
    const body = {
      event,
      payload: JSON.stringify(payload),
      ts: new Date().toISOString(),
      id: uuid()
    };
    // Local mirror first (tests, in-process consumers).
    try {
      this.local.emit(EventBus.localStream, body);
      this.local.emit(event, body);
    } catch (err) {
      logger.warn(`event bus local emit failed for ${event}: ${(err as Error).message}`);
    }
    // Stream append (best effort).
    try {
      await this.redis.getRedis().xadd(
        EVENTS_STREAM,
        "MAXLEN",
        "~",
        String(EVENTS_STREAM_MAXLEN),
        "*",
        "event",
        body.event,
        "payload",
        body.payload,
        "ts",
        body.ts
      );
    } catch (err) {
      logger.warn(`event bus stream publish failed for ${event}: ${(err as Error).message}`);
    }
  }

  /** Subscribe to local events (tests / in-process). Returns unsubscribe. */
  onLocal(event: string | typeof EventBus.localStream, handler: (body: { event: string; payload: string; ts: string; id: string }) => void): () => void {
    this.local.on(event, handler);
    return () => this.local.off(event, handler);
  }
}