import fs from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
import pg from "pg";
import { createLogger, type AgentDecision, type ObservedEntity, type ObservedMedia, type PageState } from "@src/shared";
import type { EventBus, SocialEvent } from "@src/events";

const log = createLogger("storage");
const here = path.dirname(fileURLToPath(import.meta.url));

/** PostgreSQL jsonb rejects   (and lone surrogates); page text sometimes contains them. */
function jsonb(value: unknown): string {
  return JSON.stringify(value).replace(/\\u0000/g, "").replace(/[\uD800-\uDBFF](?![\uDC00-\uDFFF])|(?<![\uD800-\uDBFF])[\uDC00-\uDFFF]/g, "");
}

/**
 * Persistence Layer (§35–§38). PostgreSQL for canonical + raw data; the JSONL log (events package)
 * remains the replay source. When no database is configured the crawler still runs (JSONL only).
 */
export class PostgresStore {
  readonly pool: pg.Pool;
  constructor(databaseUrl: string) {
    this.pool = new pg.Pool({ connectionString: databaseUrl, max: 4 });
  }

  static async connect(databaseUrl: string): Promise<PostgresStore> {
    const s = new PostgresStore(databaseUrl);
    await s.pool.query("select 1");
    return s;
  }

  async migrate(): Promise<void> {
    const sql = fs.readFileSync(path.join(here, "schema.sql"), "utf8");
    await this.pool.query(sql);
    log.info("schema applied");
  }

  async close(): Promise<void> {
    await this.pool.end();
  }

  /** Subscribe to the bus: raw observations + a few derived tables. */
  attach(bus: EventBus): () => void {
    return bus.on("*", (ev) => this.onEvent(ev));
  }

  private async onEvent(ev: SocialEvent): Promise<void> {
    const q = this.pool;
    if (ev.event_type === "SESSION_STARTED") {
      await q.query(
        `insert into sessions(session_id, platform, account_alias, mode, goal, started_at, health) values ($1,$2,$3,$4,$5,$6,'healthy')
         on conflict (session_id) do update set started_at=excluded.started_at, health='healthy', mode=coalesce(sessions.mode, excluded.mode), goal=coalesce(sessions.goal, excluded.goal)`,
        [ev.session_id, ev.platform, String(ev.payload.account_alias ?? ""), ev.payload.mode ?? null, ev.payload.goal ?? null, ev.timestamp],
      );
    }
    await q.query(
      `insert into observations(event_id, session_id, platform, event_type, step, ts, payload, provenance, discovered_via) values ($1,$2,$3,$4,$5,$6,$7,$8,$9) on conflict do nothing`,
      [ev.event_id, ev.session_id, ev.platform, ev.event_type, ev.step ?? null, ev.timestamp, jsonb(ev.payload), ev.provenance ? jsonb(ev.provenance) : null, ev.discovered_via ? jsonb(ev.discovered_via) : null],
    );
    switch (ev.event_type) {
      case "SESSION_ENDED":
        await q.query(`update sessions set ended_at=$2, health=$3, stats=$4 where session_id=$1`, [ev.session_id, ev.timestamp, String(ev.payload.health ?? "stopped"), jsonb(ev.payload)]);
        break;
      case "NETWORK_RESPONSE_OBSERVED": {
        const p = ev.payload as Record<string, unknown>;
        await q.query(
          `insert into network_responses(request_id, session_id, step, url, method, status, kind, content_type, body_size, shape_hash, hostname, path_pattern, graphql_operation, entity_count, entity_types, confidence, ts)
           values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) on conflict do nothing`,
          [p.request_id, ev.session_id, ev.step ?? null, p.url, p.method, p.status, p.kind, p.content_type, p.body_size, p.shape_hash, p.hostname, p.path_pattern, p.graphql_operation ?? null, p.entity_count, p.entity_types ?? [], p.confidence, ev.timestamp],
        );
        break;
      }
      case "NETWORK_SCHEMA_DISCOVERED": {
        const p = ev.payload as Record<string, unknown>;
        const fp = p.fingerprint as { response_shape_hash: string };
        await q.query(
          `insert into schema_patterns(platform, shape_hash, fingerprint, schema, sample_url) values ($1,$2,$3,$4,$5)
           on conflict (platform, shape_hash) do update set observed_count = schema_patterns.observed_count + 1, last_seen = now()`,
          [ev.platform, fp.response_shape_hash, jsonb(p.fingerprint), jsonb(p.schema), p.sample_url ?? null],
        );
        break;
      }
      default:
        break;
    }
  }

