// Connecteur incrémental « nouvelles ventes » — source api.qub.ca
// (le même feed que le widget « Transactions immobilières » du Journal de
// Montréal, https://www.journaldemontreal.com/argent/immobilier/transactions-immobilieres).
//
// Contrairement au pipeline batch complet (docs/PIPELINE-DONNEES.md, qui vit sur
// le laptop et reconstruit tout le .db), ce connecteur tourne SUR LE NŒUD et
// ajoute uniquement les ventes récentes directement dans data/vraiprix.db :
//
//   1. Récupère un jeton Bearer QUB (scripts/qub-token.mjs, via Scrapfly).
//   2. Liste les ~1478 secteurs de /real-estate-service/v1/locations/all.
//   3. Pour chaque secteur, interroge /v1/map (résultats triés du plus récent
//      au plus ancien, plafonnés à `nb_transactions`). Si un secteur sature le
//      plafond sans atteindre la date plancher, on subdivise sa bbox (quadtree).
//   4. Mappe chaque vente vers le schéma `transactions`, joint spatialement à
//      `units` pour id_provinc / valeur_role, puis INSERT OR IGNORE (dédup par
//      id — identique au pipeline batch, la source fournit le même id nanoid).
//
// Les `id` déjà connus sont ignorés : lancer ce script régulièrement (cron/pm2)
// garde la base à jour sans doublon. L'app lit le .db en WAL, les nouvelles
// lignes sont visibles immédiatement, sans redémarrage.
//
// Env : SCRAPFLY_KEY, QUB_EMAIL, QUB_PASSWORD (login), VRAIPRIX_DB (optionnel).
// Args : --since=YYYY-MM-DD  date plancher (défaut : max(date en base) - 45 j)
//        --region="Nom"      ne traiter que ce(s) secteur(s) (sous-chaîne)
//        --dry-run           n'écrit rien, affiche seulement le décompte
//        --concurrency=N     requêtes /map simultanées (défaut 6)

import Database from 'better-sqlite3';
import fs from 'node:fs';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { getQubToken } from './qub-token.mjs';

