/**
 * 自社ネットワークの差分を毎日こちらへ取り込み、スコアと順位を入れ直す（2026-09-05）。
 *
 * ねらい: 姉妹サイトごとに「変わったところ」と「CAPLOGUE がどこまで読んだか」を
 * 持たせ、CAPLOGUE のほうは日次で勝手に更新されて順位が動くようにする。
 *
 * **このファイルは `/algorithm/source/` でそのまま公開している。**
 * コメントに人の名前を書かない（運営の名義は「きゃっぷ」だけ）。
 *
 * ── 何をしているか
 *
 * 1. `feed_cursor` を読む。相手ごとに「どこまで読んだか」（`cursor_at`）を持っている
 * 2. 相手の表から、**その時刻より後に中身の動いた行だけ**を引く
 *    （相手の `content_changed_at`。取込が毎回押す `last_seen` は見ない）
 * 3. `feed_point`（前回の計算に使った座標の写し）と突き合わせ、
 *    **「前はどこに在ったか → いまどこに在るか」**を1件ずつ確定する
 * 4. 動いた座標の周りの町丁目について、**その指標だけ**を差分で直す
 * 5. `rank_scale`（凍らせた物差し）でスコアにし直し、観点・世代・総合まで組み直す
 * 6. 動いた町丁目を持つ市区町村を畳み直し、**全国順位と都道府県内の順位を付け直す**
 * 7. `feed_cursor` を進め、`ingest_runs` に1行残す
 *
 * ── 測り直しを全部やらないための考え方（ここが肝）
 *
 * **「点」という言葉は使わない**（2026-09-06 に決めた）。
 * 地図の上のものは**座標**、測った結果は**スコア**と書く。
 *
 * 距離の指標は、**足された座標と消えた座標だけで直せる**。
 *
 *   座標が足された … 新しい距離 = min(いまの距離, その座標までの距離)
 *   座標が消えた … **その座標が最寄りだったときだけ**、探し直す
 *
 * 半径内の数はもっと単純で、半径に入った座標を ±1 するだけ。
 * **探し直しが要るのは「消えた座標が最寄りだった町丁目」だけ**で、
 * 全国でもふつう数十件しかない。ここを分けないと、
 * 1件の閉店のために数千の町丁目を測り直すことになる。
 *
 * ── やらないと決めたこと
 *
 * **毎晩まるごと計算し直さない。**220,811 町丁目 × 48 指標を毎晩やると
 * D1 の読み書きが日に数百万になり、しかも街に何も起きていない市区町村の順位まで
 * 毎日わずかに動く。物差し（`rank_scale`）を凍らせてあるので、
 * **動いた順位は街に起きたことだけを映す。**
 *
 * **相手の変更が多すぎる回は、差分で追わない。**版がまるごと入れ替わったとき
 * （医療情報ネットは半年ごとに20万件が一度に動く）は `state='full'` を立てて
 * 手元の全件組み直し（`algorithm/run_all.sh`）を待つ。
 * 中途半端に何日もかけて追うと、その間ずっと**新旧が混ざった順位**を出すことになる。
 */
import { type Env, now } from './types';
import {
  METRIC, FIELD, GEN, NEAR, COUNT, NEAR_MAX_M, FEED_RING_M, AID_AGE,
} from './spec.gen';

/** 1回に読む変更行の上限（相手1つあたり）。超えたら差分で追わない。 */
const MAX_ROWS = 500;

/** 1回に直す町丁目の上限。 */
const MAX_AREAS = 30000;

/** `feed_far` を1つの指標について読む上限。 */
const MAX_FAR = 30000;

/** 升目の大きさ。**`algorithm/geo.py` と同じ式でなければならない。** */
const M_PER_DEG = 111320.0;
const REF_LAT = 36.0;
const CELL_M = 2000.0;
const D_LAT = CELL_M / M_PER_DEG;
const D_LON = CELL_M / (M_PER_DEG * Math.cos((REF_LAT * Math.PI) / 180));

export function cellOf(lat: number, lng: number): number {
  return Math.floor(lat / D_LAT) * 100000 + Math.floor(lng / D_LON);
}

export function distM(a: number, b: number, c: number, d: number): number {
  const dy = (a - c) * M_PER_DEG;
  const dx = (b - d) * M_PER_DEG * Math.cos(((a + c) * 0.5 * Math.PI) / 180);
  return Math.sqrt(dy * dy + dx * dx);
}

/** 緯度経度の枠。`m` メートル四方に届く範囲。 */
function box(lat: number, lng: number, m: number): [number, number, number, number] {
  const dy = m / M_PER_DEG;
  const dx = m / (M_PER_DEG * Math.max(0.2, Math.cos((lat * Math.PI) / 180)));
  return [lat - dy, lat + dy, lng - dx, lng + dx];
}

/** その升目から `m` メートル以内に届く升目の並び（探し直しのときだけ使う）。 */
function cellsWithin(lat: number, lng: number, m: number): number[] {
  const step = Math.min(CELL_M, D_LON * M_PER_DEG * Math.cos((lat * Math.PI) / 180));
  const ring = Math.ceil(m / step) + 1;
  const cy = Math.floor(lat / D_LAT);
  const cx = Math.floor(lng / D_LON);
  const out: number[] = [];
  for (let y = cy - ring; y <= cy + ring; y++) {
    for (let x = cx - ring; x <= cx + ring; x++) out.push(y * 100000 + x);
  }
  return out;
}

