import fs from "node:fs";
import path from "node:path";
import { newId, nowIso, type Platform, type Provenance } from "@src/shared";

/** Universal Social Event Format (§33, §34). */
export type SocialEventType =
  | "SESSION_STARTED"
  | "SESSION_ENDED"
  | "PAGE_OPENED"
  | "PAGE_CLASSIFIED"
  | "ENTITY_DISCOVERED"
  | "ENTITY_OBSERVED"
  | "ENTITY_UPDATED"
  | "POST_DISCOVERED"
  | "COMMENT_DISCOVERED"
  | "PROFILE_DISCOVERED"
  | "MEDIA_DISCOVERED"
  | "VIDEO_DISCOVERED"
  | "IMAGE_DISCOVERED"
  | "FEED_ITEM_OBSERVED"
  | "NETWORK_RESPONSE_OBSERVED"
  | "NETWORK_SCHEMA_DISCOVERED"
  | "WEBSOCKET_FRAME_OBSERVED"
  | "DOM_CHANGED"
  | "ACTION_PLANNED"
  | "ACTION_EXECUTED"
  | "ACTION_FAILED"
  | "NAVIGATION_COMPLETED"
  | "LOOP_DETECTED"
  | "BUDGET_EXHAUSTED"
  | "AUTH_REQUIRED"
  | "CONNECTOR_PATTERN_LEARNED"
  | "CONNECTOR_DEGRADED"
  | "CONNECTOR_REPAIRED"
  | "WORKER_ERROR";

export interface SocialEvent<T = Record<string, unknown>> {
  event_id: string;
  event_type: SocialEventType;
  platform: Platform;
  session_id: string;
  step?: number;
  timestamp: string;
  payload: T;
  provenance?: Provenance[];
  discovered_via?: { action?: string; action_id?: string; url?: string };
}

export type EventHandler = (ev: SocialEvent) => void | Promise<void>;

/**
 * Typed in-process event bus. Observers publish, storage / learner / dashboard subscribe.
 * Handlers are awaited sequentially so persistence stays ordered.
 */
export class EventBus {
  private handlers: { type: SocialEventType | "*"; fn: EventHandler }[] = [];
  private queue: Promise<void> = Promise.resolve();
  public count = 0;

  on(type: SocialEventType | "*", fn: EventHandler): () => void {
    const h = { type, fn };
    this.handlers.push(h);
    return () => {
      this.handlers = this.handlers.filter((x) => x !== h);
    };
  }

  emit<T extends Record<string, unknown>>(
    partial: Omit<SocialEvent<T>, "event_id" | "timestamp"> & { event_id?: string; timestamp?: string },
  ): SocialEvent<T> {
    const ev: SocialEvent<T> = {
      event_id: partial.event_id ?? newId("ev"),
      timestamp: partial.timestamp ?? nowIso(),
      ...partial,
    };
    this.count++;
    const targets = this.handlers.filter((h) => h.type === "*" || h.type === ev.event_type);
    this.queue = this.queue.then(async () => {
      for (const h of targets) {
        try {
          await h.fn(ev as SocialEvent);
        } catch (err) {
          process.stderr.write(`[events] handler error on ${ev.event_type}: ${(err as Error).message}\n`);
        }
      }
    });
    return ev;
  }

  /** Wait for all queued handlers (call before shutdown). */
  flush(): Promise<void> {
    return this.queue;
  }
}

/** Append-only JSONL session log — the raw record every replay (§54) is built from. */
export class JsonlEventLog {
  private stream: fs.WriteStream;
  readonly file: string;

  constructor(sessionDir: string) {
    fs.mkdirSync(sessionDir, { recursive: true });
    this.file = path.join(sessionDir, "events.jsonl");
    this.stream = fs.createWriteStream(this.file, { flags: "a" });
  }

  write(ev: SocialEvent): void {
    this.stream.write(JSON.stringify(ev) + "\n");
  }

  attach(bus: EventBus): () => void {
    return bus.on("*", (ev) => this.write(ev));
  }

  close(): Promise<void> {
    return new Promise((res) => this.stream.end(res));
  }

  static read(sessionDir: string): SocialEvent[] {
    const file = path.join(sessionDir, "events.jsonl");
    if (!fs.existsSync(file)) return [];
    return fs
      .readFileSync(file, "utf8")
      .split("\n")
      .filter(Boolean)
      .map((l) => JSON.parse(l) as SocialEvent);
  }
}