import fs from "node:fs";
import path from "node:path";
import type { PostgresStore } from "@src/storage";
import { PLATFORMS, type AppConfig, type Platform } from "@src/shared";
import { PlatformModel } from "@src/platform-model";
import { hasAdapter } from "@src/connectors";

/** Read-side queries for the research console. All read PostgreSQL; files (models, manifests, session dirs) complete them. */
export class Queries {
  constructor(private readonly store: PostgresStore, private readonly cfg: AppConfig) {}
  private get q() {
    return this.store.pool;
  }

  async overview() {
    const [totals, byType, running, recentSessions, series, surfaces, topActions, patterns] = await Promise.all([
      this.q.query(`select
          (select count(*) from sessions)::int as sessions,
          (select count(*) from entities)::int as entities,
          (select count(*) from media)::int as media,
          (select count(*) from schema_patterns)::int as schemas,
          (select count(*) from actions)::int as actions,
          (select count(*) from observations)::int as observations,
          (select count(*) from relationships)::int as relationships,
          (select count(*) from observations where event_type='CONNECTOR_PATTERN_LEARNED')::int as patterns_learned,
          (select count(*) from feed_items)::int as feed_items`),
      this.q.query(`select platform, entity_type, count(*)::int as n from entities group by 1,2 order by 3 desc`),
      this.q.query(`select session_id, platform, account_alias, started_at, health from sessions where ended_at is null and started_at > now() - interval '6 hours' order by started_at desc`),
      this.q.query(`select s.session_id, s.platform, s.goal, s.mode, s.started_at, s.ended_at, s.health,
          (select count(*) from actions a where a.session_id=s.session_id)::int as actions,
          (select count(distinct fingerprint) from entity_observations eo where eo.session_id=s.session_id)::int as entities
        from sessions s order by started_at desc limit 8`),
      this.q.query(`select date_trunc('hour', ts) + (floor(extract(minute from ts)/10)*10) * interval '1 minute' as bucket,
          count(*) filter (where event_type in ('VIDEO_DISCOVERED','POST_DISCOVERED','PROFILE_DISCOVERED','ENTITY_DISCOVERED'))::int as entities,
          count(*) filter (where event_type='NETWORK_RESPONSE_OBSERVED')::int as responses,
          count(*) filter (where event_type in ('ACTION_EXECUTED','ACTION_FAILED'))::int as actions,
          count(*) filter (where event_type='NETWORK_SCHEMA_DISCOVERED')::int as schemas
        from observations where ts > now() - interval '48 hours' group by 1 order by 1`),
      this.q.query(`select coalesce(sum((payload->>'dom_entities')::int),0)::int as dom, coalesce(sum((payload->>'network_entities')::int),0)::int as network,
          coalesce(sum((payload->>'both_surfaces')::int),0)::int as both, coalesce(sum((payload->>'field_agreements')::int),0)::int as agreements,
          coalesce(sum(jsonb_array_length(coalesce(payload->'field_conflicts','[]'::jsonb))),0)::int as conflicts
        from observations where event_type='PAGE_OPENED' and payload ? 'dom_entities'`),
      this.q.query(`select action_type, count(*)::int as n, avg(expected_gain)::float as avg_gain, avg(case when success then 1 else 0 end)::float as success_rate,
          avg((after_state->>'new_entities')::float) as avg_new_entities from actions group by 1 order by 2 desc`),
      this.q.query(`select platform, count(*)::int as n, avg((payload->>'confidence')::float) as conf from observations where event_type='CONNECTOR_PATTERN_LEARNED' group by 1`),
    ]);
    return {
      totals: totals.rows[0],
      by_type: byType.rows,
      running: running.rows,
      recent_sessions: recentSessions.rows,
      series: series.rows,
      surfaces: surfaces.rows[0],
      actions: topActions.rows,
      patterns_by_platform: patterns.rows,
      platforms: this.platforms(),
    };
  }