type Row = Record<string, unknown>;

function num(v: unknown): number | null {
  return typeof v === 'number' && Number.isFinite(v) ? v : null;
}

export type Pt = {
  pid: string; kind: string; lat: number; lng: number;
  cell: number; how: string; src: string;
};

// ───────────────────────────────────────────── 出どころ

/**
 * 出どころごとの読み方。**ここに1つ足せば日次の巡回に入る。**
 *
 * `pts` が返すのは「その行がいま作るべき座標の一式」。空なら
 * 「その行はもう座標を持たない」（閉店・休止・緯度経度が消えた）ことを表す。
 */
type Source = {
  src: string;
  db: (env: Env) => D1Database;
  sql: string;
  key: (r: Row) => string;
  pts: (r: Row) => { kind: string; lat: number; lng: number }[];
  /** 座標を持たない行の置き場所（町丁目コード）。暮らしの店だけが使う。 */
  areaOf?: (r: Row) => { kind: string; code: string } | null;
};

const SOURCES: Source[] = [
  {
    src: 'iryo.facility',
    db: (e) => e.IRYO,
    sql: `SELECT f.id AS id, f.kind AS kind, f.lat AS lat, f.lng AS lng,
                 f.beds_total AS beds, f.content_changed_at AS at,
                 (SELECT COUNT(*) FROM facility_dept d
                   WHERE d.id = f.id AND d.dept_code = '03001') AS ped,
                 (SELECT COUNT(*) FROM facility_dept d
                   WHERE d.id = f.id AND d.dept_code = '01001') AS intern
            FROM facility f
           WHERE f.content_changed_at > ?1
           ORDER BY f.content_changed_at LIMIT ?2`,
    key: (r) => `iryo:${r.id as string}`,
    pts: (r) => {
      const lat = num(r.lat); const lng = num(r.lng);
      if (lat === null || lng === null) return [];
      const out = [{ kind: `med_${r.kind as string}`, lat, lng }];
      if (r.kind === 'hospital' && (num(r.beds) ?? 0) >= 100) {
        out.push({ kind: 'med_big', lat, lng });
      }
      if ((num(r.ped) ?? 0) > 0) out.push({ kind: 'dept_ped', lat, lng });
      if ((num(r.intern) ?? 0) > 0) out.push({ kind: 'dept_intern', lat, lng });
      return out;
    },
  },
  {
    src: 'gakko.school',
    db: (e) => e.GAKKO,
    sql: `SELECT id, kind, lat, lng, content_changed_at AS at FROM school
           WHERE content_changed_at > ?1
           ORDER BY content_changed_at LIMIT ?2`,
    key: (r) => `gakko:${r.id as string}`,
    pts: (r) => {
      const lat = num(r.lat); const lng = num(r.lng);
      if (lat === null || lng === null) return [];
      return [{ kind: `sch_${r.kind as string}`, lat, lng }];
    },
  },
  {
    src: 'kaigo.facility',
    db: (e) => e.KAIGO,
    sql: `SELECT jigyosho_no AS id, lat, lng, is_residential AS res,
                 status, content_changed_at AS at
            FROM facilities
           WHERE content_changed_at > ?1
           ORDER BY content_changed_at LIMIT ?2`,
    key: (r) => `kaigo:${r.id as string}`,
    pts: (r) => {
      const lat = num(r.lat); const lng = num(r.lng);
      // **出典から消えた事業所は座標にしない**（`status` が 'active' でなくなる）
      if (lat === null || lng === null || r.status !== 'active') return [];
      const out = [{ kind: 'care', lat, lng }];
      if (num(r.res)) out.push({ kind: 'care_res', lat, lng });
      return out;
    },
  },
  {
    src: 'shop.shop',
    db: (e) => e.DB,
    sql: `SELECT id, cat, lat, lng, area_cd, closed_on,
                 content_changed_at AS at
            FROM shop
           WHERE content_changed_at > ?1
           ORDER BY content_changed_at LIMIT ?2`,
    key: (r) => `shop:${r.id as string}`,
    pts: (r) => {
      // **閉店した店は数えない。**行そのものは残っている
      if (r.closed_on) return [];
      const lat = num(r.lat); const lng = num(r.lng);
      if (lat === null || lng === null) return [];
      return [{ kind: `shop_${r.cat as string}`, lat, lng }];
    },
    // 座標を持たない店は町丁目の重心に置く（**捨てると買い物のスコアが2割ぶん低く出る**）
    areaOf: (r) => (r.closed_on || num(r.lat) !== null || !r.area_cd ? null
      : { kind: `shop_${r.cat as string}`, code: r.area_cd as string }),
  },
];

// ───────────────────────────────────────────── 物差し

type Scale = { down: boolean; x: number[]; p: number[] };

/**
 * 凍らせた物差しで1つの値をスコアにする。
 * **`algorithm/score.py` の `apply()` と同じ式**（違うと手元と日次で別のスコアが出る）。
 */
