// Reading the index in the browser. Every file comes from one commit of the dataset, so every number agrees.
import { asyncBufferFromUrl, parquetMetadataAsync, parquetReadObjects } from 'https://cdn.jsdelivr.net/npm/hyparquet@1.31.5/+esm';
import { compressors } from 'https://cdn.jsdelivr.net/npm/hyparquet-compressors@1.1.2/+esm';

export const REPO = 'Battam/agentic-env-index';
export const HF = 'https://huggingface.co';
export const DATASET_URL = `${HF}/datasets/${REPO}`;

export const fileUrl = (sha, path) => `${DATASET_URL}/resolve/${sha}/${path}`;
export const num = (v) => (v == null ? null : typeof v === 'bigint' ? Number(v) : Number(v));

async function getJson(url) {
  const r = await fetch(url);
  if (!r.ok) throw new Error(`${url}: HTTP ${r.status}`);
  return r.json();
}

async function getText(url) {
  const r = await fetch(url);
  if (!r.ok) throw new Error(`${url}: HTTP ${r.status}`);
  return r.text();
}

/** A small file read whole (the meta tables): one request instead of a size probe and ranges. */
export async function readParquet(url, options = {}) {
  const r = await fetch(url);
  if (!r.ok) throw new Error(`${url}: HTTP ${r.status}`);
  const file = await r.arrayBuffer();
  return parquetReadObjects({ file, compressors, utf8: false, ...options });
}

export async function parquetRows(url) {
  const file = await asyncBufferFromUrl({ url });
  return num((await parquetMetadataAsync(file)).num_rows);
}

/** The columns of cols that the file has, read with hyparquet (byte ranges); range: {rowStart, rowEnd}. */
export async function readColumns(url, cols, range = {}) {
  const file = await asyncBufferFromUrl({ url });
  const metadata = await parquetMetadataAsync(file);
  const names = new Set(metadata.schema.slice(1).map((s) => s.name));
  const columns = cols.filter((c) => names.has(c));
  const rows = await parquetReadObjects({ file, metadata, columns, compressors, utf8: false, ...range });
  return { rows, total: num(metadata.num_rows) };
}

/** The commit to read: ?rev=<sha or branch>, else main. Returns {sha, date, files}. */
export async function loadRevision(rev) {
  const j = await getJson(`${HF}/api/datasets/${REPO}/revision/${encodeURIComponent(rev || 'main')}`);
  if (!/^[0-9a-f]{40}$/.test(j.sha || '')) throw new Error('the Hub gave no commit id');
  return { sha: j.sha, date: j.lastModified || null, files: (j.siblings || []).map((s) => s.rfilename) };
}

/** Batch folders of one source with the date of their last commit (Hub tree API). */
export async function loadBatchDates(sha, source) {
  const j = await getJson(`${HF}/api/datasets/${REPO}/tree/${sha}/data/${encodeURIComponent(source)}?expand=true`);
  const out = new Map();
  for (const e of j) {
    if (e.type !== 'directory') continue;
    out.set(e.path.split('/').pop(), e.lastCommit?.date || null);
  }
  return out;
}

// ---- the classification contract (#252): domain is a list of codes (D1 to D10, with leaves), or multi, or
// unclassified; form is one id; names come from meta/vocabulary.json. Without them everything shows as unclassified.
const pick = (o, keys) => {
  if (!o || typeof o !== 'object') return undefined;
  for (const k of keys) if (o[k] != null) return o[k];
  return undefined;
};
const NAME_KEYS = ['name', 'label', 'title', 'en'];
const CODE_KEYS = ['code', 'id', 'key', 'value'];
const KIDS_KEYS = ['children', 'leaves', 'subdomains', 'items'];
const ABOUT_KEYS = ['description', 'about', 'definition'];

function vocabSection(node) {
  const names = new Map();
  const about = new Map();
  const order = [];
  const add = (code, name, text) => {
    if (code == null) return;
    code = String(code);
    if (!names.has(code)) order.push(code);
    if (typeof name === 'string' && name.trim()) names.set(code, name.trim());
    else if (!names.has(code)) names.set(code, null);
    if (typeof text === 'string' && text.trim()) about.set(code, text.trim());
  };
  const walk = (n) => {
    if (!n) return;
    if (Array.isArray(n)) {
      for (const it of n) {
        if (typeof it === 'string') add(it, null);
        else if (it && typeof it === 'object') {
          add(pick(it, CODE_KEYS), pick(it, NAME_KEYS), pick(it, ABOUT_KEYS));
          walk(pick(it, KIDS_KEYS));
        }
      }
    } else if (typeof n === 'object') {
      for (const [k, v] of Object.entries(n)) {
        if (typeof v === 'string') add(k, v);
        else if (v && typeof v === 'object') {
          add(k, pick(v, NAME_KEYS), pick(v, ABOUT_KEYS));
          walk(pick(v, KIDS_KEYS));
        } else add(k, null);
      }
    }
  };
  walk(node);
  return { names, about, order };
}