// Charge .env.local (racine du repo) si présent — rend le script autonome sous
// cron/pm2 (SCRAPFLY_KEY, QUB_EMAIL, QUB_PASSWORD, QUB_SESSION, VRAIPRIX_DB).
(function loadEnvLocal() {
  const envPath = path.join(fileURLToPath(new URL('..', import.meta.url)), '.env.local');
  if (!fs.existsSync(envPath)) return;
  for (const line of fs.readFileSync(envPath, 'utf8').split('\n')) {
    const m = line.match(/^\s*([A-Z0-9_]+)\s*=\s*(.*)\s*$/);
    if (m && !(m[1] in process.env)) process.env[m[1]] = m[2].replace(/^["']|["']$/g, '');
  }
})();

const API = 'https://api.qub.ca/real-estate-service/v1';
const PAGE_CAP = 500; // plafond de /map ; si atteint on subdivise
const HDRS = {
  Origin: 'https://www.journaldemontreal.com',
  Referer: 'https://www.journaldemontreal.com/',
  'User-Agent':
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36',
};

// ---------- args ----------
const args = Object.fromEntries(
  process.argv.slice(2).map((a) => {
    const m = a.match(/^--([^=]+)(?:=(.*))?$/);
    return m ? [m[1], m[2] ?? true] : [a, true];
  }),
);
const DRY = !!args['dry-run'];
const CONC = Math.max(1, parseInt(args.concurrency || '6', 10));
const REGION = args.region ? String(args.region).toLowerCase() : null;

// ---------- db ----------
const DB_PATH =
  process.env.VRAIPRIX_DB || path.join(process.cwd(), 'data', 'vraiprix.db');
const db = new Database(DB_PATH);
db.pragma('journal_mode = WAL');
db.pragma('busy_timeout = 15000');

const sinceArg = typeof args.since === 'string' ? args.since : null;
const maxDate = db.prepare('SELECT MAX(date) d FROM transactions').get().d;
function minusDays(iso, n) {
  const d = new Date(iso + 'T00:00:00Z');
  d.setUTCDate(d.getUTCDate() - n);
  return d.toISOString().slice(0, 10);
}
const SINCE = sinceArg || minusDays(maxDate || '2026-01-01', 45);

// ---------- helpers ----------
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

async function apiGet(url, token, tries = 4) {
  for (let i = 0; i < tries; i++) {
    let res;
    try {
      res = await fetch(url, { headers: { ...HDRS, Authorization: `Bearer ${token}` } });
    } catch (e) {
      if (i === tries - 1) throw e;
      await sleep(500 * (i + 1));
      continue;
    }
    if (res.ok) return res.json();
    if (res.status === 401 || res.status === 403)
      throw Object.assign(new Error(`auth ${res.status}`), { authError: true });
    if (res.status === 429 || res.status >= 500) {
      await sleep(800 * (i + 1));
      continue;
    }
    throw new Error(`HTTP ${res.status} sur ${url}`);
  }
  throw new Error(`échec après ${tries} essais : ${url}`);
}

async function fetchLocations(token) {
  const list = await apiGet(`${API}/locations/all`, token);
  return list
    .map((l) => {
      // shape.coordinates est un anneau « plat » [[lng,lat],…] chez cette API
      // (parfois imbriqué [[[lng,lat],…]] en GeoJSON standard) — on gère les deux.
      let ring = l.shape?.coordinates || [];
      if (Array.isArray(ring[0]?.[0])) ring = ring[0];
      if (!ring.length || typeof ring[0]?.[0] !== 'number') return null;
      const lngs = ring.map((c) => c[0]);
      const lats = ring.map((c) => c[1]);
      return {
        name: l.name,
        sw_lat: Math.min(...lats),
        ne_lat: Math.max(...lats),
        sw_lng: Math.min(...lngs),
        ne_lng: Math.max(...lngs),
      };
    })
    .filter(Boolean);
}

// Récupère les ventes d'une bbox, subdivise (quadtree) si le plafond est saturé
// alors que les ventes retournées sont encore >= SINCE (⇒ il en manque).
async function fetchBox(box, token, out, depth = 0) {
  const url =
    `${API}/map?ne_lat=${box.ne_lat}&ne_lng=${box.ne_lng}` +
    `&sw_lat=${box.sw_lat}&sw_lng=${box.sw_lng}&nb_transactions=${PAGE_CAP}&cid=0`;
  const rows = await apiGet(url, token);
  for (const t of rows) if (t.date >= SINCE) out.set(t.id, t);

  const oldest = rows.length ? rows[rows.length - 1].date : null;
  const saturated = rows.length >= PAGE_CAP && oldest && oldest >= SINCE;
  if (saturated && depth < 6) {
    const mlat = (box.ne_lat + box.sw_lat) / 2;
    const mlng = (box.ne_lng + box.sw_lng) / 2;
    const quads = [
      { sw_lat: box.sw_lat, sw_lng: box.sw_lng, ne_lat: mlat, ne_lng: mlng },
      { sw_lat: box.sw_lat, sw_lng: mlng, ne_lat: mlat, ne_lng: box.ne_lng },
      { sw_lat: mlat, sw_lng: box.sw_lng, ne_lat: box.ne_lat, ne_lng: mlng },
      { sw_lat: mlat, sw_lng: mlng, ne_lat: box.ne_lat, ne_lng: box.ne_lng },
    ];
    for (const q of quads) await fetchBox(q, token, out, depth + 1);
  }
}

// ---------- spatial join vers units (id_provinc, valeur_role) ----------
const unitsInBox = db.prepare(
  `SELECT id_provinc, lat, lng, valeur_role, type_prop FROM units
   WHERE lat BETWEEN ? AND ? AND lng BETWEEN ? AND ?`,
);
function haversine(aLat, aLng, bLat, bLng) {
  const R = 6371000, toR = Math.PI / 180;
  const dLat = (bLat - aLat) * toR, dLng = (bLng - aLng) * toR;
  const s =
    Math.sin(dLat / 2) ** 2 +
    Math.cos(aLat * toR) * Math.cos(bLat * toR) * Math.sin(dLng / 2) ** 2;
  return 2 * R * Math.asin(Math.sqrt(s));
}
// Rayons d'appariement croissants (~65 m, ~220 m, ~550 m).
const DEGS = [0.0006, 0.002, 0.005];
function matchUnit(lat, lng, propertyType) {
  for (const d of DEGS) {
    const cands = unitsInBox.all(lat - d, lat + d, lng - d, lng + d);
    if (!cands.length) continue;
    let best = null, bestDist = Infinity;
    for (const u of cands) {
      let dist = haversine(lat, lng, u.lat, u.lng);
      if (propertyType && u.type_prop && u.type_prop !== propertyType) dist += 25; // léger biais type
      if (dist < bestDist) { bestDist = dist; best = u; }
    }
    if (best) return best;
  }
  return null;
}

function toRow(t) {
  const coords = t.geometries?.[0]?.coordinates || [];
  const lng = coords[0], lat = coords[1];
  if (typeof lat !== 'number' || typeof lng !== 'number') return null;
  const ar = t.ar || {};
  const yb = ar.yearBuilt != null ? parseInt(ar.yearBuilt, 10) : null;
  const unit = matchUnit(lat, lng, t.propertyType);
  return {
    id: t.id,
    date: t.date,
    amount: t.amount,
    street: t.address?.street ?? null,
    city: t.address?.city ?? null,
    lat,
    lng,
    property_type: t.propertyType ?? null,
    year_built: Number.isFinite(yb) ? yb : null,
    floor_area: ar.floorArea ?? null,
    building_type: ar.buildingType ?? null,
    id_provinc: unit?.id_provinc ?? null,
    valeur_role: unit?.valeur_role ?? ar.totalArValue ?? null,
    land_area: ar.parcelArea ?? null,
  };
}

const insert = db.prepare(
  `INSERT OR IGNORE INTO transactions
   (id,date,amount,street,city,lat,lng,property_type,year_built,floor_area,
    building_type,id_provinc,valeur_role,land_area)
   VALUES (@id,@date,@amount,@street,@city,@lat,@lng,@property_type,@year_built,
    @floor_area,@building_type,@id_provinc,@valeur_role,@land_area)`,
);

// ---------- main ----------
async function main() {
  console.log(`[ingest-jdm] base : ${DB_PATH}`);
  console.log(`[ingest-jdm] date max en base : ${maxDate} → plancher SINCE=${SINCE}`);
  if (DRY) console.log('[ingest-jdm] DRY-RUN : aucune écriture');

  const token = await getQubToken({
    session: process.env.QUB_SESSION || 'qub-vraiprix',
    log: (m) => console.log('[ingest-jdm]', m),
  });

  let locations = await fetchLocations(token);
  if (REGION) locations = locations.filter((l) => l.name.toLowerCase().includes(REGION));
  console.log(`[ingest-jdm] secteurs à balayer : ${locations.length}`);

  // Balayage concurrent par secteur.
  const collected = new Map();
  let done = 0;
  async function worker(queue) {
    for (;;) {
      const box = queue.pop();
      if (!box) return;
      try {
        await fetchBox(box, token, collected);
      } catch (e) {
        if (e.authError) throw e;
        console.warn(`[ingest-jdm] secteur "${box.name}" : ${e.message}`);
      }
      if (++done % 100 === 0)
        console.log(`[ingest-jdm]   ${done}/${locations.length} secteurs, ${collected.size} ventes ≥ ${SINCE}`);
    }
  }
  const queue = [...locations];
  await Promise.all(Array.from({ length: CONC }, () => worker(queue)));
  console.log(`[ingest-jdm] ventes candidates (≥ ${SINCE}) : ${collected.size}`);

  // Mapping + insertion.
  const before = db.prepare('SELECT COUNT(*) n FROM transactions').get().n;
  let inserted = 0, skippedGeo = 0, matched = 0;
  const rows = [];
  for (const t of collected.values()) {
    const r = toRow(t);
    if (!r) { skippedGeo++; continue; }
    if (r.id_provinc) matched++;
    rows.push(r);
  }

  if (!DRY) {
    const tx = db.transaction((batch) => {
      for (const r of batch) inserted += insert.run(r).changes;
    });
    tx(rows);
  }

  const after = DRY ? before : db.prepare('SELECT COUNT(*) n FROM transactions').get().n;
  console.log('[ingest-jdm] ---------- résumé ----------');
  console.log(`  candidates mappées   : ${rows.length}`);
  console.log(`  appariées à une unité: ${matched} (${rows.length ? ((matched / rows.length) * 100).toFixed(1) : 0} %)`);
  console.log(`  sans géométrie       : ${skippedGeo}`);
  console.log(`  NOUVELLES insérées   : ${DRY ? '(dry-run)' : inserted}`);
  console.log(`  transactions totales : ${before} → ${after}`);
  const nd = db.prepare('SELECT MAX(date) d FROM transactions').get().d;
  console.log(`  date max en base     : ${nd}`);
  db.close();
}

main().catch((e) => {
  console.error('[ingest-jdm] ERREUR', e);
  db.close();
  process.exit(1);
});