export function applyScale(s: Scale | undefined, v: number | null): number | null {
  if (s === undefined || v === null || s.x.length === 0) return null;
  const x = s.down ? -v : v;
  if (x <= s.x[0]) return round2(s.p[0]);
  if (x >= s.x[s.x.length - 1]) return round2(s.p[s.p.length - 1]);
  let lo = 0; let hi = s.x.length - 1;
  while (lo < hi) {                       // bisect_left
    const mid = (lo + hi) >> 1;
    if (s.x[mid] < x) lo = mid + 1; else hi = mid;
  }
  if (s.x[lo] === x) return round2(s.p[lo]);
  const a = s.x[lo - 1]; const b = s.x[lo];
  const t = b === a ? 0 : (x - a) / (b - a);
  return round2(s.p[lo - 1] + (s.p[lo] - s.p[lo - 1]) * t);
}

const round2 = (v: number): number => Math.round(v * 100) / 100;

/** 使える項目だけで加重平均。返すのは `[スコア, 使えた重みの割合]`。 */
function weighted(
  vals: Record<string, number | null>, w: Record<string, number>,
): [number | null, number] {
  let s = 0; let t = 0; let all = 0;
  for (const [k, wk] of Object.entries(w)) {
    if (wk <= 0) continue;
    all += wk;
    const v = vals[k];
    if (v === null || v === undefined) continue;
    s += v * wk; t += wk;
  }
  if (t <= 0) return [null, 0];
  return [s / t, t / all];
}

/** 生の指標から、観点・世代・総合を組み立てる。**`score.py` と同じ順序。** */
export function scoreArea(
  raw: Record<string, number | null>, scales: Map<string, Scale>,
): Record<string, number | null> {
  const pct: Record<string, number | null> = {};
  for (const m of METRIC) pct[m.key] = applyScale(scales.get(m.key), raw[m.key] ?? null);
  const fields: Record<string, number | null> = {};
  for (const f of FIELD) {
    const [v] = weighted(pct, f.parts);
    fields[f.key] = v === null ? null : round2(v);
  }
  const out: Record<string, number | null> = {};
  for (const f of FIELD) out[`f_${f.key}`] = fields[f.key];
  const got: number[] = [];
  for (const g of GEN) {
    const [v, cov] = weighted(fields, g.w);
    out[`s_${g.key}`] = v === null ? null : round2(v);
    out[`c_${g.key}`] = Math.round(cov * 1000) / 1000;
    if (v !== null) got.push(v);
  }
  out.s_all = got.length === GEN.length
    ? round2(got.reduce((a, b) => a + b, 0) / got.length) : null;
  return out;
}

// ───────────────────────────────────────────── D1 の小道具

async function all<T = Row>(db: D1Database, sql: string, ...b: unknown[]): Promise<T[]> {
  const r = await db.prepare(sql).bind(...b).all<T>();
  return (r.results ?? []) as T[];
}

async function one<T = Row>(db: D1Database, sql: string, ...b: unknown[]): Promise<T | null> {
  return (await db.prepare(sql).bind(...b).first<T>()) ?? null;
}

/** `IN (?,?,…)` を組む。**外から来た文字を SQL に混ぜないため、数だけ数える。** */
const marks = (n: number): string => new Array(n).fill('?').join(',');

function chunks<T>(xs: T[], n: number): T[][] {
  const out: T[][] = [];
  for (let i = 0; i < xs.length; i += n) out.push(xs.slice(i, i + n));
  return out;
}

// ───────────────────────────────────────────── 本体

/** 町丁目1つぶんの、指標ごとの直し方。 */
type Fix = {
  /** 足された座標までの、いちばん近い距離（m）。 */
  cand: number | null;
  /** 消えた座標までの、いちばん近い距離（m）。 */
  gone: number | null;
  /** 半径内の数の増減。 */
  delta: number;
  /** 座標で測るのではなく、値をそのまま入れ替えるもの（子ども医療費助成）。 */
  set?: number | null;
};

export type FeedResult = {
  ok: boolean;
  changed: Record<string, number>;
  areas: number;
  cities: number;
  research: number;
  note: string;
};

/**
 * 日次の入口。**走ったことを必ず1行残す**（`ingest_runs.kind='feed'`）。
 * 落ちても残す（残らないと、走っていないのか落ちたのかが分からない）。
 */
export async function feed(env: Env): Promise<void> {
  const started = now();
  let status = 'error';
  let note = '';
  try {
    const r = await runFeed(env);
    const n = Object.values(r.changed).reduce((a, b) => a + b, 0);
    status = !r.ok ? 'held' : (r.areas > 0 ? 'ok' : 'nochange');
    note = [
      `変更行 ${n}`,
      ...Object.entries(r.changed).filter(([, v]) => v > 0)
        .map(([k, v]) => `${k} ${v}`),
      `町丁目 ${r.areas}`, `市区町村 ${r.cities}`, `探し直し ${r.research}`,
      r.note,
    ].filter(Boolean).join(' / ');
  } catch (e) {
    note = String(e).slice(0, 400);
  }
  await env.DB.prepare(
    'INSERT INTO ingest_runs (kind,started_at,finished_at,status,note)'
    + ' VALUES (?,?,?,?,?)',
  ).bind('feed', started, now(), status, note).run();
}