// meta/vocabulary.json (#304): domains [{code, name, about, parent}], forms [{id, about}]. A domain shows its name; a
// form has no name, so it shows its id written as words (test-suites -> test suites).
export function parseVocabulary(j) {
  const form = vocabSection(pick(j, ['form', 'forms']));
  for (const [id, name] of form.names) if (!name) form.names.set(id, id.replace(/[-_]+/g, ' '));
  return { domain: vocabSection(pick(j, ['domain', 'domains'])), form };
}

export const topOf = (code) => String(code).split('.')[0];

function domainOf(v) {
  let list = v;
  if (typeof list === 'string') list = [list];
  if (list && typeof list.toArray === 'function') list = list.toArray();
  list = Array.isArray(list) ? list.map(String).map((s) => s.trim()).filter(Boolean) : [];
  if (!list.length || list.includes('unclassified')) return { kind: 'none', codes: [] };
  if (list.includes('multi')) return { kind: 'multi', codes: [] };
  return { kind: 'codes', codes: [...new Set(list)] };
}

// ---- small parsers of the index's own text files
const intOf = (s) => Number(String(s).replace(/,/g, ''));

/** CHANGELOG.md -> [{date, title, kind, sourcesAdded, sourcesRemoved, rowsAdded, rowsRemoved, sources, rows}] */
export function parseChangelog(md) {
  const out = [];
  const parts = md.split(/^## /m).slice(1);
  for (const part of parts) {
    const [head, ...lines] = part.split('\n');
    const m = head.match(/^(\d{4}-\d{2}-\d{2}):\s*(.+)$/);
    if (!m) continue;
    const e = { date: m[1], title: m[2].trim(), kind: 'other', sourcesAdded: null, sourcesRemoved: null, rowsAdded: 0, rowsRemoved: 0 };
    const body = lines.join('\n');
    const start = body.match(/lists ([\d,]+) sources with ([\d,]+) rows/);
    if (start) {
      e.kind = 'start';
      e.sources = intOf(start[1]);
      e.rows = intOf(start[2]);
    }
    const sa = body.match(/^- Sources added \((\d+)\)/m);
    const sr = body.match(/^- Sources removed \((\d+)\)/m);
    if (sa || sr) {
      e.kind = 'update';
      e.sourcesAdded = sa ? Number(sa[1]) : 0;
      e.sourcesRemoved = sr ? Number(sr[1]) : 0;
    }
    for (const r of body.matchAll(/^\s+- \S+: .*?\(([\d,]+) to ([\d,]+)\)\s*$/gm)) {
      const d = intOf(r[2]) - intOf(r[1]);
      if (d > 0) e.rowsAdded += d;
      else e.rowsRemoved -= d;
    }
    out.push(e);
  }
  return out;
}

/** README.md, section "Not included, and why": category -> requirement sentence. */
export function parseRequirements(md) {
  const out = new Map();
  const sec = md.split(/^## /m).find((s) => s.startsWith('Not included'));
  if (!sec) return out;
  for (const m of sec.matchAll(/^\| `([a-z0-9_]+)` \| (.+?) \| \d+ \| \d+ \|\s*$/gm)) out.set(m[1], m[2].trim());
  return out;
}

const LINK_RE = /^(\d{4}-\d{2}-\d{2}): ([\d,]+) of ([\d,]+) reachable/;
function linkOf(text) {
  if (!text) return { kind: 'unchecked' };
  const m = String(text).match(LINK_RE);
  if (m) {
    const ok = intOf(m[2]);
    const total = intOf(m[3]);
    return { kind: ok >= total ? 'ok' : 'partial', ok, total, date: m[1] };
  }
  if (/^needs /.test(text)) return { kind: 'login', text: String(text) };
  return { kind: 'other', text: String(text) };
}

/** What every row of a source allows, from sources.json license_rows: 'yes' (every row), 'cond' (yes or
 * conditional for every row) or 'no' (some rows are no or unknown). */
function permission(lr) {
  if (!lr) return 'no';
  const keys = Object.entries(lr).filter(([, v]) => num(v?.rows) > 0).map(([k]) => k);
  if (!keys.length) return 'no';
  if (keys.every((k) => k === 'yes')) return 'yes';
  if (keys.every((k) => k === 'yes' || k === 'conditional')) return 'cond';
  return 'no';
}

async function readCatalog(url) {
  if (/\.json(\?|$)/i.test(url)) {
    const j = await getJson(url);
    return Array.isArray(j) ? j : j.sources || j.rows || [];
  }
  return readParquet(url);
}

/** Everything the page shows, from one commit. params: URLSearchParams (rev, catalog, vocabulary). */
export async function loadIndex(params, onNote) {
  const rev = await loadRevision(params.get('rev'));
  const at = (p) => fileUrl(rev.sha, p);
  const catalogUrl = params.get('catalog') || at('meta/catalog.parquet');
  const has = (p) => rev.files.includes(p);
  const vocabUrl = params.get('vocabulary') || (has('meta/vocabulary.json') ? at('meta/vocabulary.json') : null);

  const [catalog, sourcesJson, coverage, batches, notIncluded, changelog, readme, vocabJson] = await Promise.all([
    readCatalog(catalogUrl),
    getJson(at('meta/sources.json')),
    getJson(at('meta/coverage.json')),
    readParquet(at('meta/batches.parquet')).catch(() => []),
    readParquet(at('meta/not_included.parquet')).catch(() => []),
    getText(at('CHANGELOG.md')).catch(() => ''),
    getText(at('README.md')).catch(() => ''),
    vocabUrl ? getJson(vocabUrl).catch((e) => { onNote?.(`The vocabulary could not be read (${e.message}); codes are shown as they are.`); return null; }) : null,
  ]);

  const vocab = parseVocabulary(vocabJson || {});
  const srcInfo = new Map((sourcesJson.sources || []).map((e) => [e.source, e]));
  const cov = coverage.by_source || {};
  const batchesBy = new Map();
  for (const b of batches) {
    if (!batchesBy.has(b.source)) batchesBy.set(b.source, []);
    batchesBy.get(b.source).push(b);
  }
  const filesBy = new Map();
  for (const f of rev.files) {
    const m = f.match(/^data\/([^/]+)\/([^/]+)\/[^/]+\.parquet$/);
    if (!m) continue;
    if (!filesBy.has(m[1])) filesBy.set(m[1], []);
    filesBy.get(m[1]).push({ path: f, batch: m[2] });
  }

  const sources = catalog.map((r) => {
    const info = srcInfo.get(r.source) || {};
    const lic = r.license || r.licence || {};
    const dom = domainOf(r.domain);
    const rows = num(r.files ?? info.rows) || 0;
    const bytes = num(r.bytes);
    const form = r.form == null || r.form === '' || r.form === 'unclassified' ? null : String(r.form);
    // A unit is one unit group (a test case, one workbook); files are counted apart. Once the catalog carries the
    // columns units (distinct units; a raw corpus counts rows), answers (units that have an answer) and files (files
    // of every role), they are read as they are. Until then: rows of a collection carry roles (sources.json unit: unit,
    // companion, answer) and such a source is task-ready; its units are its answer files when it has any (one answer
    // per unit group), else its unit rows. Every other source is a raw corpus: each row is one unit to mine.
    const roles = info.unit || {};
    const has = (k) => r[k] != null && r[k] !== '' && Number.isFinite(num(r[k]));
    const answers = has('answers') ? num(r.answers) : num(roles.answer) || 0;
    const isTask = num(roles.unit) > 0 || answers > 0 || (has('units') && num(r.units) < rows);
    const units = has('units') ? num(r.units) : isTask ? (answers || num(roles.unit)) : rows;
    const lr = info.license_rows?.redistribution_ok || {};
    const shipRows = (num(lr.yes?.rows) || 0) + (num(lr.conditional?.rows) || 0);
    const shipShare = rows ? Math.min(1, shipRows / rows) : 0;
    return {
      isTask,
      readyUnits: isTask ? units : 0,
      rawRows: isTask ? 0 : rows,
      fromCatalog: has('units'),
      roles: { unit: num(roles.unit) || 0, companion: num(roles.companion) || 0, answer: answers },
      kinds: Object.keys(roles).filter((k) => !['unit', 'companion', 'answer'].includes(k)),
      units,
      answered: answers,
      shipShare,
      id: r.source,
      title: r.title || r.source,
      description: r.description || info.description || '',
      summary: r.summary || '',
      unit: r.unit || '',
      status: r.status || null,
      reasons: Array.isArray(r.status_reasons) ? r.status_reasons : r.status_reasons?.toArray?.() || [],
      domainKind: dom.kind,
      codes: dom.codes,
      tops: [...new Set(dom.codes.map(topOf))],
      form,
      tier: lic.tier || 'unknown',
      ship: lic.ship || null,
      spdx: lic.spdx || null,
      licenceNote: lic.note || null,
      contamination: r.contamination || {},
      pii: r.pii || {},
      rows,
      bytes: bytes && bytes > 0 ? bytes : null,
      // distinct content (#364): bytes minus the sizes of the rows meta/duplicates.parquet lists; null when not recorded
      distinct: has('distinct_bytes') ? num(r.distinct_bytes) : null,
      rowsNoSize: num(r.files_without_size) || 0,
      topTypes: Array.isArray(r.top_types) ? r.top_types : r.top_types?.toArray?.() || [],
      needsAuth: r.needs_auth || null,
      link: linkOf(r.link_check),
      linkText: r.link_check || null,
      train: permission(info.license_rows?.training_ok),
      redist: permission(info.license_rows?.redistribution_ok),
      info,
      cov: cov[r.source] || null,
      batches: batchesBy.get(r.source) || [],
      files: (filesBy.get(r.source) || []).sort((a, b) => a.path.localeCompare(b.path)),
    };
  });

  return {
    sha: rev.sha,
    date: rev.date,
    overrides: { catalog: params.get('catalog'), vocabulary: params.get('vocabulary') },
    hasVocabulary: Boolean(vocabJson),
    vocab,
    sources,
    coverage,
    notIncluded,
    requirements: parseRequirements(readme),
    changelog: parseChangelog(changelog),
  };
}