  platforms() {
    return PLATFORMS.map((p) => {
      const file = path.join(this.cfg.platformModelDir, p, "platform_model.json");
      const manifest = path.join(process.cwd(), "connectors", p, "manifest.json");
      if (!fs.existsSync(file)) return { platform: p, learned: false, adapter: hasAdapter(p), confidence: 0, page_types: 0, entity_types: 0, navigation_actions: 0, network_schemas: 0, media_patterns: 0, sessions: 0, has_manifest: fs.existsSync(manifest), profiles: this.profilesFor(p) };
      const m = new PlatformModel(p, this.cfg.platformModelDir);
      return { learned: true, adapter: hasAdapter(p), has_manifest: fs.existsSync(manifest), profiles: this.profilesFor(p), ...m.summary() };
    });
  }

  profilesFor(p: Platform) {
    if (!fs.existsSync(this.cfg.profilesDir)) return [];
    return fs
      .readdirSync(this.cfg.profilesDir)
      .filter((d) => d.startsWith(`${p}-`))
      .map((d) => ({ alias: d.slice(p.length + 1), has_state: fs.existsSync(path.join(this.cfg.profilesDir, d, "Default", "Cookies")) || fs.existsSync(path.join(this.cfg.profilesDir, d, "Default", "Network", "Cookies")) }));
  }

  platformDetail(p: Platform) {
    const file = path.join(this.cfg.platformModelDir, p, "platform_model.json");
    const manifest = path.join(process.cwd(), "connectors", p, "manifest.json");
    return {
      platform: p,
      model: fs.existsSync(file) ? JSON.parse(fs.readFileSync(file, "utf8")) : null,
      manifest: fs.existsSync(manifest) ? JSON.parse(fs.readFileSync(manifest, "utf8")) : null,
      summary: fs.existsSync(file) ? new PlatformModel(p, this.cfg.platformModelDir).summary() : null,
      profiles: this.profilesFor(p),
    };
  }

  async sessions(limit = 100) {
    return (
      await this.q.query(
        `select s.*, (select count(*) from actions a where a.session_id=s.session_id)::int as actions,
          (select count(distinct fingerprint) from entity_observations eo where eo.session_id=s.session_id)::int as entities,
          (select count(*) from observations o where o.session_id=s.session_id and o.event_type='NETWORK_SCHEMA_DISCOVERED')::int as schemas,
          (select count(*) from observations o where o.session_id=s.session_id and o.event_type='CONNECTOR_PATTERN_LEARNED')::int as patterns,
          (select payload->>'reason' from observations o where o.session_id=s.session_id and o.event_type='BUDGET_EXHAUSTED' limit 1) as ended_because
         from sessions s order by started_at desc limit $1`,
        [limit],
      )
    ).rows;
  }

  async session(id: string) {
    const row = (await this.q.query(`select * from sessions where session_id=$1`, [id])).rows[0];
    if (!row) return null;
    const dir = path.join(this.cfg.sessionsDir, id);
    const read = (f: string) => {
      try {
        return JSON.parse(fs.readFileSync(path.join(dir, f), "utf8"));
      } catch {
        return null;
      }
    };
    const counts = (await this.q.query(`select event_type, count(*)::int as n from observations where session_id=$1 group by 1`, [id])).rows;
    return { ...row, job: read("job.json"), summary: read("summary.json"), event_counts: counts };
  }

  async pages(id: string) {
    return (await this.q.query(`select step, ts, payload from observations where session_id=$1 and event_type='PAGE_OPENED' and payload ? 'page_type' order by ts`, [id])).rows;
  }

  async world(id: string) {
    const dir = path.join(this.cfg.sessionsDir, id, "world_model.json");
    if (fs.existsSync(dir)) return JSON.parse(fs.readFileSync(dir, "utf8"));
    // Reconstruct from the database for sessions crawled on another machine.
    const nodes = (await this.q.query(`select e.fingerprint, e.entity_type as type, e.name, e.url, e.platform_id, min(eo.step) as first_seen_step, max(eo.step) as last_seen_step, count(*)::int as seen_count from entities e join entity_observations eo on eo.fingerprint=e.fingerprint where eo.session_id=$1 group by e.fingerprint limit 1500`, [id])).rows;
    const edges = (await this.q.query(`select from_fp as "from", to_fp as "to", rel_type as type, step from relationships where session_id=$1 limit 4000`, [id])).rows;
    return { nodes, edges };
  }

