// Toveli — synchronisation nocturne : extraits minces (lecture seule) des trois sources → data/toveli.db
//
//   acquisition autorisée → validation → normalisation → rapprochement (propriété) → enrichissement
//   (valuations, scores) → indexation (FTS) → publication (bascule atomique + pm2 restart)
//
// Sources (toutes en LECTURE SEULE, instantané par requête sqlite3 ATTACH sur le nœud source) :
//   • Vrai-Prix (M3U96a) data/immoka.db  → annonces à vendre (copie Immo-Ka de la nuit) + vp_eval (estimations)
//   • Vrai-Prix (M3U96a) data/vraiprix.db → rôle MAMH (units, licence ouverte Données Québec), transactions, market_index
//   • Lou-Ka   (M4M64a) data/louka.db     → logements à louer canoniques + fairvalue + kascores + price_log
//
// Usage : pnpm sync            (complet)      SKIP_FETCH=1 pnpm sync (réutilise data/src/*.db)
//         NO_RESTART=1 pnpm sync              TOVELI_SSH_KEY="" pour utiliser l'agent SSH (laptop)
import Database from "better-sqlite3";
import { execFileSync, spawnSync } from "node:child_process";
import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, statSync, writeFileSync } from "node:fs";
import { homedir } from "node:os";
import path from "node:path";
import {
  addressKey, coordKey, epochToISO, normRegion, rentCategory, saleCategory, sha1Short, displayCity, normCity,
} from "../src/lib/normalize";

const ROOT = path.resolve(path.dirname(new URL(import.meta.url).pathname), "..");
const DATA = path.join(ROOT, "data");
const SRC = path.join(DATA, "src");
const LIVE = path.join(DATA, "toveli.db");
const NEW = path.join(DATA, "toveli.db.new");
const LOG = path.join(DATA, "sync.log");
mkdirSync(SRC, { recursive: true });

const env = process.env;
const VP_SSH = env.TOVELI_VP_SSH ?? "simon-pierreboucher@192.168.2.80";   // M3U96a (LAN)
const LK_SSH = env.TOVELI_LK_SSH ?? "simon-pierreboucher@192.168.2.70";   // M4M64a (LAN)
const VP_DIR = env.TOVELI_VP_DIR ?? "/Users/simon-pierreboucher/apps/vrai-prix/data";
const LK_DB = env.TOVELI_LK_DB ?? "/Users/simon-pierreboucher/apps/lou-ka/data/louka.db";
const SSH_KEY = env.TOVELI_SSH_KEY === undefined ? path.join(homedir(), ".ssh/ka_guardian_ed25519") : env.TOVELI_SSH_KEY;
const PM2_APP = env.PM2_APP ?? "toveli";
const T0 = Date.now();

function log(msg: string) {
  const line = `[${new Date().toISOString().slice(0, 19).replace("T", " ")}] ${msg}`;
  console.log(line);
  try { writeFileSync(LOG, line + "\n", { flag: "a" }); } catch { /* ignore */ }
}
function die(msg: string): never { log(`ÉCHEC : ${msg}`); process.exit(1); }

function sshArgs(target: string): string[] {
  const a = ["-o", "BatchMode=yes", "-o", "ConnectTimeout=20", "-o", "StrictHostKeyChecking=accept-new"];
  if (SSH_KEY) a.push("-i", SSH_KEY, "-o", "IdentitiesOnly=yes", "-o", "ControlMaster=no", "-o", "ControlPath=none");
  a.push(target);
  return a;
}
function sshRun(target: string, cmd: string, stdin?: string): string {
  const r = spawnSync("ssh", [...sshArgs(target), cmd], { input: stdin, encoding: "utf8", maxBuffer: 64 * 1024 * 1024 });
  if (r.status !== 0) throw new Error(`ssh ${target} « ${cmd.slice(0, 60)} » : ${r.stderr || r.stdout}`);
  return r.stdout;
}
function rsyncBack(target: string, remote: string, local: string) {
  const e = ["ssh", ...sshArgs(target).slice(0, -1)].join(" ");
  const r = spawnSync("rsync", ["-a", "--inplace", "--no-whole-file", "-e", e, `${target}:${remote}`, local], { encoding: "utf8" });
  if (r.status !== 0) throw new Error(`rsync ${remote} : ${r.stderr}`);
}