export async function runFeed(env: Env): Promise<FeedResult> {
  const started = now();
  const cur = await all(env.DB,
    'SELECT src,label,kind,cursor_at,state FROM feed_cursor ORDER BY src');
  const byCur = new Map(cur.map((c) => [c.src as string, c]));

  const changed: Record<string, number> = {};
  const notes: string[] = [];
  const advance = new Map<string, string>();
  // `pid kind` → いまの位置。読んだ行が作らなかった組み合わせは消えたとみなす
  const made = new Map<string, Pt>();
  const readPids = new Set<string>();

  for (const s of SOURCES) {
    const c = byCur.get(s.src);
    if (!c) continue;
    if (c.state === 'full') {
      notes.push(`${s.src}: 全件の組み直し待ち`);
      continue;
    }
    const from = (c.cursor_at as string | null) ?? '';
    let rows: Row[];
    try {
      rows = await all(s.db(env), s.sql, from, MAX_ROWS + 1);
    } catch (e) {
      // **読めなかったを「変わっていない」と読み替えない。**印を残して次へ
      await env.DB.prepare(
        "UPDATE feed_cursor SET state='held', checked_at=?2, note=?3 WHERE src=?1")
        .bind(s.src, started, String(e).slice(0, 200)).run();
      notes.push(`${s.src}: 読めなかった`);
      continue;
    }
    if (rows.length > MAX_ROWS) {
      await env.DB.prepare(
        "UPDATE feed_cursor SET state='full', checked_at=?2, note=?3 WHERE src=?1")
        .bind(s.src, started,
          `1回に読む上限（${MAX_ROWS}件）を超えて動いた。手元で組み直す`).run();
      notes.push(`${s.src}: 多すぎる（全件の組み直しへ）`);
      continue;
    }
    changed[s.src] = rows.length;
    if (rows.length === 0) {
      await env.DB.prepare(
        "UPDATE feed_cursor SET checked_at=?2, n_seen=0, state='ok', note=NULL"
        + ' WHERE src=?1').bind(s.src, started).run();
      continue;
    }
    advance.set(s.src, rows[rows.length - 1].at as string);

    // 座標を持たない行の置き場所（町丁目の重心）をまとめて引く
    const want = new Map<string, { kind: string; code: string }>();
    if (s.areaOf) {
      for (const r of rows) {
        const a = s.areaOf(r);
        if (a) want.set(s.key(r), a);
      }
    }
    const centroid = new Map<string, [number, number]>();
    for (const part of chunks([...new Set([...want.values()].map((v) => v.code))], 90)) {
      for (const a of await all(env.DB,
        `SELECT code,lat,lng FROM rank_area WHERE code IN (${marks(part.length)})`,
        ...part)) {
        const la = num(a.lat); const ln = num(a.lng);
        if (la !== null && ln !== null) centroid.set(a.code as string, [la, ln]);
      }
    }

    for (const r of rows) {
      const pid = s.key(r);
      readPids.add(pid);
      for (const p of s.pts(r)) {
        made.set(`${pid} ${p.kind}`, {
          pid, kind: p.kind, lat: p.lat, lng: p.lng,
          cell: cellOf(p.lat, p.lng), how: 'geo', src: s.src.split('.')[0],
        });
      }
      const a = want.get(pid);
      const c2 = a ? centroid.get(a.code) : undefined;
      if (a && c2) {
        made.set(`${pid} ${a.kind}`, {
          pid, kind: a.kind, lat: c2[0], lng: c2[1],
          cell: cellOf(c2[0], c2[1]), how: 'area', src: s.src.split('.')[0],
        });
      }
    }
  }

  // ── 子ども医療費助成。**座標ではなく市区町村ごとの制度**なので、
  //    変わった市区町村の町丁目を全部書き換える（1つの市で数百〜千件）
  const aidCities = new Map<string, number | null>();
  {
    const c = byCur.get('gakko.med_aid');
    if (c && c.state !== 'full') {
      try {
        const rows = await all(env.GAKKO,
          'SELECT city, age_out, content_changed_at AS at FROM med_aid'
          + ' WHERE content_changed_at > ?1 ORDER BY content_changed_at LIMIT ?2',
          (c.cursor_at as string | null) ?? '', MAX_ROWS + 1);
        if (rows.length > MAX_ROWS) {
          await env.DB.prepare(
            "UPDATE feed_cursor SET state='full', checked_at=?2, note=?3"
            + ' WHERE src=?1')
            .bind('gakko.med_aid', started,
              `1回に読む上限（${MAX_ROWS}件）を超えて動いた。手元で組み直す`).run();
          notes.push('学びの窓口 医療費助成: 多すぎる（全件の組み直しへ）');
        } else {
          changed['gakko.med_aid'] = rows.length;
          if (rows.length > 0) {
            advance.set('gakko.med_aid', rows[rows.length - 1].at as string);
            for (const r of rows) {
              // **知らない書き方を年齢に当てはめない。**空にする
              const a = AID_AGE[((r.age_out as string) ?? '').trim()];
              aidCities.set(r.city as string, a === undefined ? null : a);
            }
          } else {
            await env.DB.prepare(
              "UPDATE feed_cursor SET checked_at=?2, n_seen=0, state='ok',"
              + ' note=NULL WHERE src=?1').bind('gakko.med_aid', started).run();
          }
        }
      } catch (e) {
        await env.DB.prepare(
          "UPDATE feed_cursor SET state='held', checked_at=?2, note=?3"
          + ' WHERE src=?1')
          .bind('gakko.med_aid', started, String(e).slice(0, 200)).run();
        notes.push('学びの窓口 医療費助成: 読めなかった');
      }
    }
  }

  // ── LAYER-0（町丁目そのものと、地形・地価・災害・治安・駅）。
  //    **ここは差分で追えない。**座標の出入りではなく町丁目の列そのものが
  //    書き換わるうえ、出典が年に1回まとめて版を上げる。
  //    だから「動いたことに気づいて全件の組み直しを促す」までをやる。
  //    気づく仕組みだけ作って反映されない、をやらないための印
  {
    const c = byCur.get('layer0.area');
    if (c && c.state !== 'full') {
      try {
        const rows = await all(env.LAYER0,
          'SELECT key,title,COALESCE(changed_at,fetched) AS at FROM jp_source'
          + ' WHERE COALESCE(changed_at,fetched) > ?1'
          + ' ORDER BY at DESC LIMIT 20',
          (c.cursor_at as string | null) ?? '');
        if (rows.length > 0) {
          const names = rows.map((r) => r.title as string).slice(0, 3).join('・');
          await env.DB.prepare(
            "UPDATE feed_cursor SET state='full', checked_at=?2, n_seen=?3, note=?4"
            + ' WHERE src=?1')
            .bind('layer0.area', started, rows.length,
              `${names}${rows.length > 3 ? ' ほか' : ''}が新しくなった。`
              + '町丁目の列は差分で追えないので手元で組み直す').run();
          notes.push(`ゼロトゥワン LAYER-0: ${rows.length}件の出典が動いた（全件の組み直しへ）`);
        } else {
          await env.DB.prepare(
            "UPDATE feed_cursor SET checked_at=?2, n_seen=0, state='ok',"
            + ' note=NULL WHERE src=?1').bind('layer0.area', started).run();
        }
      } catch (e) {
        await env.DB.prepare(
          "UPDATE feed_cursor SET state='held', checked_at=?2, note=?3"
          + ' WHERE src=?1')
          .bind('layer0.area', started, String(e).slice(0, 200)).run();
        notes.push('ゼロトゥワン LAYER-0: 読めなかった');
      }
    }
  }

  // ── 前の位置を引く
  const before = new Map<string, Pt>();
  for (const part of chunks([...readPids], 90)) {
    for (const p of await all<Pt>(env.DB,
      `SELECT pid,kind,lat,lng,cell,how,src FROM feed_point
        WHERE pid IN (${marks(part.length)})`, ...part)) {
      before.set(`${p.pid} ${p.kind}`, p);
    }
  }

  // ── 「前 → 後」を確定する
  const adds: Pt[] = [];
  const gones: Pt[] = [];
  for (const [k, p] of made) {
    const b = before.get(k);
    if (b && b.lat === p.lat && b.lng === p.lng) continue;
    if (b) gones.push(b);
    adds.push(p);
  }
  for (const [k, b] of before) {
    if (!made.has(k)) gones.push(b);
  }

  if (adds.length === 0 && gones.length === 0 && aidCities.size === 0) {
    await advanceAll(env, advance, changed, started);
    return {
      ok: true, changed, areas: 0, cities: 0, research: 0,
      note: notes.join(' / ') || '動いた座標なし',
    };
  }

  // ── 種別 → その種別が効く指標
  const nearOf = new Map<string, string[]>();
  const countOfKind = new Map<string, { key: string; m: number }[]>();
  for (const [mk, kinds] of Object.entries(NEAR)) {
    for (const k of kinds) nearOf.set(k, [...(nearOf.get(k) ?? []), mk]);
  }
  for (const c of COUNT) {
    for (const k of c.kinds) {
      countOfKind.set(k, [...(countOfKind.get(k) ?? []), { key: c.key, m: c.m }]);
    }
  }

  // ── 影響を受ける町丁目と、指標ごとの直し方を積む
  const fixes = new Map<string, Map<string, Fix>>();
  const pos = new Map<string, [number, number]>();
  const put = (code: string, metric: string, f: (x: Fix) => void): void => {
    let m = fixes.get(code);
    if (!m) { m = new Map(); fixes.set(code, m); }
    let x = m.get(metric);
    if (!x) { x = { cand: null, gone: null, delta: 0 }; m.set(metric, x); }
    f(x);
  };

  // `feed_far` は指標ごとに1回だけ読む（動いた座標の数だけ読み直さない）。
  // **緯度で並べて持ち、必要な帯だけ見る。**並べずに毎回なめると、
  // 動いた座標 1,000 × 遠い町丁目 30,000 で3千万回の距離計算になる
  type Far = { code: string; lat: number; lng: number };
  const farCache = new Map<string, Far[]>();
  const farOf = async (metric: string): Promise<Far[]> => {
    let v = farCache.get(metric);
    if (v === undefined) {
      const rows = await all(env.DB,
        'SELECT code,lat,lng FROM feed_far WHERE metric=?1 LIMIT ?2',
        metric, MAX_FAR);
      if (rows.length >= MAX_FAR) notes.push(`${metric}: 遠い町丁目が多すぎる`);
      v = rows.map((r) => ({
        code: r.code as string, lat: r.lat as number, lng: r.lng as number,
      })).sort((a, b) => a.lat - b.lat);
      farCache.set(metric, v);
    }
    return v;
  };
  /** 緯度 `lat` から `m` メートルの帯に入る遠い町丁目だけを返す。 */
  const farBand = (xs: Far[], lat: number, m: number): Far[] => {
    const dy = m / M_PER_DEG;
    let lo = 0; let hi = xs.length;
    while (lo < hi) {                      // 下端を二分探索
      const mid = (lo + hi) >> 1;
      if (xs[mid].lat < lat - dy) lo = mid + 1; else hi = mid;
    }
    let end = lo;
    while (end < xs.length && xs[end].lat <= lat + dy) end++;
    return xs.slice(lo, end);
  };

  // 助成年齢が変わった市区町村は、その町丁目を全部入れ替える
  for (const [city, age] of aidCities) {
    for (const a of await all(env.DB,
      'SELECT code,lat,lng FROM rank_area WHERE city_cd=?1', city)) {
      const code = a.code as string;
      const la = num(a.lat); const ln = num(a.lng);
      if (la !== null && ln !== null) pos.set(code, [la, ln]);
      put(code, 'med_aid_age', (x) => { x.set = age; });
    }
  }

  let capped = false;
  for (const [isAdd, p] of [
    ...adds.map((x) => [true, x] as [boolean, Pt]),
    ...gones.map((x) => [false, x] as [boolean, Pt]),
  ]) {
    const nears = nearOf.get(p.kind) ?? [];
    const cnts = countOfKind.get(p.kind) ?? [];
    if (nears.length === 0 && cnts.length === 0) continue;
    if (fixes.size >= MAX_AREAS) { capped = true; break; }

    // 半径内の数は 3km まで。距離は 30km まで（それより遠いのは `feed_far`）
    const reach = Math.max(
      nears.length ? FEED_RING_M : 0,
      cnts.reduce((a, c) => Math.max(a, c.m), 0),
    );
    const [y0, y1, x0, x1] = box(p.lat, p.lng, reach);
    const near = await all(env.DB,
      `SELECT code,lat,lng FROM rank_area
        WHERE lat BETWEEN ?1 AND ?2 AND lng BETWEEN ?3 AND ?4`, y0, y1, x0, x1);
    for (const a of near) {
      const la = num(a.lat); const ln = num(a.lng);
      if (la === null || ln === null) continue;
      const code = a.code as string;
      const d = distM(la, ln, p.lat, p.lng);
      pos.set(code, [la, ln]);
      if (d <= FEED_RING_M) {
        for (const mk of nears) {
          put(code, mk, (x) => {
            if (isAdd) x.cand = x.cand === null ? d : Math.min(x.cand, d);
            else x.gone = x.gone === null ? d : Math.min(x.gone, d);
          });
        }
      }
      for (const c of cnts) {
        if (d <= c.m) put(code, c.key, (x) => { x.delta += isAdd ? 1 : -1; });
      }
    }

    // 升目では届かない町丁目（最寄りが30kmより遠い／測れていない）
    for (const mk of nears) {
      for (const f of farBand(await farOf(mk), p.lat, NEAR_MAX_M)) {
        const d = distM(f.lat, f.lng, p.lat, p.lng);
        if (d > NEAR_MAX_M) continue;
        const { code } = f;
        pos.set(code, [f.lat, f.lng]);
        put(code, mk, (x) => {
          if (isAdd) x.cand = x.cand === null ? d : Math.min(x.cand, d);
          else x.gone = x.gone === null ? d : Math.min(x.gone, d);
        });
      }
    }
  }

  // ── 座標の写しを入れ直す（探し直しはこの写しを読むので、先に済ませる）
  //
  // **ここから先で落ちたら、写しを前の位置に戻してから投げ直す。**
  // 戻さないと、次の回は「前とおなじ」に見えて差分が0件になり、
  // **町丁目を直さないまま印だけが進んだ**状態で静かに固まる。
  // かといって写しを後回しにすると、途中で落ちた回の半径内の数を
  // 次の回がもう一度足してしまう（距離と違って足し算は繰り返せない）。
  const put_ = async (ps: Pt[]): Promise<void> => {
    for (const part of chunks(ps, 40)) {
      await env.DB.batch(part.map((p) => env.DB.prepare(
        `INSERT INTO feed_point (pid,kind,lat,lng,cell,how,src)
         VALUES (?1,?2,?3,?4,?5,?6,?7)
         ON CONFLICT(pid,kind) DO UPDATE SET
           lat=excluded.lat, lng=excluded.lng, cell=excluded.cell,
           how=excluded.how, src=excluded.src`)
        .bind(p.pid, p.kind, p.lat, p.lng, p.cell, p.how, p.src)));
    }
  };
  const del_ = async (ps: { pid: string; kind: string }[]): Promise<void> => {
    for (const part of chunks(ps, 40)) {
      await env.DB.batch(part.map((p) => env.DB
        .prepare('DELETE FROM feed_point WHERE pid=?1 AND kind=?2')
        .bind(p.pid, p.kind)));
    }
  };
  const stillThere = new Set(adds.map((p) => `${p.pid} ${p.kind}`));
  const drop = gones.filter((p) => !stillThere.has(`${p.pid} ${p.kind}`));
  await put_(adds);
  await del_(drop);
  const undo = async (): Promise<void> => {
    // 前に在った位置へ戻す。前が無かったもの（新しく足した座標）は消す
    await put_(gones);
    await del_(adds.filter((p) => !before.has(`${p.pid} ${p.kind}`)));
  };
  try {
    return await rescore(env, fixes, pos, changed, advance, notes,
      started, capped);
  } catch (e) {
    await undo();
    throw e;
  }
}

