import feeds from './shared/regional-energy-feeds.json' with { type: 'json' };
import { readFileSync } from 'node:fs';
import { rootCertificates } from 'node:tls';
import { Agent, fetch as undiciFetch } from 'undici';
import { getOptionalUpstashCreds, upstashCommand } from './_upstash-rest.mjs';
import { assertZipExpansionWithin } from './_zip-expansion.mjs';
import {
  parsePgcb, parseNiso, parseCammesa, parseEskom, parseOns, parseAnpTrade,
  parseAnpPriceRows, parsePetrobangla, parseGse, pdfItems, pdfAllItems,
  discoverEskomCsv, discoverGsePdf, discoverPetrobanglaPdf, buildFeedSnapshot, inEffect, effectiveAt,
  discoverEvnPost, parseEvn,
  KOREM_FIXED_URLS, koremEnergySeriesUrl, koremLastDateUrl, koremCtUrls, parseKorem,
  COES_DEMAND_URL, coesDemandBody, parseCoes,
  ICCC_INDEX_URL, discoverPngFuelPosts, parsePngFuelPost,
  FCCC_INDEX_URL, discoverFijiFuelPdfs, parseFijiFuel,
  MEM_MAX_UPLOADS_PER_MONTH, memUploadFolders, memWorkbookUrl, parseMemPriceRows,
} from './lib/regional-energy-feeds.mjs';
import {
  QATARENERGY_PDF_URL, parseQatarFuel,
  ERB_INDEX_URL, discoverErbBuildUps, parseErbBuildUp,
  npaMediaUrl, discoverNpaWorkbook, parseNpaExPumpRows, parseNpaIndicatorRows,
  SONABHY_PAGE_URL, SONABHY_API_HOSTS, discoverSonabhyState, discoverSonabhyApi, parseSonabhyTariffs,
  ZERA_HOME_URL, parseZeraPrices,
  DOE_HOST, DOE_MEDIA_HOST, DOE_NCR_PAGE_URL, discoverDoeNcrPdfs, parseDoeNcrPrices,
  ANPTL_INDEX_URL, discoverAnpTlPosts, discoverAnpTlPdf, parseAnpTlStations,
  CREG_PAGE_URL, parseCregPrices,
  URSEA_TOPIC_URL, discoverUrseaPages, parseUrseaMaximumPrices, parseUrseaReferencePrices,
} from './lib/regional-fuel-feeds.mjs';

const USER_AGENT = 'WorldMonitor/1.0 (+https://www.worldmonitor.app)';
// The ANP national price sheet uses about a dozen columns, MEM's ten.
const SHEET_COLUMNS = 64;
const DOWNLOAD_HOSTS = new Set([
  'erp.powergrid.gov.bd', 'petrobangla.org.bd', 'www.niggrid.org',
  'www.eskom.co.za', 'api.cammesa.com', 'admin.gse.com.ge',
  'dados.ons.org.br', 'ons-aws-prod-opendata.s3.amazonaws.com',
  'www.gov.br', 'objectstorage.ap-dcc-gazipur-1.oraclecloud15.com',
  'evn.com.vn', 'portal.korem.kz', 'www.coes.org.pe',
  'iccc.gov.pg', 'fccc.gov.fj', 'mem.gob.gt',
  'www.qatarenergy.qa', 'www.erb.org.zm', 'npa.gov.gh', 'www.sonabhy.bf', ...SONABHY_API_HOSTS,
  'www.zera.co.zw', DOE_HOST, DOE_MEDIA_HOST, 'www.anp.tl', 'creg.gov.co', 'www.gub.uy',
]);
export const regionalEnergyKey = id => `energy:regional:${id}:v1`;
export const regionalEnergyMetaKey = id => `seed-meta:energy:regional-${id}`;
const issuerAgents = new Map();
// @railway-runtime-dependency ./data/regional-energy/pgcb-issuer.pem
// @railway-runtime-dependency ./data/regional-energy/petrobangla-issuer.pem

function sourceFetch(url, options) {
  const issuer = {
    'erp.powergrid.gov.bd': 'pgcb',
    'petrobangla.org.bd': 'petrobangla',
  }[new URL(url).hostname];
  if (!issuer) return globalThis.fetch(url, options);
  if (!issuerAgents.has(issuer)) {
    // These servers omit their intermediates. Complete their chains using
    // the public Sectigo issuers, while retaining Node's roots and TLS checks.
    const intermediate = readFileSync(new URL(`./data/regional-energy/${issuer}-issuer.pem`, import.meta.url), 'utf8');
    issuerAgents.set(issuer, new Agent({ connect: { ca: [...rootCertificates, intermediate] } }));
  }
  return undiciFetch(url, { ...options, dispatcher: issuerAgents.get(issuer) });
}