  async searchEntities(o: { q?: string; type?: string; platform?: string; limit: number; offset: number; sort?: string }) {
    const params: unknown[] = [];
    const where: string[] = [];
    if (o.q) {
      params.push(`%${o.q}%`);
      where.push(`(name ilike $${params.length} or text_excerpt ilike $${params.length} or author ilike $${params.length} or platform_id ilike $${params.length})`);
    }
    if (o.type) {
      params.push(o.type);
      where.push(`entity_type = $${params.length}`);
    }
    if (o.platform) {
      params.push(o.platform);
      where.push(`platform = $${params.length}`);
    }
    const w = where.length ? `where ${where.join(" and ")}` : "";
    const order = o.sort === "views" ? `(metrics->>'views')::numeric desc nulls last` : o.sort === "seen" ? `seen_count desc` : `last_seen desc`;
    params.push(o.limit, o.offset);
    const rows = (await this.q.query(`select *, (select count(*) from entity_observations eo where eo.fingerprint=entities.fingerprint)::int as observations from entities ${w} order by ${order} limit $${params.length - 1} offset $${params.length}`, params)).rows;
    const total = (await this.q.query(`select count(*)::int as n from entities ${w}`, params.slice(0, -2))).rows[0]?.n ?? 0;
    return { rows, total };
  }

  async entity(fp: string) {
    const e = (await this.q.query(`select * from entities where fingerprint=$1`, [fp])).rows[0];
    if (!e) return null;
    const observations = (await this.q.query(`select eo.session_id, eo.step, eo.ts, eo.surfaces, eo.snapshot, s.goal from entity_observations eo left join sessions s on s.session_id=eo.session_id where eo.fingerprint=$1 order by eo.ts desc limit 100`, [fp])).rows;
    const relations = (await this.q.query(`select r.rel_type, r.from_fp, r.to_fp, r.step, coalesce(a.name, a.platform_id) as from_name, coalesce(b.name, b.platform_id) as to_name, a.entity_type as from_type, b.entity_type as to_type from relationships r left join entities a on a.fingerprint=r.from_fp left join entities b on b.fingerprint=r.to_fp where r.from_fp=$1 or r.to_fp=$1 limit 200`, [fp])).rows;
    const media = (await this.q.query(`select * from media where fingerprint=$1`, [fp])).rows[0] ?? null;
    const feed = (await this.q.query(`select session_id, step, page_type, feed_position, visible, ts from feed_items where fingerprint=$1 order by ts desc limit 50`, [fp])).rows;
    return { entity: e, observations, relations, media, feed };
  }

  async schemas(platform?: string) {
    const params: unknown[] = [];
    let w = "";
    if (platform) {
      params.push(platform);
      w = `where platform=$1`;
    }
    return (await this.q.query(`select * from schema_patterns ${w} order by observed_count desc limit 300`, params)).rows;
  }

  async liveEvents(since: string, session?: string, limit = 200) {
    const params: unknown[] = [since];
    let w = `ts > $1`;
    if (session) {
      params.push(session);
      w += ` and session_id=$2`;
    }
    params.push(limit);
    return (await this.q.query(`select event_id, session_id, platform, event_type, step, ts, payload, provenance from observations where ${w} order by ts asc limit $${params.length}`, params)).rows;
  }

  async jobs(limit = 50) {
    return (await this.q.query(`select j.*, s.health, (select count(*) from actions a where a.session_id=j.session_id)::int as actions, (select count(distinct fingerprint) from entity_observations eo where eo.session_id=j.session_id)::int as entities from crawl_jobs j left join sessions s on s.session_id=j.session_id order by created_at desc limit $1`, [limit])).rows;
  }
}