/** 直すぶんが決まったあとの入れ直し。**落ちたら座標の写しを戻せるように分けてある。** */
async function rescore(
  env: Env,
  fixes: Map<string, Map<string, Fix>>,
  pos: Map<string, [number, number]>,
  changed: Record<string, number>,
  advance: Map<string, string>,
  notes: string[],
  started: string,
  capped: boolean,
): Promise<FeedResult> {

  // ── 物差し
  const scales = new Map<string, Scale>();
  for (const r of await all(env.DB, 'SELECT metric,down,x,p FROM rank_scale')) {
    scales.set(r.metric as string, {
      down: (r.down as number) === 1,
      x: JSON.parse(r.x as string) as number[],
      p: JSON.parse(r.p as string) as number[],
    });
  }
  if (scales.size === 0) {
    return {
      ok: false, changed, areas: 0, cities: 0, research: 0, note: '物差しが無い',
    };
  }

  // ── 町丁目を直す
  const codes = [...fixes.keys()];
  const cityHit = new Set<string>();
  let nArea = 0; let nSearch = 0;
  const farFix: { code: string; metric: string }[] = [];
  const farDel: { code: string; metric: string }[] = [];
  for (const part of chunks(codes, 90)) {
    const rows = await all(env.DB,
      `SELECT code,city_cd,lat,lng,raw FROM rank_area
        WHERE code IN (${marks(part.length)})`, ...part);
    const ups: D1PreparedStatement[] = [];
    for (const a of rows) {
      const code = a.code as string;
      const lat = num(a.lat); const lng = num(a.lng);
      if (lat === null || lng === null) continue;
      const raw = JSON.parse((a.raw as string) || '{}') as Record<string, number | null>;
      let dirty = false;
      for (const [metric, fx] of fixes.get(code) ?? []) {
        const old = raw[metric] ?? null;
        let next: number | null;
        if (fx.set !== undefined) {
          next = fx.set;
        } else if (metric in NEAR) {
          // **消えた座標が最寄りだったときだけ探し直す。**それ以外は足し算で足りる
          const wasNearest = fx.gone !== null && old !== null
            && Math.abs(fx.gone - old) <= 1;
          if (wasNearest && (fx.cand === null || fx.cand > old)) {
            next = await nearestOf(env, lat, lng, NEAR[metric]);
            nSearch++;
          } else if (fx.cand === null) {
            next = old;
          } else if (fx.cand > NEAR_MAX_M) {
            next = old;
          } else {
            next = old === null ? Math.round(fx.cand) : Math.min(old, Math.round(fx.cand));
          }
        } else {
          const n = (old ?? 0) + fx.delta;
          next = n < 0 ? 0 : n;
        }
        if ((old ?? null) !== (next ?? null)) { raw[metric] = next; dirty = true; }
        // 30km より遠いままの組み合わせは控えを直す。
        // **近くなったら控えから外す**（残すと毎回よけいに突き合わせる）
        if (metric in NEAR) {
          if (next === null || next > FEED_RING_M) farFix.push({ code, metric });
          else farDel.push({ code, metric });
        }
      }
      if (!dirty) continue;
      const sc = scoreArea(raw, scales);
      const cols = Object.keys(sc);
      const set = cols.map((k, i) => `${k}=?${i + 2}`).join(',');
      ups.push(env.DB.prepare(
        `UPDATE rank_area SET ${set}, raw=?${cols.length + 2} WHERE code=?1`)
        .bind(code, ...cols.map((k) => sc[k]),
          JSON.stringify(Object.fromEntries(
            Object.entries(raw).filter(([, v]) => v !== null && v !== undefined)))));
      cityHit.add(a.city_cd as string);
      nArea++;
    }
    for (const b of chunks(ups, 20)) await env.DB.batch(b);
  }

  // ── 遠い組み合わせの控えを直す。**ここを直さないと次の回から見えなくなる**
  //
  // 重心を持っていない控えは書かない。**0,0 で埋めると、
  // 赤道の沖にある町丁目として毎回すべての座標に当たりに行く。**
  for (const part of chunks(farFix.filter((f) => pos.has(f.code)), 30)) {
    await env.DB.batch(part.map((f) => {
      const p2 = pos.get(f.code)!;
      return env.DB.prepare(
        `INSERT INTO feed_far (code,metric,lat,lng,d)
         VALUES (?1,?2,?3,?4,(SELECT json_extract(raw,'$.' || ?2) FROM rank_area
                               WHERE code=?1))
         ON CONFLICT(code,metric) DO UPDATE SET d=excluded.d`)
        .bind(f.code, f.metric, p2[0], p2[1]);
    }));
  }
  for (const part of chunks(farDel, 40)) {
    await env.DB.batch(part.map((f) => env.DB
      .prepare('DELETE FROM feed_far WHERE code=?1 AND metric=?2')
      .bind(f.code, f.metric)));
  }

  // ── 市区町村を畳み直し、順位を付け直す
  for (const city of cityHit) await rollupCity(env, city);
  if (cityHit.size > 0) await reRank(env);

  await advanceAll(env, advance, changed, started);
  if (capped) notes.push(`町丁目が ${MAX_AREAS} を超えたので残りは次の回`);

  return {
    ok: true, changed, areas: nArea, cities: cityHit.size, research: nSearch,
    note: notes.join(' / '),
  };
}

