/**
 * WorthDoing.ai
 * Author: Simon-Pierre Boucher
 * Contact: contact@spboucher.ai
 * File: src/lib/agent/events.ts
 * Description: Agent event bus — persists AgentEvents to Postgres and publishes them in-process for SSE.
 */
import { EventEmitter } from "events";
import { and, asc, eq, gt } from "drizzle-orm";
import { db } from "@/lib/db/client";
import { agentEvents } from "@/lib/db/schema";

export type AgentEventType =
  | "investigation.started"
  | "agent.plan"
  | "agent.error"
  | "phase.changed"
  | "budget.updated"
  | "search.started"
  | "search.completed"
  | "scrape.started"
  | "scrape.completed"
  | "scrape.failed"
  | "crawl.started"
  | "crawl.completed"
  | "crawl.failed"
  | "extract.started"
  | "extract.completed"
  | "extract.failed"
  | "hypothesis.created"
  | "hypothesis.updated"
  | "hypothesis.rejected"
  | "evidence.saved"
  | "claim.created"
  | "opportunity.created"
  | "opportunity.updated"
  | "report.started"
  | "report.delta"
  | "report.completed"
  | "investigation.completed"
  | "investigation.failed";

export type AgentEvent = {
  seq: number;
  investigationId: string;
  type: AgentEventType;
  payload: Record<string, unknown>;
  createdAt: string;
};

const globalForBus = globalThis as unknown as {
  __wdBus?: EventEmitter;
  __wdSeq?: Map<string, number>;
};
const bus = (globalForBus.__wdBus ??= new EventEmitter());
bus.setMaxListeners(500);
const seqCounters = (globalForBus.__wdSeq ??= new Map<string, number>());

async function nextSeq(investigationId: string): Promise<number> {
  const current = seqCounters.get(investigationId);
  if (current !== undefined) {
    seqCounters.set(investigationId, current + 1);
    return current + 1;
  }
  const rows = await db
    .select({ seq: agentEvents.seq })
    .from(agentEvents)
    .where(eq(agentEvents.investigationId, investigationId))
    .orderBy(asc(agentEvents.seq));
  const max = rows.length ? rows[rows.length - 1].seq : 0;
  seqCounters.set(investigationId, max + 1);
  return max + 1;
}

/**
 * Persist and publish a real agent event. Every UI event flows through here —
 * nothing user-visible is ever fabricated.
 */
export async function emitEvent(
  investigationId: string,
  type: AgentEventType,
  payload: Record<string, unknown>,
  options: { transient?: boolean } = {},
): Promise<AgentEvent> {
  const seq = await nextSeq(investigationId);
  const event: AgentEvent = {
    seq,
    investigationId,
    type,
    payload,
    createdAt: new Date().toISOString(),
  };
  // report.delta events are high-frequency; publish live but skip per-delta persistence
  // (the full report is persisted on report.completed).
  if (!options.transient) {
    await db.insert(agentEvents).values({ investigationId, seq, type, payload });
  }
  bus.emit(`inv:${investigationId}`, event);
  return event;
}

export function subscribe(investigationId: string, listener: (e: AgentEvent) => void): () => void {
  const channel = `inv:${investigationId}`;
  bus.on(channel, listener);
  return () => bus.off(channel, listener);
}

/** Replay persisted events after a given seq (for SSE Last-Event-ID reconnects). */
export async function eventsAfter(investigationId: string, afterSeq: number): Promise<AgentEvent[]> {
  const rows = await db
    .select()
    .from(agentEvents)
    .where(and(eq(agentEvents.investigationId, investigationId), gt(agentEvents.seq, afterSeq)))
    .orderBy(asc(agentEvents.seq));
  return rows.map((r) => ({
    seq: r.seq,
    investigationId: r.investigationId,
    type: r.type as AgentEventType,
    payload: r.payload as Record<string, unknown>,
    createdAt: r.createdAt.toISOString(),
  }));
}