  async upsertEntities(entities: ObservedEntity[], sessionId: string, step: number, ts: string): Promise<void> {
    for (const e of entities) {
      const conf = Math.max(0, ...e.provenance.map((p) => p.confidence));
      await this.pool.query(
        `insert into entities(fingerprint, platform, entity_type, platform_id, url, name, text_excerpt, author, metrics, media, fields, confidence, first_seen, last_seen, first_session)
         values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$13,$14)
         on conflict (fingerprint) do update set
           name = coalesce(nullif(excluded.name,''), entities.name),
           text_excerpt = coalesce(excluded.text_excerpt, entities.text_excerpt),
           author = coalesce(excluded.author, entities.author),
           url = coalesce(excluded.url, entities.url),
           metrics = coalesce(entities.metrics,'{}'::jsonb) || coalesce(excluded.metrics,'{}'::jsonb),
           media = coalesce(entities.media,'{}'::jsonb) || coalesce(excluded.media,'{}'::jsonb),
           fields = entities.fields || excluded.fields,
           confidence = greatest(entities.confidence, excluded.confidence),
           last_seen = excluded.last_seen,
           seen_count = entities.seen_count + 1`,
        [e.fingerprint, e.platform, e.type, e.platform_id ?? null, e.url ?? null, e.name ?? null, e.text ?? null, e.author ?? null, e.metrics ? jsonb(e.metrics) : null, e.media ? jsonb(e.media) : null, jsonb(e.fields), conf, ts, sessionId],
      );
      await this.pool.query(`insert into entity_observations(fingerprint, session_id, step, ts, surfaces, snapshot) values ($1,$2,$3,$4,$5,$6)`, [e.fingerprint, sessionId, step, ts, [...new Set(e.provenance.map((p) => p.surface))], jsonb({ name: e.name, text: e.text, author: e.author, metrics: e.metrics, context: e.context, provenance: e.provenance })]);
    }
  }

  async upsertMedia(media: ObservedMedia[], frames: Record<string, string[]>, ts: string): Promise<void> {
    for (const m of media) {
      await this.pool.query(
        `insert into media(fingerprint, platform, media_type, platform_media_id, page_url, url, title, author, duration_s, width, height, thumbnail_url, delivery, frames, provenance, first_seen, last_seen)
         values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$16)
         on conflict (fingerprint) do update set
           title = coalesce(excluded.title, media.title), author = coalesce(excluded.author, media.author),
           duration_s = coalesce(excluded.duration_s, media.duration_s), width = coalesce(excluded.width, media.width), height = coalesce(excluded.height, media.height),
           thumbnail_url = coalesce(excluded.thumbnail_url, media.thumbnail_url), delivery = coalesce(excluded.delivery, media.delivery),
           frames = case when array_length(excluded.frames,1) > 0 then excluded.frames else media.frames end, last_seen = excluded.last_seen`,
        [m.fingerprint, m.platform, m.media_type, m.platform_media_id ?? null, m.page_url ?? null, m.url ?? null, m.title ?? null, m.author ?? null, m.duration_s ?? null, m.width ?? null, m.height ?? null, m.thumbnail_url ?? null, m.delivery ? jsonb(m.delivery) : null, frames[m.fingerprint] ?? [], jsonb(m.provenance), ts],
      );
    }
  }

  async recordFeedItems(sessionId: string, step: number, state: PageState, ts: string): Promise<void> {
    if (!["HOME_FEED", "SEARCH_RESULTS", "GROUP", "CHANNEL", "PROFILE"].includes(state.classification.page_type)) return;
    let pos = 0;
    for (const e of state.entities) {
      if (!["video", "post", "comment"].includes(e.type)) continue;
      await this.pool.query(`insert into feed_items(session_id, step, page_url, page_type, feed_position, fingerprint, entity_type, visible, ts) values ($1,$2,$3,$4,$5,$6,$7,$8,$9)`, [sessionId, step, state.url, state.classification.page_type, pos++, e.fingerprint, e.type, /visible/.test(e.context ?? ""), ts]);
    }
  }

  async recordAction(sessionId: string, decision: AgentDecision, before: Record<string, unknown>, after: Record<string, unknown> | null, success: boolean, error: string | undefined, durationMs: number): Promise<void> {
    const a = decision.chosen_action;
    await this.pool.query(
      `insert into actions(session_id, step, action_id, action_type, label, target_url, planner, expected_gain, novelty, relevance, reason, scores, before_state, after_state, success, error, duration_ms)
       values ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)`,
      [sessionId, decision.step, a.id, a.type, a.label, a.target_url ?? null, decision.planner, decision.expected_information_gain, decision.novelty, decision.relevance, decision.reason, jsonb(decision.scores), jsonb(before), after ? jsonb(after) : null, success, error ?? null, durationMs],
    );
  }

  async recordRelationships(sessionId: string, edges: { from: string; to: string; type: string; step: number }[]): Promise<void> {
    for (const e of edges) await this.pool.query(`insert into relationships(session_id, from_fp, to_fp, rel_type, step) values ($1,$2,$3,$4,$5) on conflict do nothing`, [sessionId, e.from, e.to, e.type, e.step]);
  }

