spb/social-runtime-crawler

1import fs from "node:fs";2import path from "node:path";3import type { PostgresStore } from "@src/storage";4import { PLATFORMS, type AppConfig, type Platform } from "@src/shared";5import { PlatformModel } from "@src/platform-model";6import { hasAdapter } from "@src/connectors";78/** Read-side queries for the research console. All read PostgreSQL; files (models, manifests, session dirs) complete them. */9export class Queries {10  constructor(private readonly store: PostgresStore, private readonly cfg: AppConfig) {}11  private get q() {12    return this.store.pool;13  }1415  async overview() {16    const [totals, byType, running, recentSessions, series, surfaces, topActions, patterns] = await Promise.all([17      this.q.query(`select18          (select count(*) from sessions)::int as sessions,19          (select count(*) from entities)::int as entities,20          (select count(*) from media)::int as media,21          (select count(*) from schema_patterns)::int as schemas,22          (select count(*) from actions)::int as actions,23          (select count(*) from observations)::int as observations,24          (select count(*) from relationships)::int as relationships,25          (select count(*) from observations where event_type='CONNECTOR_PATTERN_LEARNED')::int as patterns_learned,26          (select count(*) from feed_items)::int as feed_items`),27      this.q.query(`select platform, entity_type, count(*)::int as n from entities group by 1,2 order by 3 desc`),28      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`),29      this.q.query(`select s.session_id, s.platform, s.goal, s.mode, s.started_at, s.ended_at, s.health,30          (select count(*) from actions a where a.session_id=s.session_id)::int as actions,31          (select count(distinct fingerprint) from entity_observations eo where eo.session_id=s.session_id)::int as entities32        from sessions s order by started_at desc limit 8`),33      this.q.query(`select date_trunc('hour', ts) + (floor(extract(minute from ts)/10)*10) * interval '1 minute' as bucket,34          count(*) filter (where event_type in ('VIDEO_DISCOVERED','POST_DISCOVERED','PROFILE_DISCOVERED','ENTITY_DISCOVERED'))::int as entities,35          count(*) filter (where event_type='NETWORK_RESPONSE_OBSERVED')::int as responses,36          count(*) filter (where event_type in ('ACTION_EXECUTED','ACTION_FAILED'))::int as actions,37          count(*) filter (where event_type='NETWORK_SCHEMA_DISCOVERED')::int as schemas38        from observations where ts > now() - interval '48 hours' group by 1 order by 1`),39      this.q.query(`select coalesce(sum((payload->>'dom_entities')::int),0)::int as dom, coalesce(sum((payload->>'network_entities')::int),0)::int as network,40          coalesce(sum((payload->>'both_surfaces')::int),0)::int as both, coalesce(sum((payload->>'field_agreements')::int),0)::int as agreements,41          coalesce(sum(jsonb_array_length(coalesce(payload->'field_conflicts','[]'::jsonb))),0)::int as conflicts42        from observations where event_type='PAGE_OPENED' and payload ? 'dom_entities'`),43      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,44          avg((after_state->>'new_entities')::float) as avg_new_entities from actions group by 1 order by 2 desc`),45      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`),46    ]);47    return {48      totals: totals.rows[0],49      by_type: byType.rows,50      running: running.rows,51      recent_sessions: recentSessions.rows,52      series: series.rows,53      surfaces: surfaces.rows[0],54      actions: topActions.rows,55      patterns_by_platform: patterns.rows,56      platforms: this.platforms(),57    };58  }5960  platforms() {61    return PLATFORMS.map((p) => {62      const file = path.join(this.cfg.platformModelDir, p, "platform_model.json");63      const manifest = path.join(process.cwd(), "connectors", p, "manifest.json");64      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) };65      const m = new PlatformModel(p, this.cfg.platformModelDir);66      return { learned: true, adapter: hasAdapter(p), has_manifest: fs.existsSync(manifest), profiles: this.profilesFor(p), ...m.summary() };67    });68  }6970  profilesFor(p: Platform) {71    if (!fs.existsSync(this.cfg.profilesDir)) return [];72    return fs73      .readdirSync(this.cfg.profilesDir)74      .filter((d) => d.startsWith(`${p}-`))75      .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")) }));76  }7778  platformDetail(p: Platform) {79    const file = path.join(this.cfg.platformModelDir, p, "platform_model.json");80    const manifest = path.join(process.cwd(), "connectors", p, "manifest.json");81    return {82      platform: p,83      model: fs.existsSync(file) ? JSON.parse(fs.readFileSync(file, "utf8")) : null,84      manifest: fs.existsSync(manifest) ? JSON.parse(fs.readFileSync(manifest, "utf8")) : null,85      summary: fs.existsSync(file) ? new PlatformModel(p, this.cfg.platformModelDir).summary() : null,86      profiles: this.profilesFor(p),87    };88  }8990  async sessions(limit = 100) {91    return (92      await this.q.query(93        `select s.*, (select count(*) from actions a where a.session_id=s.session_id)::int as actions,94          (select count(distinct fingerprint) from entity_observations eo where eo.session_id=s.session_id)::int as entities,95          (select count(*) from observations o where o.session_id=s.session_id and o.event_type='NETWORK_SCHEMA_DISCOVERED')::int as schemas,96          (select count(*) from observations o where o.session_id=s.session_id and o.event_type='CONNECTOR_PATTERN_LEARNED')::int as patterns,97          (select payload->>'reason' from observations o where o.session_id=s.session_id and o.event_type='BUDGET_EXHAUSTED' limit 1) as ended_because98         from sessions s order by started_at desc limit $1`,99        [limit],100      )101    ).rows;102  }103104  async session(id: string) {105    const row = (await this.q.query(`select * from sessions where session_id=$1`, [id])).rows[0];106    if (!row) return null;107    const dir = path.join(this.cfg.sessionsDir, id);108    const read = (f: string) => {109      try {110        return JSON.parse(fs.readFileSync(path.join(dir, f), "utf8"));111      } catch {112        return null;113      }114    };115    const counts = (await this.q.query(`select event_type, count(*)::int as n from observations where session_id=$1 group by 1`, [id])).rows;116    return { ...row, job: read("job.json"), summary: read("summary.json"), event_counts: counts };117  }118119  async pages(id: string) {120    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;121  }122123  async world(id: string) {124    const dir = path.join(this.cfg.sessionsDir, id, "world_model.json");125    if (fs.existsSync(dir)) return JSON.parse(fs.readFileSync(dir, "utf8"));126    // Reconstruct from the database for sessions crawled on another machine.127    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;128    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;129    return { nodes, edges };130  }131132  async searchEntities(o: { q?: string; type?: string; platform?: string; limit: number; offset: number; sort?: string }) {133    const params: unknown[] = [];134    const where: string[] = [];135    if (o.q) {136      params.push(`%${o.q}%`);137      where.push(`(name ilike $${params.length} or text_excerpt ilike $${params.length} or author ilike $${params.length} or platform_id ilike $${params.length})`);138    }139    if (o.type) {140      params.push(o.type);141      where.push(`entity_type = $${params.length}`);142    }143    if (o.platform) {144      params.push(o.platform);145      where.push(`platform = $${params.length}`);146    }147    const w = where.length ? `where ${where.join(" and ")}` : "";148    const order = o.sort === "views" ? `(metrics->>'views')::numeric desc nulls last` : o.sort === "seen" ? `seen_count desc` : `last_seen desc`;149    params.push(o.limit, o.offset);150    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;151    const total = (await this.q.query(`select count(*)::int as n from entities ${w}`, params.slice(0, -2))).rows[0]?.n ?? 0;152    return { rows, total };153  }154155  async entity(fp: string) {156    const e = (await this.q.query(`select * from entities where fingerprint=$1`, [fp])).rows[0];157    if (!e) return null;158    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;159    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;160    const media = (await this.q.query(`select * from media where fingerprint=$1`, [fp])).rows[0] ?? null;161    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;162    return { entity: e, observations, relations, media, feed };163  }164165  async schemas(platform?: string) {166    const params: unknown[] = [];167    let w = "";168    if (platform) {169      params.push(platform);170      w = `where platform=$1`;171    }172    return (await this.q.query(`select * from schema_patterns ${w} order by observed_count desc limit 300`, params)).rows;173  }174175  async liveEvents(since: string, session?: string, limit = 200) {176    const params: unknown[] = [since];177    let w = `ts > $1`;178    if (session) {179      params.push(session);180      w += ` and session_id=$2`;181    }182    params.push(limit);183    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;184  }185186  async jobs(limit = 50) {187    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;188  }189}190