/** Exécute un script sqlite3 sur le nœud source : la base source est ouverte telle quelle (aucune écriture), la sortie va dans un fichier /tmp. */
function remoteExtract(target: string, srcDb: string, outName: string, sql: string) {
  const remoteOut = `/tmp/toveli-${outName}.db`;
  const local = path.join(SRC, `${outName}.db`);
  log(`  ↳ extraction ${outName} sur ${target.split("@").pop()}…`);
  const script = `.bail on\nATTACH '${remoteOut}' AS out;\n${sql}\nDETACH out;\n`;
  sshRun(target, `rm -f ${remoteOut} && /usr/bin/sqlite3 '${srcDb}'`, script);
  const size = sshRun(target, `stat -f %z ${remoteOut}`).trim();
  log(`  ↳ ${outName} : ${(Number(size) / 1e6).toFixed(0)} Mo, rapatriement…`);
  rsyncBack(target, remoteOut, local);
  sshRun(target, `rm -f ${remoteOut}`);
  return local;
}

const LIVE_SALE = "status='a-vendre' AND published=1 AND active=1 AND dup_hidden=0 AND price>0";
const LIVE_RENT = "active=1 AND published=1 AND dup_of IS NULL AND province='QC' AND price>0";

function fetchAll() {
  log("1/6 extraction des instantanés (lecture seule)");
  remoteExtract(VP_SSH, `${VP_DIR}/immoka.db`, "immoka", `
    CREATE TABLE out.listings AS SELECT uid, source, external_id, url, title, address, sector, city, region, property_type, price, price_label,
      bedrooms, bathrooms, powder_rooms, area_sqft, lot_sqft, year_built, mls, status, agency, description, features, details, images,
      lat, lng, first_seen, last_seen, updated_at, quality_score FROM listings WHERE ${LIVE_SALE};
    CREATE TABLE out.vp_eval AS SELECT * FROM vp_eval;
    CREATE TABLE out.price_log AS SELECT p.uid, p.ts, p.price FROM price_log p WHERE p.uid IN (SELECT uid FROM out.listings);
  `);
  remoteExtract(VP_SSH, `${VP_DIR}/vraiprix.db`, "vraiprix", `
    CREATE TABLE out.units AS SELECT id_provinc, adresse, apt, municipalite, code_mun, lat, lng, cubf, cubf_libelle, type_prop, annee_construction,
      aire_etages_m2, superficie_terrain_m2, nb_etages, nb_logements, valeur_terrain, valeur_batiment, valeur_role, dat_cond_marche, est_2026, p10, p90
      FROM units WHERE lat IS NOT NULL;
    CREATE TABLE out.transactions AS SELECT * FROM transactions WHERE amount >= 50000 AND lat IS NOT NULL;
    CREATE TABLE out.market_index AS SELECT * FROM market_index;
  `);
  rsyncBack(VP_SSH, `${VP_DIR}/marche-stats.json`, path.join(SRC, "vp-marche-stats.json"));
  remoteExtract(LK_SSH, LK_DB, "louka", `
    CREATE TABLE out.listings AS SELECT uid, source, external_id, url, title, address, sector, city, unit_type, price, price_label, availability,
      availability_date, description, amenities, images, images_ok, lat, lng, first_seen, last_seen, updated_at, area_sqft, pets, furnished, details,
      bedrooms, bathrooms, completeness, building_key, coord_key FROM listings WHERE ${LIVE_RENT};
    CREATE TABLE out.fairvalue AS SELECT f.* FROM fairvalue f WHERE f.uid IN (SELECT uid FROM out.listings);
    CREATE TABLE out.kascores AS SELECT coord_key, lat, lng, walk, transit, bike, calme, services, global, version, computed_at FROM kascores;
    CREATE TABLE out.price_log AS SELECT p.uid, p.ts, p.price FROM price_log p WHERE p.uid IN (SELECT uid FROM out.listings);
  `);
}