  /** Bind a crawl job (created by the CLI or by the API job runner) to its browser session. */
  async linkJob(job: { job_id: string; platform: string; account_alias: string; mode: string; goal: string; budget: unknown }, sessionId: string): Promise<void> {
    // The SESSION_STARTED row may not be inserted yet (bus handlers are async): upsert the session here so goal/mode are never lost.
    await this.pool.query(
      `insert into sessions(session_id, platform, account_alias, mode, goal, health) values ($1,$2,$3,$4,$5,'starting')
       on conflict (session_id) do update set mode=excluded.mode, goal=excluded.goal`,
      [sessionId, job.platform, job.account_alias, job.mode, job.goal],
    );
    await this.pool.query(
      `insert into crawl_jobs(job_id, session_id, platform, mode, goal, budget, status) values ($1,$2,$3,$4,$5,$6,'running')
       on conflict (job_id) do update set session_id=excluded.session_id, status='running'`,
      [job.job_id, sessionId, job.platform, job.mode, job.goal, jsonb(job.budget)],
    );
  }

  async recordConnectorVersion(platform: string, summary: Record<string, unknown>, manifestPath: string): Promise<void> {
    await this.pool.query(`insert into connector_versions(platform, confidence, summary, manifest_path) values ($1,$2,$3,$4)`, [platform, Number(summary.confidence ?? 0), jsonb(summary), manifestPath]);
  }

  // ---- dashboard queries ----
  async listSessions(limit = 50) {
    return (await this.pool.query(`select s.*, (select count(*) from actions a where a.session_id=s.session_id) as actions, (select count(distinct fingerprint) from entity_observations eo where eo.session_id=s.session_id) as entities from sessions s order by started_at desc limit $1`, [limit])).rows;
  }
  async sessionEvents(sessionId: string, opts: { types?: string[]; limit?: number; afterStep?: number } = {}) {
    const params: unknown[] = [sessionId];
    let where = "session_id=$1";
    if (opts.types?.length) {
      params.push(opts.types);
      where += ` and event_type = any($${params.length})`;
    }
    if (opts.afterStep !== undefined) {
      params.push(opts.afterStep);
      where += ` and step > $${params.length}`;
    }
    params.push(opts.limit ?? 300);
    return (await this.pool.query(`select event_id, event_type, step, ts, payload, provenance, discovered_via from observations where ${where} order by ts desc limit $${params.length}`, params)).rows;
  }
  async sessionActions(sessionId: string) {
    return (await this.pool.query(`select * from actions where session_id=$1 order by step`, [sessionId])).rows;
  }
  async sessionEntities(sessionId: string, limit = 500) {
    return (await this.pool.query(`select e.* , max(eo.step) as last_step from entities e join entity_observations eo on eo.fingerprint=e.fingerprint where eo.session_id=$1 group by e.fingerprint order by last_step desc, e.last_seen desc limit $2`, [sessionId, limit])).rows;
  }
  async sessionMedia(sessionId: string) {
    return (await this.pool.query(`select m.* from media m where exists (select 1 from entity_observations eo where eo.fingerprint=m.fingerprint and eo.session_id=$1) or m.page_url in (select distinct payload->>'url' from observations where session_id=$1 and event_type='PAGE_OPENED') order by last_seen desc limit 200`, [sessionId])).rows;
  }
  async sessionRelationships(sessionId: string) {
    return (await this.pool.query(`select from_fp, to_fp, rel_type, step from relationships where session_id=$1 limit 2000`, [sessionId])).rows;
  }
  async schemas(platform?: string) {
    return (await this.pool.query(`select * from schema_patterns ${platform ? "where platform=$1" : ""} order by observed_count desc limit 200`, platform ? [platform] : [])).rows;
  }
  async entityStats() {
    return (await this.pool.query(`select platform, entity_type, count(*)::int as n from entities group by 1,2 order by 1,3 desc`)).rows;
  }
  async entity(fingerprint: string) {
    const e = (await this.pool.query(`select * from entities where fingerprint=$1`, [fingerprint])).rows[0];
    const obs = (await this.pool.query(`select session_id, step, ts, surfaces, snapshot from entity_observations where fingerprint=$1 order by ts desc limit 50`, [fingerprint])).rows;
    return { entity: e, observations: obs };
  }
}

export async function openStore(databaseUrl: string | undefined): Promise<PostgresStore | undefined> {
  if (!databaseUrl) {
    log.warn("SRC_DATABASE_URL not set — running with the JSONL event log only");
    return undefined;
  }
  try {
    const s = await PostgresStore.connect(databaseUrl);
    await s.migrate();
    return s;
  } catch (err) {
    log.error("cannot connect to PostgreSQL — continuing with JSONL only", { err: (err as Error).message });
    return undefined;
  }
}