async function advanceAll(
  env: Env, advance: Map<string, string>,
  changed: Record<string, number>, at: string,
): Promise<void> {
  for (const [src, cursor] of advance) {
    await env.DB.prepare(
      "UPDATE feed_cursor SET cursor_at=?2, checked_at=?3, n_seen=?4,"
      + " n_total=n_total+?4, state='ok', note=NULL WHERE src=?1")
      .bind(src, cursor, at, changed[src] ?? 0).run();
  }
}

/** 最寄りまでの距離（m）。**`geo.py` の `Index.nearest()` と同じ考え方。** */
async function nearestOf(
  env: Env, lat: number, lng: number, kinds: string[],
): Promise<number | null> {
  for (const r of [4000, 12000, NEAR_MAX_M]) {
    const cells = cellsWithin(lat, lng, r);
    let best: number | null = null;
    for (const part of chunks(cells, 80)) {
      for (const p of await all(env.DB,
        `SELECT lat,lng FROM feed_point
          WHERE kind IN (${marks(kinds.length)}) AND cell IN (${marks(part.length)})`,
        ...kinds, ...part)) {
        const d = distM(lat, lng, p.lat as number, p.lng as number);
        if (best === null || d < best) best = d;
      }
    }
    // **見つからなかったを大きい数にすり替えない。**上限まで見て無ければ `null`
    if (best !== null && best <= r) return Math.round(best);
  }
  return null;
}