// ─── construction ───────────────────────────────────────────────────────────
function num(v: unknown): number | null { if (v == null || v === "") return null; const n = Number(v); return Number.isFinite(n) ? n : null; }
function jsonArr(v: unknown): string[] { if (!v) return []; try { const a = JSON.parse(String(v)); return Array.isArray(a) ? a.filter((x) => typeof x === "string") : []; } catch { return []; } }
function median(a: number[]): number | null { if (!a.length) return null; const s = [...a].sort((x, y) => x - y); const m = Math.floor(s.length / 2); return s.length % 2 ? s[m] : (s[m - 1] + s[m]) / 2; }
function quantile(a: number[], q: number): number | null { if (!a.length) return null; const s = [...a].sort((x, y) => x - y); return s[Math.min(s.length - 1, Math.floor(q * s.length))]; }

function build() {
  log("2/6 construction de toveli.db.new");
  rmSync(NEW, { force: true });
  const db = new Database(NEW);
  db.pragma("foreign_keys = OFF");
  db.exec(readFileSync(path.join(ROOT, "scripts/schema.sql"), "utf8"));
  db.exec(`ATTACH '${path.join(SRC, "immoka.db")}' AS ik; ATTACH '${path.join(SRC, "vraiprix.db")}' AS vp; ATTACH '${path.join(SRC, "louka.db")}' AS lk;`);
  // Connexion de lecture séparée : better-sqlite3 interdit d'écrire pendant qu'un itérateur est ouvert sur la même connexion.
  const rd = new Database(":memory:");
  rd.exec(`ATTACH '${path.join(SRC, "immoka.db")}' AS ik; ATTACH '${path.join(SRC, "vraiprix.db")}' AS vp; ATTACH '${path.join(SRC, "louka.db")}' AS lk;`);
  const now = new Date().toISOString();

  // Sources + provenance
  const nSale = (db.prepare("SELECT count(*) c FROM ik.listings").get() as { c: number }).c;
  const nRent = (db.prepare("SELECT count(*) c FROM lk.listings").get() as { c: number }).c;
  const nTx = (db.prepare("SELECT count(*) c FROM vp.transactions").get() as { c: number }).c;
  const nUnits = (db.prepare("SELECT count(*) c FROM vp.units").get() as { c: number }).c;
  if (nSale < 10000) die(`instantané Immo-Ka suspect (${nSale} annonces vivantes)`);
  if (nRent < 5000) die(`instantané Lou-Ka suspect (${nRent} logements vivants)`);
  const insSrc = db.prepare("INSERT INTO source VALUES (?,?,?,?,?,?,?,?,?)");
  insSrc.run("immo-ka", "Immo-Ka", "listings-sale", "M4M64b→M3U96a", "~/apps/vrai-prix/data/immoka.db", now, nSale, "Agrégateur Groupe Ka ; données factuelles publiques ; photos hotlinkées, droits du site source", "Copie nocturne Vrai-Prix de la base Immo-Ka (03:30) + vp_eval");
  insSrc.run("lou-ka", "Lou-Ka", "listings-rent", "M4M64a", "~/apps/lou-ka/data/louka.db", now, nRent, "Agrégateur Groupe Ka ; données factuelles publiques ; photos hotlinkées, droits du site source", "Logements canoniques actifs (dup_of IS NULL), fairvalue fv-logement-1.0, KA Scores");
  insSrc.run("vrai-prix", "Vrai-Prix", "valuation", "M3U96a", "~/apps/vrai-prix/data/vraiprix.db", now, nTx, "Rôle d'évaluation MAMH (Données Québec, licence ouverte, attribution requise) ; transactions : flux qub.ca — usage interne au moteur, non republié brut", `units ${nUnits}, transactions ${nTx}, market_index`);

  // Rôle, transactions, indice
  log("  ↳ rôle MAMH + transactions + indice");
  db.exec(`INSERT INTO role_unit SELECT id_provinc, adresse, apt, municipalite, code_mun, lat, lng, cubf, cubf_libelle, type_prop, annee_construction,
    aire_etages_m2, superficie_terrain_m2, nb_logements, nb_etages, valeur_terrain, valeur_batiment, valeur_role, dat_cond_marche, est_2026, p10, p90 FROM vp.units;`);
  db.exec(`INSERT INTO sale_transaction SELECT id, date, amount, street, city, lat, lng, property_type, year_built, floor_area, land_area, id_provinc, valeur_role FROM vp.transactions;`);
  db.exec(`INSERT INTO market_index SELECT month, type_prop, ppm2, idx, n FROM vp.market_index;`);

  // Scores de quartier
  db.exec(`INSERT OR IGNORE INTO neighbourhood_score SELECT coord_key, lat, lng, walk, transit, bike, calme, services, global, version,
    datetime(computed_at, 'unixepoch'), 'lou-ka' FROM lk.kascores;`);

  // Annonces → propriétés
  log("  ↳ annonces à vendre (Immo-Ka) → propriétés");
  const insProp = db.prepare(`INSERT OR IGNORE INTO property (id, key_kind, key_value, address, unit, city, region, lat, lng, category, bedrooms, bathrooms,
    area_sqft, lot_sqft, year_built, role_value, role_land, role_building, role_unit_id, dedup_confidence, updated_at)
    VALUES (@id,@key_kind,@key_value,@address,@unit,@city,@region,@lat,@lng,@category,@bedrooms,@bathrooms,@area_sqft,@lot_sqft,@year_built,@role_value,@role_land,@role_building,@role_unit_id,@dedup_confidence,@updated_at)`);
  const insListing = db.prepare(`INSERT OR REPLACE INTO listing (uid, property_id, source_id, connector, external_id, market, url, status, title, description, price, price_label,
    address, unit, city, region, sector, lat, lng, category, type_label, bedrooms, bathrooms, powder_rooms, area_sqft, lot_sqft, year_built, features, details, images,
    image_count, agency, broker, mls, availability, furnished, pets, first_seen, last_seen, updated_at, quality_score, dup_of, provenance)
    VALUES (@uid,@property_id,@source_id,@connector,@external_id,@market,@url,@status,@title,@description,@price,@price_label,@address,@unit,@city,@region,@sector,@lat,@lng,
    @category,@type_label,@bedrooms,@bathrooms,@powder_rooms,@area_sqft,@lot_sqft,@year_built,@features,@details,@images,@image_count,@agency,@broker,@mls,@availability,
    @furnished,@pets,@first_seen,@last_seen,@updated_at,@quality_score,@dup_of,@provenance)`);
  const insVal = db.prepare(`INSERT OR REPLACE INTO valuation (listing_uid, property_id, model, model_version, estimate, low, high, confidence_pct, confidence, method_detail, ratio, diff, verdict, evaluated_at)
    VALUES (@listing_uid,@property_id,@model,@model_version,@estimate,@low,@high,@confidence_pct,@confidence,@method_detail,@ratio,@diff,@verdict,@evaluated_at)`);
  const getUnit = db.prepare("SELECT id, role_value, role_land, role_building FROM role_unit WHERE id = ?");
  const getEval = rd.prepare("SELECT * FROM ik.vp_eval WHERE uid = ?");

  type Row = Record<string, unknown>;
  const verdictOf = (ratio: number | null) => ratio == null ? "na" : ratio > 1.05 ? "sous" : ratio < 0.95 ? "sur" : "aligne";

  const txSale = db.transaction(() => {
    let i = 0;
    for (const r of rd.prepare("SELECT * FROM ik.listings").iterate() as Iterable<Row>) {
      const lat = num(r.lat), lng = num(r.lng);
      const city = displayCity(r.city as string);
      const region = normRegion(r.region as string, city);
      const cat = saleCategory(r.property_type as string, r.title as string);
      const ev = getEval.get(r.uid) as Row | undefined;
      // identité : rôle (jumelage spatial ≤ 40 m) > adresse normalisée > cellule de coordonnées
      let keyKind = "coord", keyValue: string | null = null, conf = 0.4, unit: Row | undefined;
      if (ev?.unit_id && num(ev.match_m) != null && (num(ev.match_m) as number) <= 40) {
        unit = getUnit.get(ev.unit_id) as Row | undefined;
        if (unit) { keyKind = "role"; keyValue = String(ev.unit_id); conf = 0.9; }
      }
      if (!keyValue) { const ak = addressKey(r.address as string, city); if (ak) { keyKind = "address"; keyValue = ak; conf = 0.7; } }
      if (!keyValue) { if (lat == null || lng == null) continue; keyValue = coordKey(lat, lng) + "|" + cat; }
      const pid = sha1Short(`${keyKind}:${keyValue}`);
      const images = jsonArr(r.images);
      const details = r.details ? String(r.details) : null;
      insProp.run({
        id: pid, key_kind: keyKind, key_value: keyValue, address: r.address ?? null, unit: null, city, region, lat, lng, category: cat,
        bedrooms: num(r.bedrooms), bathrooms: num(r.bathrooms), area_sqft: num(r.area_sqft), lot_sqft: num(r.lot_sqft), year_built: num(r.year_built),
        role_value: unit ? num(unit.role_value) : num(ev?.valeur_role), role_land: unit ? num(unit.role_land) : null, role_building: unit ? num(unit.role_building) : null,
        role_unit_id: unit ? String(unit.id) : null, dedup_confidence: conf, updated_at: now,
      });
      const uid = `immo-ka:${r.uid}`;
      insListing.run({
        uid, property_id: pid, source_id: "immo-ka", connector: r.source, external_id: r.external_id, market: "sale", url: r.url ?? null, status: "a-vendre",
        title: r.title ?? null, description: r.description ?? null, price: num(r.price), price_label: r.price_label ?? null, address: r.address ?? null, unit: null,
        city, region, sector: r.sector ?? null, lat, lng, category: cat, type_label: r.property_type || null, bedrooms: num(r.bedrooms), bathrooms: num(r.bathrooms),
        powder_rooms: num(r.powder_rooms), area_sqft: num(r.area_sqft), lot_sqft: num(r.lot_sqft), year_built: num(r.year_built),
        features: r.features ? String(r.features) : "[]", details, images: JSON.stringify(images), image_count: images.length,
        agency: r.agency ?? null, broker: null, mls: (r.mls as string) || null, availability: null, furnished: null, pets: null,
        first_seen: epochToISO(r.first_seen as number), last_seen: epochToISO(r.last_seen as number), updated_at: epochToISO(r.updated_at as number),
        quality_score: num(r.quality_score), dup_of: null,
        provenance: JSON.stringify({ source_uid: r.uid, source: "immo-ka", connector: r.source, collected_at: epochToISO(r.last_seen as number), origin_url: r.url ?? null, media: "hotlink-source" }),
      });
      if (ev && num(ev.est)) {
        const est = num(ev.est) as number, price = num(r.price) as number;
        const ratio = price > 0 ? est / price : null;
        insVal.run({
          listing_uid: uid, property_id: pid, model: "vrai-prix-hybride", model_version: "toveli-estim-2026.10 (hédonique 2026-08 + comparables ajustés)", estimate: est, low: num(ev.low), high: num(ev.high),
          confidence_pct: num(ev.confidence_pct), confidence: ev.confidence ?? null,
          method_detail: JSON.stringify({ model_est: num(ev.model_est), comps_est: num(ev.comps_est), model_weight: num(ev.model_weight), n_comps: num(ev.n_comps), cost_est: num(ev.cost_est), role_est: num(ev.role_est), ens_est: num(ev.ens_est), valeur_role: num(ev.valeur_role), type_prop: ev.type_prop, match_m: num(ev.match_m), unit_id: ev.unit_id }),
          ratio, diff: price - est, verdict: verdictOf(ratio), evaluated_at: ev.evaluated_at ?? null,
        });
      }
      if (++i % 20000 === 0) log(`     ${i} annonces…`);
    }
    db.exec(`INSERT INTO price_event SELECT 'immo-ka:'||p.uid, strftime('%Y-%m-%dT%H:%M:%SZ', p.ts, 'unixepoch'), p.price, 'price' FROM ik.price_log p JOIN listing l ON l.uid = 'immo-ka:'||p.uid WHERE p.price IS NOT NULL;`);
  });
  txSale();

  log("  ↳ logements à louer (Lou-Ka) → immeubles/propriétés");
  const getFv = rd.prepare("SELECT * FROM lk.fairvalue WHERE uid = ?");
  const txRent = db.transaction(() => {
    let i = 0;
    for (const r of rd.prepare("SELECT * FROM lk.listings").iterate() as Iterable<Row>) {
      const lat = num(r.lat), lng = num(r.lng);
      const city = displayCity(r.city as string);
      const region = normRegion(null, city);
      const cat = rentCategory(r.unit_type as string);
      let keyKind = "address", keyValue: string | null = addressKey(r.address as string, city), conf = 0.7;
      if (!keyValue && r.building_key) { keyValue = String(r.building_key); conf = 0.6; }
      if (!keyValue) { if (lat == null || lng == null) continue; keyKind = "coord"; keyValue = coordKey(lat, lng) + "|immeuble"; conf = 0.4; }
      const pid = sha1Short(`${keyKind}:${keyValue}`);
      const images = jsonArr(r.images_ok).length ? jsonArr(r.images_ok) : jsonArr(r.images);
      insProp.run({
        id: pid, key_kind: keyKind, key_value: keyValue, address: r.address ?? null, unit: null, city, region, lat, lng, category: cat === "logement" || cat === "chambre" ? "immeuble" : cat,
        bedrooms: null, bathrooms: null, area_sqft: null, lot_sqft: null, year_built: null, role_value: null, role_land: null, role_building: null, role_unit_id: null,
        dedup_confidence: conf, updated_at: now,
      });
      const uid = `lou-ka:${r.uid}`;
      const details = r.details ? String(r.details) : null;
      insListing.run({
        uid, property_id: pid, source_id: "lou-ka", connector: r.source, external_id: r.external_id, market: "rent", url: r.url ?? null, status: "a-louer",
        title: r.title ?? null, description: r.description ?? null, price: num(r.price), price_label: r.price_label ?? null, address: r.address ?? null, unit: null,
        city, region, sector: r.sector ?? null, lat, lng, category: cat, type_label: (r.unit_type as string) || null, bedrooms: num(r.bedrooms), bathrooms: num(r.bathrooms),
        powder_rooms: null, area_sqft: num(r.area_sqft), lot_sqft: null, year_built: null, features: r.amenities ? String(r.amenities) : "[]", details,
        images: JSON.stringify(images), image_count: images.length, agency: null, broker: null, mls: null,
        availability: (r.availability_date as string) || (r.availability as string) || null, furnished: num(r.furnished), pets: (r.pets as string) || null,
        first_seen: epochToISO(r.first_seen as number), last_seen: epochToISO(r.last_seen as number), updated_at: epochToISO(r.updated_at as number),
        quality_score: num(r.completeness), dup_of: null,
        provenance: JSON.stringify({ source_uid: r.uid, source: "lou-ka", connector: r.source, collected_at: epochToISO(r.last_seen as number), origin_url: r.url ?? null, media: "hotlink-source" }),
      });
      const fv = getFv.get(r.uid) as Row | undefined;
      if (fv && num(fv.fv)) {
        const est = num(fv.fv) as number, price = num(r.price) as number;
        const ratio = price > 0 ? est / price : null;
        insVal.run({
          listing_uid: uid, property_id: pid, model: "lou-ka-fairvalue", model_version: (fv.model_version as string) ?? "fv-logement-1.0", estimate: est, low: num(fv.fv_low), high: num(fv.fv_high),
          confidence_pct: null, confidence: (fv.confidence as string) ?? null,
          method_detail: JSON.stringify({ method: fv.method, comps: num(fv.comps), segment: fv.segment, deviation: num(fv.deviation), source_verdict: fv.verdict }),
          ratio, diff: price - est, verdict: verdictOf(ratio), evaluated_at: epochToISO(fv.computed_at as number),
        });
      }
      if (++i % 10000 === 0) log(`     ${i} logements…`);
    }
    db.exec(`INSERT INTO price_event SELECT 'lou-ka:'||p.uid, strftime('%Y-%m-%dT%H:%M:%SZ', p.ts, 'unixepoch'), p.price, 'price' FROM lk.price_log p JOIN listing l ON l.uid = 'lou-ka:'||p.uid WHERE p.price IS NOT NULL;`);
  });
  txRent();

  log("3/6 agrégats par propriété, lieux et métriques de marché");
  {
    type Agg = { property_id: string; n: number; n_sale: number; n_rent: number; min_price: number | null; max_price: number | null; bedrooms: number | null; bathrooms: number | null; area_sqft: number | null; year_built: number | null };
    const aggs = db.prepare(`SELECT property_id, count(*) n, sum(market='sale') n_sale, sum(market='rent') n_rent, min(price) min_price, max(price) max_price,
      max(bedrooms) bedrooms, max(bathrooms) bathrooms, max(area_sqft) area_sqft, max(year_built) year_built FROM listing GROUP BY property_id`).all() as Agg[];
    const up = db.prepare(`UPDATE property SET n_listings=@n, n_sale=@n_sale, n_rent=@n_rent, min_price=@min_price, max_price=@max_price,
      bedrooms=COALESCE(bedrooms,@bedrooms), bathrooms=COALESCE(bathrooms,@bathrooms), area_sqft=COALESCE(area_sqft,@area_sqft), year_built=COALESCE(year_built,@year_built),
      attr_quality = ((lat IS NOT NULL) + (COALESCE(bedrooms,@bedrooms) IS NOT NULL) + (COALESCE(bathrooms,@bathrooms) IS NOT NULL) + (COALESCE(area_sqft,@area_sqft) IS NOT NULL) + (COALESCE(year_built,@year_built) IS NOT NULL) + (role_value IS NOT NULL)) / 6.0
      WHERE id=@property_id`);
    db.transaction(() => { for (const a of aggs) up.run(a); })();
    log(`     ${aggs.length} propriétés agrégées`);
  }

  db.exec(`INSERT INTO place (kind, name, region, lat, lng, n_sale, n_rent)
    SELECT 'city', city, max(region), avg(lat), avg(lng), sum(market='sale'), sum(market='rent') FROM listing WHERE city <> '' AND lat IS NOT NULL GROUP BY city HAVING count(*) >= 3;
    INSERT INTO place (kind, name, region, lat, lng, n_sale, n_rent)
    SELECT 'region', region, region, avg(lat), avg(lng), sum(market='sale'), sum(market='rent') FROM listing WHERE region IS NOT NULL AND lat IS NOT NULL GROUP BY region;`);

  // Métriques de marché : province / région / ville × marché × catégorie
  const insMetric = db.prepare(`INSERT OR REPLACE INTO market_metric VALUES (@scope,@scope_value,@market,@category,@n,@median_price,@p25_price,@p75_price,@median_ppsf,@median_days,@new_30d,@median_ratio,@computed_at)`);
  const rows = db.prepare(`SELECT l.market, l.category, l.region, l.city, l.price, l.area_sqft, l.first_seen, v.ratio FROM listing l LEFT JOIN valuation v ON v.listing_uid = l.uid WHERE l.price > 0`).all() as Row[];
  const groups = new Map<string, { prices: number[]; ppsf: number[]; days: number[]; new30: number; ratios: number[] }>();
  const nowMs = Date.now();
  const add = (k: string, r: Row) => {
    let g = groups.get(k); if (!g) { g = { prices: [], ppsf: [], days: [], new30: 0, ratios: [] }; groups.set(k, g); }
    const p = r.price as number; g.prices.push(p);
    const a = num(r.area_sqft); if (a && a > 100 && (r.market === "sale")) g.ppsf.push(p / a);
    const fs = r.first_seen ? Date.parse(r.first_seen as string) : NaN; if (!Number.isNaN(fs)) { const d = (nowMs - fs) / 86400000; g.days.push(d); if (d <= 30) g.new30++; }
    const ra = num(r.ratio); if (ra) g.ratios.push(ra);
  };
  for (const r of rows) {
    const m = r.market as string, c = r.category as string;
    for (const cat of [c, "all"]) {
      add(`province|Québec|${m}|${cat}`, r);
      if (r.region) add(`region|${r.region}|${m}|${cat}`, r);
      if (r.city) add(`city|${r.city}|${m}|${cat}`, r);
    }
  }
  const txM = db.transaction(() => {
    for (const [k, g] of groups) {
      if (g.prices.length < 5) continue;
      const [scope, scope_value, market, category] = k.split("|");
      insMetric.run({ scope, scope_value, market, category, n: g.prices.length, median_price: median(g.prices), p25_price: quantile(g.prices, 0.25), p75_price: quantile(g.prices, 0.75),
        median_ppsf: g.ppsf.length >= 5 ? median(g.ppsf) : null, median_days: g.days.length ? Math.round(median(g.days) as number) : null, new_30d: g.new30,
        median_ratio: g.ratios.length >= 10 ? median(g.ratios) : null, computed_at: now });
    }
  });
  txM();

  log("4/6 index plein texte");
  db.exec(`CREATE VIRTUAL TABLE listing_fts USING fts5(uid UNINDEXED, address, city, title, tokenize="unicode61 remove_diacritics 2");
    INSERT INTO listing_fts SELECT uid, address, city, title FROM listing;
    CREATE VIRTUAL TABLE role_unit_fts USING fts5(id UNINDEXED, address, city, tokenize="unicode61 remove_diacritics 2");
    INSERT INTO role_unit_fts SELECT id, address, city FROM role_unit WHERE address IS NOT NULL;`);

  const stats = {
    built_at: now, listings_sale: nSale, listings_rent: nRent, properties: (db.prepare("SELECT count(*) c FROM property").get() as { c: number }).c,
    valuations: (db.prepare("SELECT count(*) c FROM valuation").get() as { c: number }).c, transactions: nTx, role_units: nUnits,
    multi_listing_properties: (db.prepare("SELECT count(*) c FROM property WHERE n_listings > 1").get() as { c: number }).c,
  };
  const insMeta = db.prepare("INSERT OR REPLACE INTO meta VALUES (?,?)");
  for (const [k, v] of Object.entries(stats)) insMeta.run(k, String(v));
  insMeta.run("vp_marche_stats", existsSync(path.join(SRC, "vp-marche-stats.json")) ? readFileSync(path.join(SRC, "vp-marche-stats.json"), "utf8") : "{}");
  rd.close();
  db.exec("DETACH ik; DETACH vp; DETACH lk;");
  db.exec("PRAGMA journal_mode = WAL;");
  db.exec("ANALYZE; VACUUM;");
  db.close();
  log(`   ${JSON.stringify(stats)}`);
  return stats;
}

function publish() {
  log("5/6 bascule atomique");
  const sz = statSync(NEW).size;
  if (sz < 50e6) die(`toveli.db.new trop petite (${sz} o)`);
  for (const suf of ["-wal", "-shm"]) rmSync(LIVE + suf, { force: true });
  renameSync(NEW, LIVE);
  log("6/6 redémarrage PM2");
  if (env.NO_RESTART !== "1") {
    try { execFileSync("pm2", ["restart", PM2_APP, "--update-env"], { stdio: "ignore", env: { ...env, PATH: `/opt/homebrew/bin:${env.PATH}` } }); }
    catch { log("avertissement : pm2 restart a échoué (pm2 absent ou app non lancée)"); }
  }
}

(function main() {
  log(`=== toveli sync — ${new Date().toISOString()} ===`);
  if (env.SKIP_FETCH !== "1") fetchAll(); else log("1/6 extraction sautée (SKIP_FETCH=1)");
  for (const f of ["immoka.db", "vraiprix.db", "louka.db"]) if (!existsSync(path.join(SRC, f))) die(`${f} absent dans data/src`);
  build();
  publish();
  log(`=== terminé en ${((Date.now() - T0) / 1000).toFixed(0)} s ===`);
})();