// ZIP headers are not trusted: inflate each entry with a bounded output
// before ExcelJS inflates the same bytes without one.
function assertWorkbookWithinBudget(bytes, source) {
  try {
    assertZipExpansionWithin(bytes, 50_000_000);
  } catch (err) {
    throw new Error(/expands past/.test(err.message) ? `${source} workbook exceeds size budget` : `${source} workbook is ${err.message}`);
  }
}
const cellValue = cell => (cell.value && typeof cell.value === 'object' && !(cell.value instanceof Date)
  ? ('result' in cell.value ? cell.value.result : 'richText' in cell.value ? cell.value.richText.map(r => r.text).join('') : cell.value)
  : cell.value);

// Loads one worksheet as arrays of cell values within the first SHEET_COLUMNS.
async function readSheetRows(bytes, sheetName, source) {
  assertWorkbookWithinBudget(bytes, source);
  const { default: ExcelJS } = await import('exceljs');
  const book = new ExcelJS.Workbook();
  await book.xlsx.load(bytes);
  const sheet = book.getWorksheet(sheetName);
  if (!sheet) throw new Error(`Missing ${source} price sheet`);
  // ExcelJS indexes row values by column, so copying row.values would
  // allocate up to a far-right cell. Copy only cells within the price table.
  const sheetRows = [];
  sheet.eachRow(row => {
    const values = [];
    row.eachCell((cell, column) => { if (column <= SHEET_COLUMNS) values[column - 1] = cellValue(cell); });
    sheetRows.push(Array.from(values));
  });
  return sheetRows;
}

// Streams only the first worksheet. The NPA ex-pump workbook keeps one sheet
// per publication day (hundreds of sheets, ~3 MB); loading it whole costs
// about 600 MB, streaming the newest sheet about 160 MB.
async function readFirstSheetRows(bytes, source) {
  assertWorkbookWithinBudget(bytes, source);
  const { default: ExcelJS } = await import('exceljs');
  const { Readable } = await import('node:stream');
  const reader = new ExcelJS.stream.xlsx.WorkbookReader(Readable.from([bytes]), { worksheets: 'emit', sharedStrings: 'cache', hyperlinks: 'ignore', styles: 'ignore', entries: 'emit' });
  for await (const sheet of reader) {
    const sheetRows = [];
    for await (const row of sheet) {
      const values = [];
      row.eachCell((cell, column) => { if (column <= SHEET_COLUMNS) values[column - 1] = cellValue(cell); });
      sheetRows.push(Array.from(values));
    }
    return sheetRows;
  }
  throw new Error(`Missing ${source} worksheet`);
}