/**
 * 市区町村を町丁目から畳み直す。**人口で重み付けする**（`score.py` と同じ）。
 * 面積では割らない。
 */
async function rollupCity(env: Env, city: string): Promise<void> {
  const keys = [...FIELD.map((f) => `f_${f.key}`),
    ...GEN.map((g) => `s_${g.key}`), 's_all'];
  const cov = GEN.map((g) => `c_${g.key}`);
  const sel = [
    ...keys.map((k) => `SUM(CASE WHEN ${k} IS NOT NULL AND pop>0 THEN ${k}*pop END) AS n_${k}`),
    ...keys.map((k) => `SUM(CASE WHEN ${k} IS NOT NULL AND pop>0 THEN pop END) AS w_${k}`),
    ...cov.map((k) => `SUM(CASE WHEN s_all IS NOT NULL AND pop>0 THEN ${k}*pop END) AS n_${k}`),
    'COUNT(*) AS areas',
    'SUM(CASE WHEN s_all IS NOT NULL AND pop>0 THEN 1 ELSE 0 END) AS scored',
    'SUM(COALESCE(pop,0)) AS pop',
    'SUM(CASE WHEN s_all IS NOT NULL AND pop>0 THEN pop ELSE 0 END) AS pop_scored',
    'SUM(COALESCE(area_m2,0)) AS m2',
  ].join(',');
  const r = await one(env.DB, `SELECT ${sel} FROM rank_area WHERE city_cd=?1`, city);
  if (!r) return;
  const set: string[] = []; const vals: unknown[] = [];
  const put = (col: string, v: unknown): void => {
    set.push(`${col}=?${set.length + 2}`); vals.push(v);
  };
  for (const k of keys) {
    const n = num(r[`n_${k}`]); const w = num(r[`w_${k}`]);
    put(k, n !== null && w !== null && w > 0 ? Math.round((n / w) * 100) / 100 : null);
  }
  const ps = num(r.pop_scored) ?? 0;
  for (const k of cov) {
    const n = num(r[`n_${k}`]);
    put(k, n !== null && ps > 0 ? Math.round((n / ps) * 1000) / 1000 : null);
  }
  const pop = num(r.pop) ?? 0;
  put('areas', num(r.areas) ?? 0);
  put('scored', num(r.scored) ?? 0);
  put('pop', pop);
  put('pop_scored', ps);
  put('pop_cover', pop > 0 ? Math.round((ps / pop) * 1000) / 1000 : null);
  put('area_km2', Math.round(((num(r.m2) ?? 0) / 1_000_000) * 100) / 100);
  await env.DB.prepare(`UPDATE rank_city SET ${set.join(',')} WHERE city_cd=?1`)
    .bind(city, ...vals).run();
}

/** 全国順位と都道府県内の順位を付け直す。**1,896 行しかないので毎回まるごと。** */
async function reRank(env: Env): Promise<void> {
  for (const g of [...GEN.map((x) => x.key), 'all']) {
    const rows = await all(env.DB,
      `SELECT city_cd,pref_cd FROM rank_city
        WHERE s_${g} IS NOT NULL ORDER BY s_${g} DESC`);
    const perPref = new Map<string, number>();
    const ups: D1PreparedStatement[] = [];
    rows.forEach((r, i) => {
      const p = (perPref.get(r.pref_cd as string) ?? 0) + 1;
      perPref.set(r.pref_cd as string, p);
      ups.push(env.DB.prepare(
        `UPDATE rank_city SET r_${g}=?2, rp_${g}=?3 WHERE city_cd=?1`)
        .bind(r.city_cd, i + 1, p));
    });
    for (const b of chunks(ups, 50)) await env.DB.batch(b);
  }
}