// `pdfReader` is an injection seam for tests ONLY: it receives the downloaded
// bytes and their URL and returns text items in place of pdfjs.
export async function fetchRegionalEnergyFeed(id, { fetchFn = sourceFetch, nowMs = Date.now(), pdfReader = null } = {}) {
  const feed = feeds.find(f => f.id === id);
  if (!feed) throw new Error(`Unknown regional energy feed ${id}`);
  const downloaded = [];
  const approved = url => {
    if (url.protocol !== 'https:' || url.username || url.password || url.port || !DOWNLOAD_HOSTS.has(url.hostname)) throw new Error('Unapproved regional energy URL');
  };
  // Probes a file a provider names predictably. Only 404 means absent: a
  // challenge, redirect or outage fails the run rather than skipping a file.
  const exists = async input => {
    const url = new URL(input);
    approved(url);
    const response = await fetchFn(url.href, { method: 'HEAD', headers: { 'User-Agent': USER_AGENT }, signal: AbortSignal.timeout(25000), redirect: 'manual' });
    if (response.status === 404) return false;
    if (!response.ok) throw new Error(`${id} provider HTTP ${response.status}`);
    return true;
  };
  // `form` sends a urlencoded POST body; a redirect would drop it, so a POST must not redirect.
  // `timeoutMs` covers the whole response, body included.
  const read = async (input, format = 'text', form = undefined, timeoutMs = 25000) => {
    let url = new URL(input);
    for (let hop = 0; hop < 4; hop++) {
      approved(url);
      const request = form === undefined
        ? { headers: { 'User-Agent': USER_AGENT } }
        : { method: 'POST', body: form, headers: { 'User-Agent': USER_AGENT, 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8' } };
      const response = await fetchFn(url.href, { ...request, signal: AbortSignal.timeout(timeoutMs), redirect: 'manual' });
      if ([301, 302, 303, 307, 308].includes(response.status)) {
        if (form !== undefined) throw new Error('Unexpected regional energy POST redirect');
        const location = response.headers.get('location');
        if (!location) throw new Error('Missing regional energy redirect location');
        url = new URL(location, url); continue;
      }
      if (!response.ok) throw new Error(`${id} provider HTTP ${response.status}`);
      const limit = 25_000_000;
      if (Number(response.headers.get('content-length')) > limit) throw new Error('Energy download exceeds size budget');
      const parts = [];
      let size = 0;
      for await (const chunk of response.body) {
        size += chunk.byteLength;
        if (size > limit) throw new Error('Energy download exceeds size budget');
        parts.push(chunk);
      }
      const bytes = Buffer.concat(parts);
      downloaded.push(url.href);
      if (format === 'bytes') return bytes;
      const body = bytes.toString('utf8');
      return format === 'json' ? JSON.parse(body) : body;
    }
    throw new Error('Too many regional energy redirects');
  };
  const readPdf = async (input, pages = 'first') => {
    const bytes = await read(input, 'bytes');
    return pdfReader ? pdfReader(bytes, new URL(input).href) : (pages === 'all' ? pdfAllItems : pdfItems)(bytes);
  };

  let points;
  if (id === 'pgcb') points = parsePgcb(await read(feed.sourceUrl));
  if (id === 'niso') points = parseNiso(await read(feed.sourceUrl));
  if (id === 'petrobangla') {
    const url = discoverPetrobanglaPdf(await read(`${feed.sourceUrl}?filters=${encodeURIComponent(JSON.stringify({ reports_type: '6922d2b181fc96cef9e99f16' }))}`));
    points = parsePetrobangla(await pdfItems(await read(url, 'bytes')));
  }
  if (id === 'cammesa') {
    const base = 'https://api.cammesa.com/demanda-svc';
    const demand = await read(`${base}/demanda/ObtieneDemandaYTemperaturaRegion?id_region=1002`, 'json');
    const generation = await read(`${base}/generacion/ObtieneGeneracioEnergiaPorRegion?id_region=1002`, 'json');
    points = parseCammesa(demand, generation);
  }
  if (id === 'eskom') {
    const datasets = ['System_hourly_actual_and_forecasted_demand', 'Weekly_unplanned_outages', 'Hourly_Generation'];
    const csvs = [];
    for (const title of datasets) {
      const media = await read(`https://www.eskom.co.za/dataportal/wp-json/wp/v2/media?search=${encodeURIComponent(title)}&per_page=20&orderby=date&order=desc`, 'json');
      csvs.push(await read(discoverEskomCsv(media, title)));
    }
    points = parseEskom(...csvs);
  }
  if (id === 'gse') {
    const url = discoverGsePdf(await read(feed.sourceUrl, 'json'));
    points = parseGse(await pdfItems(await read(url, 'bytes')));
  }
  if (id === 'ons') {
    const catalog = await read(feed.sourceUrl, 'json');
    if (catalog.success !== true) throw new Error('ONS catalog unavailable');
    const resources = (catalog.result?.resources ?? []).filter(r => r.format === 'CSV' && /BALANCO_ENERGIA_SUBSISTEMA_20\d{2}\.csv$/.test(r.url)).sort((a, b) => b.url.localeCompare(a.url));
    if (!resources.length) throw new Error('Missing ONS balance CSV');
    points = parseOns(await read(resources[0].url));
  }
  if (id === 'anp') {
    const tradePage = await read('https://www.gov.br/anp/pt-br/centrais-de-conteudo/dados-abertos/importacoes-e-exportacoes');
    const tradeUrl = [...tradePage.matchAll(/href="([^"]+)"/g)].find(m => /\/ie\/derivados\/importacoes-exportacoes-derivados-[^/]+\.csv$/.test(m[1]))?.[1];
    if (!tradeUrl) throw new Error('Missing ANP petroleum trade CSV');
    const trade = parseAnpTrade(await read(tradeUrl));
    const pricePage = await read('https://www.gov.br/anp/pt-br/assuntos/precos-e-defesa-da-concorrencia/precos/levantamento-de-precos-de-combustiveis-ultimas-semanas-pesquisadas');
    const priceUrl = [...pricePage.matchAll(/href="([^"]+)"/g)].find(m => /\/resumo[_-]semanal[_-]lpc[_-].+\.xlsx$/.test(m[1]))?.[1];
    if (!priceUrl) throw new Error('Missing ANP weekly price workbook');
    const sheetRows = await readSheetRows(await read(priceUrl, 'bytes'), 'BRASIL', 'ANP');
    points = [...trade, ...parseAnpPriceRows(sheetRows)];
  }
  if (id === 'mem') {
    // Newest folder first; within a folder the highest suffix is the newest upload.
    let url;
    for (const folder of memUploadFolders(nowMs)) {
      for (const nameYear of folder.nameYears) {
        for (let n = 0; n <= MEM_MAX_UPLOADS_PER_MONTH; n++) {
          const candidate = memWorkbookUrl(folder.year, folder.month, nameYear, n);
          if (!(await exists(candidate))) break;
          url = candidate;
        }
        if (url) break;
      }
      if (url) break;
    }
    if (!url) throw new Error('Missing MEM daily price workbook');
    points = parseMemPriceRows(await readSheetRows(await read(url, 'bytes'), 'PUBLICACIÓN WEB', 'MEM'));
  }
  if (id === 'evn') points = parseEvn(await read(discoverEvnPost(await read(feed.sourceUrl))));
  if (id === 'korem') {
    const summary = await read(KOREM_FIXED_URLS.summary, 'json');
    const maxConsumption = await read(KOREM_FIXED_URLS.maxConsumption, 'json');
    const torgNames = await read(KOREM_FIXED_URLS.torgNames, 'json');
    const zones = await read(KOREM_FIXED_URLS.zones, 'json');
    const series = await read(koremEnergySeriesUrl(summary), 'json');
    const ct = [];
    for (const plan of koremCtUrls(torgNames, zones, await read(koremLastDateUrl(torgNames), 'json'))) {
      ct.push({ zone: plan.zone, prices: await read(plan.prices, 'json'), volumes: await read(plan.volumes, 'json') });
    }
    points = parseKorem({ series, maxConsumption, ct });
  }
  if (id === 'coes') points = parseCoes(await read(COES_DEMAND_URL, 'json', coesDemandBody(nowMs)));
  if (id === 'iccc' || id === 'fccc') {
    // Releases are posted before they take effect. Use the newest release
    // already in effect, reading at most three.
    const candidates = id === 'iccc' ? discoverPngFuelPosts(await read(ICCC_INDEX_URL, 'json')) : discoverFijiFuelPdfs(await read(FCCC_INDEX_URL, 'json'));
    for (const url of candidates.slice(0, 3)) {
      const parsed = id === 'iccc' ? parsePngFuelPost(await read(url, 'json')) : parseFijiFuel(await pdfAllItems(await read(url, 'bytes')));
      points = inEffect(parsed, nowMs, feed.utcOffsetMin);
      if (points.length) break;
    }
    if (!points?.length) throw new Error(`No ${id} fuel price release in effect`);
  }
  if (id === 'qatarenergy') points = parseQatarFuel(await readPdf(QATARENERGY_PDF_URL, 'all'));
  if (id === 'erb') {
    // Each monthly release is posted the day before its prices start; read
    // the newest release in effect, at most three.
    for (const release of discoverErbBuildUps(await read(ERB_INDEX_URL, 'json')).slice(0, 3)) {
      points = inEffect(parseErbBuildUp(await readPdf(release.url), release.period), nowMs, feed.utcOffsetMin);
      if (points.length) break;
    }
    if (!points?.length) throw new Error('No erb pump price release in effect');
  }
  if (id === 'npa') {
    // The two lookups and the two downloads are independent, so each pair
    // runs together: the fetch phase has 80 s, the host answers in 5-20 s,
    // and the 3 MB listing workbook took 3 s to 47 s on 2026-11-02. The
    // worst case is then 20 s + 55 s.
    const [listing, indicators] = await Promise.all([
      read(npaMediaUrl('Ex-Pump Prices of OMCs'), 'json', undefined, 20000).then(media => discoverNpaWorkbook(media, 'expump')),
      read(npaMediaUrl('Petroleum Price Indicators'), 'json', undefined, 20000).then(media => discoverNpaWorkbook(media, 'indicators')),
    ]);
    const [listingBytes, indicatorBytes] = await Promise.all([read(listing.url, 'bytes', undefined, 55000), read(indicators.url, 'bytes', undefined, 20000)]);
    // The workbook keeps one sheet per day with the newest first; only that
    // sheet is read, and it must be the day the upload is named for.
    const prices = parseNpaExPumpRows(await readFirstSheetRows(listingBytes, 'NPA'));
    if (prices[0].period !== listing.day) throw new Error('NPA listing sheet does not match its upload day');
    points = [...prices, ...parseNpaIndicatorRows(await readSheetRows(indicatorBytes, 'Media Publication', 'NPA'))];
  }
  if (id === 'sonabhy') {
    const state = await read(discoverSonabhyState(await read(SONABHY_PAGE_URL)));
    points = parseSonabhyTariffs(await read(discoverSonabhyApi(state), 'json'));
  }
  if (id === 'zera') points = parseZeraPrices(await read(ZERA_HOME_URL));
  if (id === 'doe') {
    // The monitoring sheet is posted during its week; a sheet posted early
    // waits for its week to start, and the previous week serves meanwhile.
    for (const week of discoverDoeNcrPdfs(await read(DOE_NCR_PAGE_URL), nowMs).slice(0, 3)) {
      const parsed = parseDoeNcrPrices(await readPdf(week.url, 'all'));
      if (parsed[0].period !== week.weekStart) throw new Error('DOE monitoring week does not match its link');
      points = inEffect(parsed, nowMs, feed.utcOffsetMin);
      if (points.length) break;
    }
    if (!points?.length) throw new Error('No doe price monitoring week in effect');
  }
  if (id === 'anptl') {
    const [post] = discoverAnpTlPosts(await read(ANPTL_INDEX_URL, 'json'));
    points = parseAnpTlStations(await readPdf(discoverAnpTlPdf(await read(post.url, 'json')), 'all'), post.period);
  }
  if (id === 'creg') points = parseCregPrices(await read(CREG_PAGE_URL));
  if (id === 'ursea') {
    // Maximum prices and reference prices are published days ahead of the
    // month they apply to; each falls back to its previous publication.
    const pages = discoverUrseaPages(await read(URSEA_TOPIC_URL));
    const inForce = async (urls, parse) => {
      for (const url of urls.slice(0, 2)) {
        const current = inEffect(parse(await read(url)), nowMs, feed.utcOffsetMin);
        if (current.length) return current;
      }
      throw new Error('No ursea price publication in effect');
    };
    points = [...await inForce(pages.maximum, parseUrseaMaximumPrices), ...await inForce(pages.reference, parseUrseaReferencePrices)];
  }
  if (feed.publishesAhead) points = inEffect(points, nowMs, feed.utcOffsetMin);
  const snapshot = buildFeedSnapshot(id, points, downloaded.at(-1), nowMs);
  return { ...snapshot, dataUrls: downloaded };
}

// Health reads a future newestItemAt as stale content. Date periods start at
// the provider's local midnight, and items published up to an hour ahead (COES
// schedules) count only once they have started.
export function regionalEnergyContentMeta(data, runStartedAtMs = Date.now()) {
  const offset = feeds.find(f => f.id === data.sourceId)?.utcOffsetMin;
  const times = data.observations.map(p => effectiveAt(p.period, offset)).filter(t => t <= runStartedAtMs);
  if (!times.length) return null;
  return { newestItemAt: Math.max(...times), oldestItemAt: Math.min(...times) };
}

export function regionalEnergySeedOptions(id) {
  const feed = feeds.find(f => f.id === id);
  if (!feed) throw new Error('Unknown regional energy feed');
  return {
    ttlSeconds: 14 * 86400,
    sourceVersion: `regional-energy-${id}-v1`,
    schemaVersion: 1,
    declareRecords: data => data.observations.length,
    recordCount: data => data.observations.length,
    validateFn: data => data.sourceId === id && data.observations.length > 0,
    emptyDataIsFailure: true,
    recordInitialFetchFailure: true,
    maxStaleMin: feed.intervalMin * 3,
    contentMeta: regionalEnergyContentMeta,
    maxContentAgeMin: feed.maxContentAgeMin,
    lockTtlMs: 300000,
    fetchPhaseTimeoutMs: 240000,
    afterFreshness: async () => {
      const creds = getOptionalUpstashCreds();
      if (!creds) throw new Error('Missing regional energy activation credentials');
      await upstashCommand(creds, ['SET', `seed-activated:energy:regional-${id}`, '1']);
    },
  };
}