ukmesh/viewshed-worker/worker.py

"""
MeshCore Analytics — Viewshed Worker
Consumes jobs from Redis, downloads SRTM1 tiles, computes a raycasting
viewshed, clips to the UK mainland, stores the result polygon in
node_coverage, then notifies the frontend.
"""

import argparse
import hashlib
import json
import logging
import math
import multiprocessing
import os
import random
import resource
import tempfile
import threading
import time
import datetime as dt
from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable, Optional

import numpy as np
import psycopg2
import requests
import shapely
from scipy.ndimage import minimum_filter as _min_filter
from scipy.spatial import cKDTree
import redis
from psycopg2 import sql
from osgeo import gdal
from shapely.geometry import mapping, shape as shapely_shape, Polygon as ShapelyPolygon
from shapely.ops import unary_union
from rf.config import (
    ANTENNA_HEIGHT_M,
    CALIBRATION_EXTRA_MARGIN_DB,
    CALIBRATION_MAX_THRESHOLD_BOOST_DB,
    CALIBRATION_MIN_LINKS,
    CALIBRATION_MIN_OBSERVED_COUNT,
    CALIBRATION_PERCENTILE,
    CALIBRATION_REFRESH_S,
    COVERAGE_TARGET_HEIGHT_M,
    DEFAULT_USABLE_PATH_LOSS_DB,
    FREQ_MHZ,
    K_FACTOR,
    LAMBDA_M,
    LINK_LOS_MAX_V,
    PROFILE_STEP_M,
    RADIO_NEIGHBOR_SYNC,
    RF_CALIBRATION,
    R_EARTH_M,
    current_signal_thresholds_db,
    current_usable_path_loss_db,
    radio_snr_band,
)
from rf.loss import compute_path_loss, compute_prefix_path_losses
from rf.terrain import (
    InvalidTerrainRequest,
    PermanentOutOfScope,
    RetryableTerrainError,
    build_vrt,
    build_link_vrt,
    ensure_tiles,
    ensure_tiles_for_link,
    ensure_tiles_for_radius,
    load_uk_mainland,
    raster_window_for_radius,
    radio_horizon_m,
    sample_elevation,
)
import link_queue_v3
import viewshed_queue_v2
from worker_metrics import heartbeat, record_job, start_worker_metrics

gdal.UseExceptions()

logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] %(levelname)s %(message)s',
    datefmt='%H:%M:%S',
)
log = logging.getLogger(__name__)


@dataclass(frozen=True)
class Computed:
    geom: dict
    strength_geoms: dict[str, dict]
    radius_m: float
    elevation_m: float
    stage_seconds: dict[str, float] = field(default_factory=dict)
    peak_rss_mb: float = 0.0


class InvalidJob(ValueError):
    pass


class Cancelled(RuntimeError):
    pass


class JobDeadlineExceeded(RetryableTerrainError):
    pass

# ── Config ────────────────────────────────────────────────────────────────────

def env_flag_enabled(name: str, default: bool = True) -> bool:
    value = os.environ.get(name)
    if value is None or value.strip() == '':
        return default
    return value.strip().lower() not in ('0', 'false', 'no', 'off', 'disabled')


SRTM_DIR     = Path(os.environ.get('SRTM_DIR', '/data/srtm'))
REDIS_URL    = os.environ.get('REDIS_URL', 'redis://redis:6379')
REDIS_PASSWORD = os.environ.get('REDIS_PASSWORD') or None
DATABASE_URL = os.environ.get('DATABASE_URL')
RADIO_BOT_URL = os.environ.get('RADIO_BOT_URL', '').strip()
WORKER_MODE  = os.environ.get('WORKER_MODE', 'all').lower()
VIEWSHED_ENABLED = env_flag_enabled('VIEWSHED_ENABLED', False)
COVERAGE_MODEL = os.environ.get('COVERAGE_MODEL', 'rf_radial_100m').lower()
DB_APPLICATION_NAME = os.environ.get(
    'DATABASE_APPLICATION_NAME',
    f'meshcore-{WORKER_MODE if WORKER_MODE in ("viewshed", "link") else "viewshed-worker"}',
)

JOB_QUEUE      = 'meshcore:viewshed_jobs'
JOB_PENDING_SET = 'meshcore:viewshed_pending'
JOB_PAYLOADS = 'meshcore:viewshed:payloads'
JOB_COUNTERS = 'meshcore:viewshed:counters'
LINK_JOB_QUEUE = 'meshcore:link_jobs'
LIVE_CHANNEL   = 'meshcore:live'
VIEWSHED_WORKER_HEARTBEAT = 'meshcore:viewshed:worker_heartbeat'
PLANNED_RESULT_TTL_S = max(
    3600,
    min(7 * 24 * 3600, int(os.environ.get('PLANNED_COVERAGE_TTL_SECONDS', '86400'))),
)
PLANNED_COVERAGE_QUEUE_MAX = max(
    1,
    min(1000, int(os.environ.get('PLANNED_COVERAGE_QUEUE_MAX', '100'))),
)

COVERAGE_MODEL_VERSION = int(os.environ.get(
    'COVERAGE_MODEL_VERSION',
    '7' if COVERAGE_MODEL == 'rf_radial_100m' else '2',
))
COVERAGE_JOB_DEADLINE_SECONDS = max(
    60,
    min(3600, int(os.environ.get('COVERAGE_JOB_DEADLINE_SECONDS', '600'))),
)
RF_PREFIX_ENDPOINT_BATCH = max(
    1,
    min(512, int(os.environ.get('RF_PREFIX_ENDPOINT_BATCH', '64'))),
)
RF_PREFIX_RAY_BATCH = max(
    1,
    min(64, int(os.environ.get('RF_PREFIX_RAY_BATCH', '8'))),
)
MIN_LINK_OBSERVATIONS = 5  # must match backend db/index.ts
PREFIX_AMBIGUITY_RADIUS_KM = 45.0  # only penalize same-prefix ambiguity when nodes are realistically in range
MAX_RADIUS_M     = 100_000  # absolute cap on viewshed radius (m)
SIMPLIFY_DEG     = 0.001    # Douglas-Peucker tolerance (~100 m)
GEOMETRY_GRID_DEG = 1e-9    # fixed-precision overlay grid (~0.1 mm)
N_RAYS           = 720      # number of radial rays cast from the observer
STEP_M           = 50.0     # ray step size in metres
ANGLE_EPS        = 1e-9     # numerical tolerance for horizon comparisons
RF_RADIAL_STEP_M = 100.0    # radial search precision for RF coverage mode
RF_N_RAYS        = 360      # 1-degree azimuth resolution keeps RF mode tractable
RF_RADIUS_MULTIPLIER = 1.35 # search beyond geometric horizon to allow limited diffraction gain
RF_MIN_RADIUS_M  = 20_000   # avoid under-searching low-elevation repeaters
RF_SOURCE_LINK_RADIUS_MULTIPLIER = float(os.environ.get('RF_SOURCE_LINK_RADIUS_MULTIPLIER', '1.25'))
DEFAULT_PHYSICAL_LINK_RADIUS_KM = float(os.environ.get('DEFAULT_PHYSICAL_LINK_RADIUS_KM', '60'))
MIN_PHYSICAL_LINK_RADIUS_KM = float(os.environ.get('MIN_PHYSICAL_LINK_RADIUS_KM', '20'))
MAX_PHYSICAL_LINK_RADIUS_KM = float(os.environ.get('MAX_PHYSICAL_LINK_RADIUS_KM', '100'))
SUPPORT_REFRESH_S = int(os.environ.get('COVERAGE_SUPPORT_REFRESH_S', '900'))
LINK_TOPOLOGY_REFRESH_S = max(5, int(os.environ.get('LINK_TOPOLOGY_REFRESH_S', '60')))
SUPPORT_NEARBY_REPEATER_KM = float(os.environ.get('COVERAGE_SUPPORT_NEARBY_REPEATER_KM', '12'))
SUPPORT_PENALTY_PER_KM_DB = float(os.environ.get('COVERAGE_SUPPORT_PENALTY_PER_KM_DB', '0.6'))
SUPPORT_MAX_PENALTY_DB = float(os.environ.get('COVERAGE_SUPPORT_MAX_PENALTY_DB', '14'))
SUPPORT_PROJECTION_LAT = float(os.environ.get('COVERAGE_SUPPORT_PROJECTION_LAT', '54.0'))

RADIO_NEIGHBOR_REFRESH_S = int(os.environ.get('RADIO_NEIGHBOR_REFRESH_S', '300'))
RADIO_NEIGHBOR_MAX_AGE_HOURS = float(os.environ.get('RADIO_NEIGHBOR_MAX_AGE_HOURS', '72'))

# Radio horizon parameters
# K-factor, wavelength, RF calibration state, and path-loss thresholds now
# live in rf.config so the RF model can be reused outside the worker loop.

SUPPORT_CONTEXT = {
    'tree': None,
    'node_ids': [],
    'node_index_by_id': {},
    'max_link_km_by_node': {},
    'updated_at': 0.0,
    'source_signature': None,
}

# Link observations can arrive much faster than topology changes. Keep a short-lived
# process-local snapshot so each job does not reload every repeater and viable pair.
LINK_TOPOLOGY = {
    'nodes': {},
    'physical_pairs': set(),
    'updated_at': 0.0,
}

UK_LAT_MIN = 49.5
UK_LAT_MAX = 61.5
UK_LON_MIN = -8.5
UK_LON_MAX = 2.5


def peak_rss_mb() -> float:
    # Linux reports ru_maxrss in KiB; the production worker image is Linux.
    return float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024.0


def check_job_budget(
    deadline_monotonic: Optional[float],
    cancellation_check: Optional[Callable[[], None]] = None,
) -> None:
    if cancellation_check is not None:
        try:
            cancellation_check()
        except Cancelled:
            raise
        except Exception as exc:
            raise Cancelled(str(exc)) from exc
    if deadline_monotonic is not None and time.monotonic() >= deadline_monotonic:
        raise JobDeadlineExceeded('coverage job deadline exceeded')


def is_viewshed_eligible_coordinate(lat: float, lon: float) -> bool:
    if not math.isfinite(lat) or not math.isfinite(lon):
        return False
    if abs(lat) < 1e-9 and abs(lon) < 1e-9:
        return False
    return UK_LAT_MIN <= lat <= UK_LAT_MAX and UK_LON_MIN <= lon <= UK_LON_MAX

def weighted_quantile(values: np.ndarray, weights: np.ndarray, q: float) -> float:
    if values.size < 1:
        raise ValueError('No values provided')
    order = np.argsort(values)
    v = values[order]
    w = weights[order]
    cumulative = np.cumsum(w)
    target = float(np.clip(q, 0.0, 1.0)) * cumulative[-1]
    idx = int(np.searchsorted(cumulative, target, side='left'))
    idx = max(0, min(idx, len(v) - 1))
    return float(v[idx])


def project_xy_km(latitudes, longitudes) -> np.ndarray:
    lats = np.asarray(latitudes, dtype=np.float64)
    lons = np.asarray(longitudes, dtype=np.float64)
    cos_ref = math.cos(math.radians(SUPPORT_PROJECTION_LAT))
    return np.column_stack((lons * 111.32 * cos_ref, lats * 111.32))


def node_dist_km(a: dict, b: dict) -> float:
    cos_m = math.cos(math.radians((a['lat'] + b['lat']) / 2))
    return math.sqrt(
        ((a['lat'] - b['lat']) * 111.32) ** 2 +
        ((a['lon'] - b['lon']) * 111.32 * cos_m) ** 2
    )


def physical_candidate_radius_km(radius_m: Optional[float]) -> float:
    derived = (radius_m / 1000.0) * RF_SOURCE_LINK_RADIUS_MULTIPLIER if radius_m is not None else DEFAULT_PHYSICAL_LINK_RADIUS_KM
    return min(MAX_PHYSICAL_LINK_RADIUS_KM, max(MIN_PHYSICAL_LINK_RADIUS_KM, derived))


def refresh_rf_calibration(db, force: bool = False) -> None:
    now = time.time()
    if not force and now - float(RF_CALIBRATION['updated_at']) < CALIBRATION_REFRESH_S:
        return

    with db.cursor() as cur:
        cur.execute(
            '''SELECT
                 (SELECT MAX(p.time)
                    FROM packets p
                   WHERE p.snr IS NOT NULL
                     AND p.rx_node_id IS NOT NULL
                     AND p.src_node_id IS NOT NULL
                     AND p.rx_node_id <> p.src_node_id
                     AND p.hop_count = 0
                     AND p.time > NOW() - (%s * INTERVAL '1 hour')),
                 (SELECT MAX(nl.itm_computed_at) FROM node_links nl),
                 (SELECT COUNT(*) FROM node_links nl
                   WHERE nl.itm_path_loss_db IS NOT NULL AND nl.force_viable = false)''',
            (RADIO_NEIGHBOR_MAX_AGE_HOURS,),
        )
        source_signature = cur.fetchone()
        if not force and source_signature == RF_CALIBRATION.get('source_signature'):
            RF_CALIBRATION['updated_at'] = now
            return
        cur.execute(
            '''
            WITH link_packet_stats AS (
              SELECT
                LEAST(p.rx_node_id, p.src_node_id)    AS node_a_id,
                GREATEST(p.rx_node_id, p.src_node_id) AS node_b_id,
                PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY p.snr) AS median_snr_db,
                COUNT(*)                               AS packet_count
              FROM packets p
              WHERE p.snr IS NOT NULL
                AND p.rx_node_id IS NOT NULL
                AND p.src_node_id IS NOT NULL
                AND p.rx_node_id <> p.src_node_id
                AND p.hop_count = 0
                AND p.time > NOW() - (%s * INTERVAL \'1 hour\')
              GROUP BY node_a_id, node_b_id
              HAVING COUNT(*) >= %s
            )
            SELECT
              nl.itm_path_loss_db,
              lps.packet_count    AS neighbor_report_count,
              lps.median_snr_db   AS neighbor_best_snr_db
            FROM node_links nl
            JOIN link_packet_stats lps
              ON lps.node_a_id = nl.node_a_id
             AND lps.node_b_id = nl.node_b_id
            WHERE nl.itm_path_loss_db IS NOT NULL
              AND nl.force_viable = false
            ''',
            (RADIO_NEIGHBOR_MAX_AGE_HOURS, CALIBRATION_MIN_OBSERVED_COUNT),
        )
        rows = cur.fetchall()

    RF_CALIBRATION['source_signature'] = source_signature

    if len(rows) < CALIBRATION_MIN_LINKS:
        RF_CALIBRATION['usable_path_loss_db'] = DEFAULT_USABLE_PATH_LOSS_DB
        RF_CALIBRATION['signal_thresholds_db'] = {
            'green': max(116.0, DEFAULT_USABLE_PATH_LOSS_DB - 16.0),
            'amber': max(124.0, DEFAULT_USABLE_PATH_LOSS_DB - 8.0),
            'red': DEFAULT_USABLE_PATH_LOSS_DB,
        }
        RF_CALIBRATION['samples'] = len(rows)
        RF_CALIBRATION['updated_at'] = now
        log.info(
            'RF calibration: insufficient observed links '
            f'({len(rows)}/{CALIBRATION_MIN_LINKS}) — using default threshold {DEFAULT_USABLE_PATH_LOSS_DB:.1f} dB'
        )
        return

    losses = np.asarray([float(row[0]) for row in rows], dtype=np.float64)
    counts = np.asarray([max(1.0, float(row[1])) for row in rows], dtype=np.float64)
    snrs = np.asarray([float(row[2]) if row[2] is not None else -6.0 for row in rows], dtype=np.float64)
    snr_weights = np.asarray([
        1.75 if radio_snr_band(snr) == 'strong'
        else 1.35 if radio_snr_band(snr) == 'medium'
        else 1.10 if radio_snr_band(snr) == 'weak'
        else 0.75
        for snr in snrs
    ], dtype=np.float64)
    weights = np.clip(np.sqrt(counts) * snr_weights, 0.5, 24.0)

    green_threshold = weighted_quantile(losses, weights, 0.35)
    amber_threshold = weighted_quantile(losses, weights, 0.65)
    observed_tail = weighted_quantile(losses, weights, CALIBRATION_PERCENTILE)
    usable_threshold = observed_tail + CALIBRATION_EXTRA_MARGIN_DB
    usable_threshold = min(DEFAULT_USABLE_PATH_LOSS_DB + CALIBRATION_MAX_THRESHOLD_BOOST_DB, usable_threshold)
    usable_threshold = max(110.0, usable_threshold)
    green_threshold = min(green_threshold, usable_threshold)
    amber_threshold = min(max(green_threshold, amber_threshold), usable_threshold)

    RF_CALIBRATION['usable_path_loss_db'] = round(float(usable_threshold), 2)
    RF_CALIBRATION['signal_thresholds_db'] = {
        'green': round(float(green_threshold), 2),
        'amber': round(float(amber_threshold), 2),
        'red': round(float(usable_threshold), 2),
    }
    RF_CALIBRATION['samples'] = len(rows)
    RF_CALIBRATION['updated_at'] = now
    log.info(
        'RF calibration: '
        f'samples={len(rows)}, q35={green_threshold:.1f} dB, q65={amber_threshold:.1f} dB, '
        f'p{int(CALIBRATION_PERCENTILE * 100)}={observed_tail:.1f} dB, '
        f'usable={RF_CALIBRATION["usable_path_loss_db"]:.1f} dB, '
        f'green={RF_CALIBRATION["signal_thresholds_db"]["green"]:.1f}, '
        f'amber={RF_CALIBRATION["signal_thresholds_db"]["amber"]:.1f}'
    )


def refresh_radio_neighbor_reports(db, r_client, force: bool = False) -> None:
    if not RADIO_BOT_URL:
        return

    now = time.time()
    if not force and now - float(RADIO_NEIGHBOR_SYNC['updated_at']) < RADIO_NEIGHBOR_REFRESH_S:
        return

    try:
        response = requests.get(f'{RADIO_BOT_URL}/state', timeout=10)
        response.raise_for_status()
        payload = response.json()
    except Exception as exc:
        log.warning(f'Radio neighbour sync failed: {exc}')
        RADIO_NEIGHBOR_SYNC['updated_at'] = now
        return

    monitors = payload.get('monitors') if isinstance(payload, dict) else None
    if not isinstance(monitors, list):
        RADIO_NEIGHBOR_SYNC['updated_at'] = now
        return

    imported = 0
    queued = 0
    with db.cursor() as cur:
        for monitor in monitors:
            if not isinstance(monitor, dict):
                continue
            reporter_id = str(monitor.get('fullPublicKey') or monitor.get('targetHex') or '').strip().lower()
            if len(reporter_id) != 64:
                continue

            neighbours_at_raw = monitor.get('lastNeighboursAt') or monitor.get('lastSuccessAt')
            base_seen_at: Optional[dt.datetime] = None
            if neighbours_at_raw:
                try:
                    base_seen_at = dt.datetime.fromisoformat(str(neighbours_at_raw).replace('Z', '+00:00'))
                except Exception:
                    base_seen_at = None

            last_neighbours = monitor.get('lastNeighbours')
            if not isinstance(last_neighbours, list):
                continue

            for neighbour in last_neighbours:
                if not isinstance(neighbour, dict):
                    continue
                peer_id = str(neighbour.get('fullPublicKey') or '').strip().lower()
                if len(peer_id) != 64 or peer_id == reporter_id:
                    continue
                snr_db = neighbour.get('snrDb')
                if not isinstance(snr_db, (int, float)):
                    continue

                seen_at = base_seen_at
                heard_seconds_ago = neighbour.get('heardSecondsAgo')
                if seen_at is not None and isinstance(heard_seconds_ago, (int, float)):
                    seen_at = seen_at - dt.timedelta(seconds=float(heard_seconds_ago))
                if seen_at is None:
                    seen_at = dt.datetime.now(dt.timezone.utc)

                a_id, b_id = sorted((reporter_id, peer_id))
                cur.execute(
                    '''
                    INSERT INTO node_links (
                      node_a_id, node_b_id, observed_count, last_observed, count_a_to_b, count_b_to_a
                    )
                    VALUES (%s, %s, 0, NOW(), 0, 0)
                    ON CONFLICT (node_a_id, node_b_id) DO NOTHING
                    ''',
                    (a_id, b_id),
                )
                cur.execute(
                    '''
                    INSERT INTO node_link_radio_reports (
                      node_a_id, node_b_id, reporter_node_id, peer_node_id,
                      last_snr_db, best_snr_db, last_seen, sample_count
                    )
                    VALUES (%s, %s, %s, %s, %s, %s, %s, 1)
                    ON CONFLICT (node_a_id, node_b_id, reporter_node_id) DO UPDATE
                    SET peer_node_id = EXCLUDED.peer_node_id,
                        last_snr_db = CASE
                          WHEN EXCLUDED.last_seen >= node_link_radio_reports.last_seen
                            THEN EXCLUDED.last_snr_db
                          ELSE node_link_radio_reports.last_snr_db
                        END,
                        best_snr_db = GREATEST(
                          COALESCE(node_link_radio_reports.best_snr_db, EXCLUDED.best_snr_db),
                          EXCLUDED.best_snr_db
                        ),
                        last_seen = GREATEST(node_link_radio_reports.last_seen, EXCLUDED.last_seen),
                        sample_count = CASE
                          WHEN EXCLUDED.last_seen > node_link_radio_reports.last_seen
                            THEN node_link_radio_reports.sample_count + 1
                          ELSE node_link_radio_reports.sample_count
                        END
                    ''',
                    (a_id, b_id, reporter_id, peer_id, float(snr_db), float(snr_db), seen_at),
                )
                status, _ = link_queue_v3.admit_physical(r_client, a_id, b_id)
                if status not in ('accepted', 'coalesced', 'duplicate'):
                    log.warning(f'Radio neighbour link job not admitted: {status}')
                imported += 1
                if status in ('accepted', 'coalesced', 'duplicate'):
                    queued += 1
    db.commit()
    RADIO_NEIGHBOR_SYNC['updated_at'] = now
    if imported > 0:
        log.info(f'Radio neighbour sync: imported {imported} report(s), queued {queued} physical link job(s)')


def refresh_support_context(db, force: bool = False) -> None:
    now = time.time()
    if not force and now - float(SUPPORT_CONTEXT['updated_at']) < SUPPORT_REFRESH_S:
        return

    with db.cursor() as cur:
        cur.execute(
            '''WITH positioned AS (
                 SELECT node_id, lat, lon
                 FROM nodes
                 WHERE lat IS NOT NULL
                   AND lon IS NOT NULL
                   AND lat BETWEEN %s AND %s
                   AND lon BETWEEN %s AND %s
                   AND NOT (ABS(lat) < 1e-9 AND ABS(lon) < 1e-9)
                   AND (name IS NULL OR name NOT LIKE %s)
                   AND (role IS NULL OR role = 2)
               ), viable AS (
                 SELECT node_a_id, node_b_id, itm_viable, force_viable
                 FROM node_links
                 WHERE itm_viable = true OR force_viable = true
               )
               SELECT
                 (SELECT md5(COALESCE(string_agg(
                    node_id || ':' || lat::text || ':' || lon::text,
                    ',' ORDER BY node_id), '')) FROM positioned),
                 (SELECT md5(COALESCE(string_agg(
                    node_a_id || ':' || node_b_id || ':' || itm_viable::text || ':' || force_viable::text,
                    ',' ORDER BY node_a_id, node_b_id), '')) FROM viable)''',
            (UK_LAT_MIN, UK_LAT_MAX, UK_LON_MIN, UK_LON_MAX, '%🚫%',),
        )
        source_signature = cur.fetchone()
        if not force and source_signature == SUPPORT_CONTEXT.get('source_signature'):
            SUPPORT_CONTEXT['updated_at'] = now
            return
        cur.execute(
            '''
            SELECT node_id, lat, lon
            FROM nodes
            WHERE lat IS NOT NULL
              AND lon IS NOT NULL
              AND lat BETWEEN %s AND %s
              AND lon BETWEEN %s AND %s
              AND NOT (ABS(lat) < 1e-9 AND ABS(lon) < 1e-9)
              AND (name IS NULL OR name NOT LIKE %s)
              AND (role IS NULL OR role = 2)
            ''',
            (UK_LAT_MIN, UK_LAT_MAX, UK_LON_MIN, UK_LON_MAX, '%🚫%',),
        )
        repeater_rows = cur.fetchall()
        cur.execute(
            '''
            SELECT nl.node_a_id, nl.node_b_id,
                   na.lat, na.lon, nb.lat, nb.lon
            FROM node_links nl
            JOIN nodes na ON na.node_id = nl.node_a_id
            JOIN nodes nb ON nb.node_id = nl.node_b_id
            WHERE na.lat IS NOT NULL
              AND na.lon IS NOT NULL
              AND nb.lat IS NOT NULL
              AND nb.lon IS NOT NULL
              AND (nl.itm_viable = true OR nl.force_viable = true)
            '''
        )
        link_rows = cur.fetchall()

    node_ids = [row[0] for row in repeater_rows]
    xy = project_xy_km([row[1] for row in repeater_rows], [row[2] for row in repeater_rows]) if repeater_rows else np.empty((0, 2))
    SUPPORT_CONTEXT['tree'] = cKDTree(xy) if len(node_ids) > 0 else None
    SUPPORT_CONTEXT['node_ids'] = node_ids
    SUPPORT_CONTEXT['node_index_by_id'] = {node_id: idx for idx, node_id in enumerate(node_ids)}

    max_link_km_by_node: dict[str, float] = {}
    for a_id, b_id, a_lat, a_lon, b_lat, b_lon in link_rows:
      cos_mid = math.cos(math.radians((a_lat + b_lat) / 2))
      dist_km = math.sqrt(
          ((a_lat - b_lat) * 111.32) ** 2 +
          ((a_lon - b_lon) * 111.32 * cos_mid) ** 2
      )
      if dist_km <= 0:
          continue
      max_link_km_by_node[a_id] = max(max_link_km_by_node.get(a_id, 0.0), dist_km)
      max_link_km_by_node[b_id] = max(max_link_km_by_node.get(b_id, 0.0), dist_km)
    SUPPORT_CONTEXT['max_link_km_by_node'] = max_link_km_by_node
    SUPPORT_CONTEXT['source_signature'] = source_signature
    SUPPORT_CONTEXT['updated_at'] = now
    log.info(
        f'Mesh support context: repeaters={len(node_ids)}, '
        f'link-capped nodes={len(max_link_km_by_node)}'
    )


def source_support_radius_m(node_id: str, fallback_radius_m: float) -> float:
    max_link_km = SUPPORT_CONTEXT['max_link_km_by_node'].get(node_id)
    if not max_link_km:
        return fallback_radius_m
    return min(
        fallback_radius_m,
        max(RF_MIN_RADIUS_M, max_link_km * 1000.0 * RF_SOURCE_LINK_RADIUS_MULTIPLIER),
    )


def support_penalty_db(source_node_id: str, sample_lats: np.ndarray, sample_lons: np.ndarray) -> np.ndarray:
    tree: Optional[cKDTree] = SUPPORT_CONTEXT['tree']
    node_ids: list[str] = SUPPORT_CONTEXT['node_ids']
    source_index = SUPPORT_CONTEXT['node_index_by_id'].get(source_node_id)
    if tree is None or len(node_ids) < 1:
        return np.zeros(sample_lats.shape[0], dtype=np.float32)

    points_xy = project_xy_km(sample_lats, sample_lons)
    k = 2 if source_index is not None and len(node_ids) > 1 else 1
    distances, indices = tree.query(points_xy, k=k)

    if k == 1:
        nearest_km = np.asarray(distances, dtype=np.float32)
    else:
        d = np.asarray(distances, dtype=np.float32)
        i = np.asarray(indices, dtype=np.int32)
        primary_is_source = i[:, 0] == source_index
        nearest_km = np.where(primary_is_source, d[:, 1], d[:, 0]).astype(np.float32)

    penalty = np.maximum(0.0, nearest_km - SUPPORT_NEARBY_REPEATER_KM) * SUPPORT_PENALTY_PER_KM_DB
    return np.clip(penalty, 0.0, SUPPORT_MAX_PENALTY_DB).astype(np.float32)


def rf_search_radius_m(base_radius_m: float) -> float:
    """Return the exact radius both terrain acquisition and RF rays must cover."""
    return min(
        MAX_RADIUS_M,
        max(base_radius_m * RF_RADIUS_MULTIPLIER, RF_MIN_RADIUS_M),
    )


def approximate_dtm_in_place(elevation: np.ndarray) -> np.ndarray:
    """Strip narrow DSM spikes without allocating another full-size raster."""
    if elevation.ndim != 2 or elevation.dtype not in (np.dtype(np.int16), np.dtype(np.float32)):
        raise ValueError('DTM raster must be a two-dimensional int16 or float32 array')
    _min_filter(elevation, size=9, output=elevation)
    return elevation


def resolve_rf_radial_boundaries(node_id: str,
                                 lat: float,
                                 lon: float,
                                 elev: np.ndarray,
                                 gt: tuple[float, float, float, float, float, float],
                                 observer_h: float,
                                 base_radius_m: float,
                                 *,
                                 deadline_monotonic: Optional[float] = None,
                                 cancellation_check: Optional[Callable[[], None]] = None,
                                 raster_x_offset: int = 0,
                                 raster_y_offset: int = 0,
                                 ) -> tuple[dict[str, list[tuple[float, float]]], float]:
    check_job_budget(deadline_monotonic, cancellation_check)
    search_radius_m = rf_search_radius_m(base_radius_m)
    n_rows, n_cols = elev.shape
    dpmlat = 1.0 / 111_320.0
    dpmlon = 1.0 / (111_320.0 * math.cos(math.radians(lat)))
    ds_arr = np.arange(RF_RADIAL_STEP_M, search_radius_m + RF_RADIAL_STEP_M, RF_RADIAL_STEP_M, dtype=np.float32)
    thetas = np.linspace(0.0, 2.0 * math.pi, RF_N_RAYS, endpoint=False, dtype=np.float32)
    cos_t = np.cos(thetas)
    sin_t = np.sin(thetas)

    signal_thresholds = current_signal_thresholds_db()
    boundaries: dict[str, list[tuple[float, float]]] = {key: [] for key in signal_thresholds}
    max_reached = 0.0

    # Materialise coordinates, raster samples, and support penalties once per
    # job. Prefix path loss remains ray-specific, but runs in bounded NumPy
    # endpoint batches instead of rebuilding every preceding prefix in Python.
    pt_lats = lat + sin_t[:, None] * ds_arr[None, :] * dpmlat
    pt_lons = lon + cos_t[:, None] * ds_arr[None, :] * dpmlon
    # Convert in the original VRT pixel space before subtracting the integer
    # read-window origin. Recomputing against a shifted floating-point origin
    # can move samples on exact pixel boundaries by one cell.
    pxs = np.clip(
        ((pt_lons - gt[0]) / gt[1]).astype(np.int32) - raster_x_offset,
        0,
        n_cols - 1,
    )
    pys = np.clip(
        ((pt_lats - gt[3]) / gt[5]).astype(np.int32) - raster_y_offset,
        0,
        n_rows - 1,
    )
    ray_heights = elev[pys, pxs].astype(np.float32)
    support_penalties = support_penalty_db(
        node_id,
        pt_lats.reshape(-1),
        pt_lons.reshape(-1),
    ).reshape(pt_lats.shape)

    for ray_start in range(0, RF_N_RAYS, RF_PREFIX_RAY_BATCH):
        check_job_budget(deadline_monotonic, cancellation_check)
        ray_end = min(RF_N_RAYS, ray_start + RF_PREFIX_RAY_BATCH)
        ray_slice = slice(ray_start, ray_end)
        heights = ray_heights[ray_slice]
        prefix_result = compute_prefix_path_losses(
            ds_arr,
            heights,
            observer_h,
            heights + COVERAGE_TARGET_HEIGHT_M,
            endpoint_batch_size=RF_PREFIX_ENDPOINT_BATCH,
            ray_batch_size=RF_PREFIX_RAY_BATCH,
        )
        effective_losses = (
            prefix_result.path_loss_db
            + support_penalties[ray_slice]
        )

        for band, threshold in signal_thresholds.items():
            passing = effective_losses <= threshold
            any_passing = np.any(passing, axis=1)
            last_passing = len(ds_arr) - 1 - np.argmax(
                passing[:, ::-1],
                axis=1,
            )
            end_dists = np.where(
                any_passing,
                ds_arr[last_passing],
                ds_arr[0],
            )
            if band == 'red':
                max_reached = max(max_reached, float(np.max(end_dists)))
            boundaries[band].extend(zip(
                lon + cos_t[ray_slice] * end_dists * dpmlon,
                lat + sin_t[ray_slice] * end_dists * dpmlat,
            ))

    check_job_budget(deadline_monotonic, cancellation_check)
    for band_boundary in boundaries.values():
        if band_boundary:
            band_boundary.append(band_boundary[0])
    return boundaries, max_reached


def clip_and_simplify_polygon(poly) -> Optional[dict]:
    if poly.is_empty:
        return None
    if not poly.is_valid:
        poly = poly.buffer(0)
    if UK_MAINLAND is not None:
        poly = poly.intersection(UK_MAINLAND)
        if poly.is_empty:
            return None
    result = poly.simplify(SIMPLIFY_DEG, preserve_topology=True)
    if result.is_empty or result.geom_type not in ('Polygon', 'MultiPolygon'):
        return None
    return mapping(result)


def require_mapped_coverage(node_id: str, geom: Optional[dict], model_name: str) -> dict:
    """Return mapped coverage or classify a stable land-mask miss as terminal."""
    if geom is None:
        raise PermanentOutOfScope(
            f'{node_id}: {model_name} coverage does not intersect the map land mask'
        )
    return geom


def _polygonal_geometry(poly):
    if poly is None or poly.is_empty:
        return None
    if not poly.is_valid:
        poly = poly.buffer(0)
    if poly.is_empty:
        return None
    if poly.geom_type in ('Polygon', 'MultiPolygon'):
        return poly
    if poly.geom_type == 'GeometryCollection':
        polygon_parts = [
            part for part in poly.geoms
            if part.geom_type in ('Polygon', 'MultiPolygon') and not part.is_empty
        ]
        if polygon_parts:
            return unary_union(polygon_parts)
    return None


def _clipped_simplified_polygon(poly):
    if poly is None:
        return None
    geom = clip_and_simplify_polygon(poly)
    return _polygonal_geometry(shapely_shape(geom)) if geom is not None else None


def build_exclusive_strength_geoms(
    band_polys: dict[str, ShapelyPolygon],
    *,
    red_outer_geom: Optional[dict] = None,
) -> dict[str, dict]:
    """Convert nested strength polygons into exclusive green/amber/red areas.

    The strongest band should own the fill for a location. Without this, the
    frontend ends up stacking green over amber over red and the center reads as
    muddy yellow instead of a clean strength gradient.
    """
    red_outer = (
        _polygonal_geometry(shapely_shape(red_outer_geom))
        if red_outer_geom is not None
        else _clipped_simplified_polygon(band_polys.get('red'))
    )
    if red_outer is None:
        return {}

    amber_outer = _clipped_simplified_polygon(band_polys.get('amber'))
    if amber_outer is not None:
        amber_outer = _polygonal_geometry(shapely.intersection(
            amber_outer,
            red_outer,
            grid_size=GEOMETRY_GRID_DEG,
        ))

    green_outer = _clipped_simplified_polygon(band_polys.get('green'))
    if green_outer is not None:
        green_limit = amber_outer if amber_outer is not None else red_outer
        green_outer = _polygonal_geometry(shapely.intersection(
            green_outer,
            green_limit,
            grid_size=GEOMETRY_GRID_DEG,
        ))

    exclusive_shapes: dict[str, object] = {}
    occupied = None
    if green_outer is not None:
        exclusive_shapes['green'] = green_outer
        occupied = green_outer
    if amber_outer is not None:
        amber_only = _polygonal_geometry(
            shapely.difference(
                amber_outer,
                occupied,
                grid_size=GEOMETRY_GRID_DEG,
            )
            if occupied is not None
            else amber_outer
        )
        if amber_only is not None:
            exclusive_shapes['amber'] = amber_only
            occupied = (
                shapely.union_all(
                    (occupied, amber_only),
                    grid_size=GEOMETRY_GRID_DEG,
                )
                if occupied is not None
                else amber_only
            )
    red_only = _polygonal_geometry(
        shapely.difference(
            red_outer,
            occupied,
            grid_size=GEOMETRY_GRID_DEG,
        )
        if occupied is not None
        else red_outer
    )
    if red_only is not None:
        exclusive_shapes['red'] = red_only
    return {name: mapping(poly) for name, poly in exclusive_shapes.items()}


UK_MAINLAND = load_uk_mainland(Path(__file__).parent, log)


# ── Viewshed calculation ──────────────────────────────────────────────────────

def calculate_viewshed(
    node_id: str,
    lat: float,
    lon: float,
    antenna_height_m: float = ANTENNA_HEIGHT_M,
    *,
    deadline_monotonic: Optional[float] = None,
    cancellation_check: Optional[Callable[[], None]] = None,
) -> Computed:
    calculation_started = time.monotonic()
    stage_seconds: dict[str, float] = {}

    def record_stage(name: str, started: float) -> None:
        stage_seconds[name] = round(time.monotonic() - started, 6)

    check_job_budget(deadline_monotonic, cancellation_check)
    if not is_viewshed_eligible_coordinate(lat, lon):
        raise PermanentOutOfScope(
            f'coordinate outside UK coverage bounds at ({lat:.4f}, {lon:.4f})'
        )
    with tempfile.TemporaryDirectory() as tmp:
        # 1. Download the observer's own tile and sample terrain elevation.
        #    This single tile is sufficient to determine node height; we need
        #    it before we know how far to reach for surrounding tiles.
        stage_started = time.monotonic()
        obs_tile = (math.floor(lat), math.floor(lon))
        obs_path = ensure_tiles(
            SRTM_DIR,
            (obs_tile,),
            log,
            required_tiles=(obs_tile,),
            max_tiles=1,
        )[0]
        record_stage('observer_tile_acquisition', stage_started)
        check_job_budget(deadline_monotonic, cancellation_check)

        stage_started = time.monotonic()
        obs_vrt = f'{tmp}/observer.vrt'
        build_vrt((obs_path,), obs_vrt)
        elevation_m = sample_elevation(obs_vrt, lat, lon)
        record_stage('observer_vrt_and_sample', stage_started)
        check_job_budget(deadline_monotonic, cancellation_check)

        # 2. Radio-horizon radius: node ASL + per-node antenna height.
        effective_height_m = elevation_m + antenna_height_m
        radius_m = min(radio_horizon_m(effective_height_m), MAX_RADIUS_M)
        radius_m = source_support_radius_m(node_id, radius_m)
        log.info(
            f'  {node_id[:12]}…: elevation={elevation_m:.0f} m ASL, '
            f'antenna={effective_height_m:.0f} m, horizon={radius_m / 1000:.1f} km'
        )

        # 3. Download all tiles covering the computed horizon radius
        stage_started = time.monotonic()
        terrain_radius_m = (
            rf_search_radius_m(radius_m)
            if COVERAGE_MODEL == 'rf_radial_100m'
            else radius_m
        )
        paths = ensure_tiles_for_radius(
            SRTM_DIR,
            lat,
            lon,
            terrain_radius_m,
            log,
        )
        record_stage('coverage_tile_acquisition', stage_started)
        check_job_budget(deadline_monotonic, cancellation_check)

        # 4. Merge tiles into a single VRT
        stage_started = time.monotonic()
        vrt = f'{tmp}/input.vrt'
        build_vrt(paths, vrt)
        record_stage('coverage_vrt_construction', stage_started)
        check_job_budget(deadline_monotonic, cancellation_check)

        # 5. Read only the radius-bounded elevation window into memory once.
        #    NODATA ocean pixels (INT16 -32768) are clamped to 0 — treated as sea level.
        stage_started = time.monotonic()
        ds   = gdal.Open(vrt)
        if ds is None:
            raise RetryableTerrainError('GDAL could not open the coverage VRT')
        full_gt = ds.GetGeoTransform()   # (x_origin, px_lon, 0, y_origin, 0, px_lat)
        band = ds.GetRasterBand(1)
        if band is None:
            ds = None
            raise RetryableTerrainError('coverage VRT has no elevation band')
        x_offset, y_offset, x_size, y_size, _window_gt = raster_window_for_radius(
            full_gt,
            ds.RasterXSize,
            ds.RasterYSize,
            lat,
            lon,
            terrain_radius_m,
            # minimum_filter(size=9) needs four real neighbouring pixels at
            # the coverage edge. One extra pixel also absorbs float-to-index
            # rounding at the final radial step.
            margin_pixels=5,
        )
        elev = band.ReadAsArray(x_offset, y_offset, x_size, y_size)
        if elev is None:
            ds = None
            raise RetryableTerrainError('GDAL could not read the coverage elevation raster')
        # SRTM is signed int16. Keep the full radius window in that native
        # representation through the DTM filter; only the bounded ray samples
        # are promoted to float32 later. This halves the dominant per-job
        # raster allocation without changing any elevation values.
        if elev.dtype != np.int16:
            elev = elev.astype(np.int16)
        np.maximum(elev, 0, out=elev)
        n_rows, n_cols = elev.shape
        ds = None
        record_stage('terrain_raster_read', stage_started)
        check_job_budget(deadline_monotonic, cancellation_check)

        # 5b. Approximate DTM from SRTM DSM via spatial minimum filter.
        #     SRTM is a Digital Surface Model — building heights corrupt urban
        #     areas causing raycasting to terminate within metres of the observer.
        #     A 9-pixel (~270 m for SRTM1 at 30 m/px) minimum filter strips
        #     building-height spikes while preserving genuine terrain features
        #     (hills, ridges) whose footprints are wider than ~270 m.
        stage_started = time.monotonic()
        approximate_dtm_in_place(elev)

        # 5c. Re-sample observer elevation from the DTM-approximated raster.
        #     This corrects the radio-horizon radius when SRTM reads building tops.
        obs_px = int(np.clip(
            int((lon - full_gt[0]) / full_gt[1]) - x_offset,
            0,
            n_cols - 1,
        ))
        obs_py = int(np.clip(
            int((lat - full_gt[3]) / full_gt[5]) - y_offset,
            0,
            n_rows - 1,
        ))
        dtm_elev = float(elev[obs_py, obs_px])
        # Guard against coastal bleed-in: min filter near shoreline may return 0
        # (ocean NODATA) even for land pixels.  Fall back to raw SRTM in that case.
        if dtm_elev > 0.0 or elevation_m <= 0.0:
            elevation_m = dtm_elev
        record_stage('terrain_filter_and_origin_sample', stage_started)
        check_job_budget(deadline_monotonic, cancellation_check)
        effective_height_m = elevation_m + antenna_height_m
        radius_m = min(radio_horizon_m(effective_height_m), MAX_RADIUS_M)
        radius_m = source_support_radius_m(node_id, radius_m)
        log.info(
            f'  {node_id[:12]}… DTM elevation={elevation_m:.0f} m ASL, '
            f'horizon={radius_m / 1000:.1f} km'
        )

        observer_h = elevation_m + antenna_height_m
        strength_geoms: dict[str, dict] = {}
        radial_started = time.monotonic()
        if COVERAGE_MODEL == 'terrain_los':
            # Vectorised raycasting terrain line-of-sight model.
            dpmlat = 1.0 / 111_320.0                                       # deg/m northward
            dpmlon = 1.0 / (111_320.0 * math.cos(math.radians(lat)))       # deg/m eastward
            R_eff_2 = 2.0 * K_FACTOR * R_EARTH_M                          # 2kR curvature denom

            n_steps = max(1, int(radius_m / STEP_M))
            ds_arr  = np.linspace(STEP_M, radius_m, n_steps)    # (M,) distances in metres
            thetas  = np.linspace(0.0, 2.0 * math.pi, N_RAYS, endpoint=False)   # (N,) angles

            # Ray sample coordinates: (N, M)
            sin_t   = np.sin(thetas)[:, None]    # (N, 1)
            cos_t   = np.cos(thetas)[:, None]    # (N, 1)
            pt_lats = lat + sin_t * ds_arr[None, :] * dpmlat   # (N, M)
            pt_lons = lon + cos_t * ds_arr[None, :] * dpmlon   # (N, M)

            # Pixel indices — clamped to raster bounds (N, M)
            pxs = np.clip(
                ((pt_lons - full_gt[0]) / full_gt[1]).astype(np.int32) - x_offset,
                0,
                n_cols - 1,
            )
            pys = np.clip(
                ((pt_lats - full_gt[3]) / full_gt[5]).astype(np.int32) - y_offset,
                0,
                n_rows - 1,
            )

            # Terrain heights at each ray step: (N, M)
            hs = elev[pys, pxs]

            # Angles with Earth-curvature correction: (N, M)
            curvature = ds_arr[None, :] ** 2 / R_eff_2
            terrain_angles = (hs - observer_h - curvature) / ds_arr[None, :]
            target_angles = ((hs + COVERAGE_TARGET_HEIGHT_M) - observer_h - curvature) / ds_arr[None, :]

            running_max = np.maximum.accumulate(terrain_angles, axis=1)
            prev_max  = np.concatenate([np.full((N_RAYS, 1), -np.inf), running_max[:, :-1]], axis=1)
            in_shadow = target_angles + ANGLE_EPS < prev_max

            has_shadow  = in_shadow.any(axis=1)
            first_shad  = np.where(has_shadow, in_shadow.argmax(axis=1), n_steps)
            last_js     = np.clip(first_shad - 1, 0, n_steps - 1)
            last_ds     = ds_arr[last_js]

            lons_b = lon + np.cos(thetas) * last_ds * dpmlon
            lats_b = lat + np.sin(thetas) * last_ds * dpmlat
            boundary = list(zip(lons_b.tolist(), lats_b.tolist()))
            boundary.append(boundary[0])
            poly = ShapelyPolygon(boundary)
            geom = require_mapped_coverage(
                node_id,
                clip_and_simplify_polygon(poly),
                'terrain',
            )
            strength_geoms = {'green': geom}
        elif COVERAGE_MODEL == 'rf_radial_100m':
            band_boundaries, radius_m = resolve_rf_radial_boundaries(
                node_id,
                lat,
                lon,
                elev,
                full_gt,
                observer_h,
                radius_m,
                deadline_monotonic=deadline_monotonic,
                cancellation_check=cancellation_check,
                raster_x_offset=x_offset,
                raster_y_offset=y_offset,
            )
            record_stage('radial_calculation', radial_started)
            check_job_budget(deadline_monotonic, cancellation_check)
            polygon_started = time.monotonic()
            raw_band_polys: dict[str, ShapelyPolygon] = {}
            for band, band_boundary in band_boundaries.items():
                if len(band_boundary) < 4:
                    continue
                band_poly = ShapelyPolygon(band_boundary)
                if not band_poly.is_empty:
                    raw_band_polys[band] = band_poly
            geom = require_mapped_coverage(
                node_id,
                (
                    clip_and_simplify_polygon(raw_band_polys.get('red'))
                    if raw_band_polys.get('red') is not None
                    else None
                ),
                'RF',
            )
            strength_geoms = build_exclusive_strength_geoms(
                raw_band_polys,
                red_outer_geom=geom,
            )
            record_stage('polygonization_and_clipping', polygon_started)
        else:
            raise InvalidJob(f'Unknown COVERAGE_MODEL={COVERAGE_MODEL}')
        if 'radial_calculation' not in stage_seconds:
            record_stage('radial_calculation', radial_started)
        check_job_budget(deadline_monotonic, cancellation_check)
        stage_seconds['calculation_total'] = round(
            time.monotonic() - calculation_started,
            6,
        )
        return Computed(
            geom,
            strength_geoms,
            radius_m,
            elevation_m,
            stage_seconds,
            peak_rss_mb(),
        )

# ── DB helpers ────────────────────────────────────────────────────────────────

def already_calculated(db, node_id: str, *, require_predicted_links: bool = False) -> bool:
    with db.cursor() as cur:
        cur.execute(
            '''SELECT 1
               FROM node_coverage nc
               LEFT JOIN nodes n ON n.node_id = nc.node_id
               WHERE nc.node_id = %s
                 AND nc.model_version >= %s
                 AND nc.calculation_status IN ('computed', 'permanent')
                 AND (%s IS FALSE OR nc.predicted_links IS NOT NULL OR nc.calculation_status = 'permanent')
                 AND (n.node_id IS NULL OR n.elevation_m IS NOT NULL)''',
            (node_id, COVERAGE_MODEL_VERSION, require_predicted_links),
        )
        return cur.fetchone() is not None


# ---- BUG-006: side-effect completion markers ----
# store_coverage() commits on an autocommit connection, then the worker queues
# physical-link jobs and publishes Redis notifications. If Redis fails after
# the DB commit, the job is NACKed and retried — but already_calculated() then
# sees the coverage row and returns early, permanently skipping the link work
# and frontend notifications. We record a completion marker in Redis only after
# EVERY side effect succeeds; on retry, a missing marker means the side effects
# must be replayed (link admission is idempotent; notifications re-publish the
# same stored data, which is harmless).
SIDE_EFFECT_MARKER_KEY = 'viewshed:side-effects:{node_id}'
SIDE_EFFECT_MARKER_TTL_S = 90 * 24 * 60 * 60  # 90 days — long enough to cover retry windows


def side_effects_complete(r_client, node_id: str) -> bool:
    try:
        return bool(r_client.exists(SIDE_EFFECT_MARKER_KEY.format(node_id=node_id)))
    except Exception:
        # If Redis is down we cannot confirm completion; treat as incomplete so
        # the caller replays (which will also fail and re-raise, NACKing the job).
        return False


def mark_side_effects_complete(r_client, node_id: str) -> None:
    r_client.set(SIDE_EFFECT_MARKER_KEY.format(node_id=node_id), '1', ex=SIDE_EFFECT_MARKER_TTL_S)


def replay_coverage_side_effects(db, r_client, node_id: str) -> None:
    """Re-enqueue link jobs and re-publish notifications for a node whose
    coverage row exists but whose post-commit side effects never completed."""
    with db.cursor() as cur:
        cur.execute(
            '''SELECT geom, strength_geoms, radius_m, elevation_m
               FROM node_coverage nc
               LEFT JOIN nodes n ON n.node_id = nc.node_id
               WHERE nc.node_id = %s''',
            (node_id,),
        )
        row = cur.fetchone()
    if row is None or row[0] is None:
        log.warning(f'Cannot replay side effects for {node_id[:12]}…: no coverage row')
        return
    geom, strength_geoms, radius_m, elevation_m = row
    if WORKER_MODE in ('all', 'link'):
        _replay_link_jobs(db, r_client, node_id, radius_m)
    _publish_coverage_notifications(r_client, node_id, geom, strength_geoms, elevation_m)
    mark_side_effects_complete(r_client, node_id)
    log.info(f'Replayed missing side effects for {node_id[:12]}…')


def _replay_link_jobs(db, r_client, node_id: str, radius_m: Optional[float]) -> None:
    """Reload the node position and enqueue physical link jobs (idempotent)."""
    with db.cursor() as cur:
        cur.execute(
            'SELECT lat, lon FROM nodes WHERE node_id = %s',
            (node_id,),
        )
        row = cur.fetchone()
    if row is None or row[0] is None or row[1] is None:
        log.warning(f'Cannot replay link jobs for {node_id[:12]}…: no position')
        return
    enqueue_physical_link_jobs_for_node(db, r_client, node_id, row[0], row[1], radius_m)


def _publish_coverage_notifications(r_client, node_id: str, geom, strength_geoms, elevation_m) -> None:
    r_client.publish(LIVE_CHANNEL, json.dumps({
        'type': 'coverage_update',
        'data': {'node_id': node_id, 'geom': geom, 'strength_geoms': strength_geoms},
        'ts':   int(time.time() * 1000),
    }))
    r_client.publish(LIVE_CHANNEL, json.dumps({
        'type': 'node_upsert',
        'data': {'node_id': node_id, 'elevation_m': round(float(elevation_m), 1) if elevation_m is not None else None},
        'ts':   int(time.time() * 1000),
    }))

def store_coverage(
    db,
    node_id: str,
    geom: dict,
    strength_geoms: dict[str, dict],
    radius_m: float,
    elevation_m: float,
    antenna_height_m: float = ANTENNA_HEIGHT_M,
    *,
    commit: bool = True,
):
    with db.cursor() as cur:
        cur.execute(
            '''INSERT INTO node_coverage (
                   node_id, geom, strength_geoms, antenna_height_m, radius_m,
                   model_version, calculation_status, permanent_reason,
                   last_error, retry_after
               )
               VALUES (%s, %s::jsonb, %s::jsonb, %s, %s, %s, 'computed', NULL, NULL, NULL)
               ON CONFLICT (node_id) DO UPDATE
                 SET geom = EXCLUDED.geom,
                     strength_geoms = EXCLUDED.strength_geoms,
                     antenna_height_m = EXCLUDED.antenna_height_m,
                     radius_m = EXCLUDED.radius_m,
                     model_version = EXCLUDED.model_version,
                     calculation_status = 'computed',
                     permanent_reason = NULL,
                     last_error = NULL,
                     retry_after = NULL,
                     calculated_at = NOW()''',
            (node_id, json.dumps(geom), json.dumps(strength_geoms), antenna_height_m, radius_m, COVERAGE_MODEL_VERSION),
        )
        cur.execute(
            'UPDATE nodes SET elevation_m = %s WHERE node_id = %s',
            (round(elevation_m, 1), node_id),
        )
    if commit:
        db.commit()


def store_permanent_coverage(
    db,
    node_id: str,
    reason: str,
    *,
    antenna_height_m: float = ANTENNA_HEIGHT_M,
    is_planned: bool = False,
) -> None:
    safe_reason = str(reason).strip()[:240] or 'unsupported'
    with db.cursor() as cur:
        cur.execute(
            '''INSERT INTO node_coverage (
                   node_id, geom, strength_geoms, antenna_height_m, radius_m,
                   model_version, calculation_status, permanent_reason,
                   last_error, retry_after, is_planned
               )
               VALUES (
                   %s, '{"type":"Polygon","coordinates":[]}'::jsonb, NULL,
                   %s, NULL, %s, 'permanent', %s, NULL, NULL, %s
               )
               ON CONFLICT (node_id) DO UPDATE
                 SET geom = EXCLUDED.geom,
                     strength_geoms = NULL,
                     antenna_height_m = EXCLUDED.antenna_height_m,
                     radius_m = NULL,
                     model_version = EXCLUDED.model_version,
                     calculation_status = 'permanent',
                     permanent_reason = EXCLUDED.permanent_reason,
                     last_error = NULL,
                     retry_after = NULL,
                     is_planned = EXCLUDED.is_planned,
                     calculated_at = NOW()''',
            (
                node_id,
                antenna_height_m,
                COVERAGE_MODEL_VERSION,
                safe_reason,
                is_planned,
            ),
        )
    db.commit()

def backfill_elevations(db):
    """Legacy placeholder.

    Elevation must come from sampled terrain at the node coordinates. Reversing
    stored radius_m is unsafe because coverage radii may be capped, support
    limited, or model-adjusted.
    """
    return

def load_positioned_repeaters(db) -> dict[str, dict]:
    with db.cursor() as cur:
        cur.execute(
            '''
            SELECT n.node_id, n.lat, n.lon, n.elevation_m, n.name, n.role, nc.radius_m, nc.antenna_height_m
            FROM nodes n
            LEFT JOIN node_coverage nc ON nc.node_id = n.node_id
            WHERE n.lat IS NOT NULL
              AND n.lon IS NOT NULL
              AND n.lat BETWEEN %s AND %s
              AND n.lon BETWEEN %s AND %s
              AND NOT (ABS(n.lat) < 1e-9 AND ABS(n.lon) < 1e-9)
              AND (n.name IS NULL OR n.name NOT LIKE %s)
              AND (n.role IS NULL OR n.role = 2)
            ''',
            (UK_LAT_MIN, UK_LAT_MAX, UK_LON_MIN, UK_LON_MAX, '%🚫%'),
        )
        return {
            row[0]: {
                'lat': row[1],
                'lon': row[2],
                'elevation_m': row[3],
                'name': row[4],
                'role': row[5],
                'radius_m': row[6],
                'antenna_height_m': row[7] if row[7] is not None else ANTENNA_HEIGHT_M,
            }
            for row in cur.fetchall()
        }


def link_pair_key(a_id: str, b_id: str) -> tuple[str, str]:
    return (a_id, b_id) if a_id < b_id else (b_id, a_id)


def refresh_link_topology(db, force: bool = False) -> None:
    now = time.time()
    if not force and now - float(LINK_TOPOLOGY['updated_at']) < LINK_TOPOLOGY_REFRESH_S:
        return

    nodes = load_positioned_repeaters(db)
    with db.cursor() as cur:
        cur.execute(
            'SELECT node_a_id, node_b_id FROM node_links WHERE itm_viable = true OR force_viable = true',
        )
        physical_pairs = {link_pair_key(a_id, b_id) for a_id, b_id in cur.fetchall()}

    LINK_TOPOLOGY['nodes'] = nodes
    LINK_TOPOLOGY['physical_pairs'] = physical_pairs
    LINK_TOPOLOGY['updated_at'] = now


def get_link_topology(db, required_node_ids: tuple[str, ...] = ()) -> tuple[dict[str, dict], set[tuple[str, str]]]:
    refresh_link_topology(db)
    nodes = LINK_TOPOLOGY['nodes']
    if any(node_id and node_id not in nodes for node_id in required_node_ids):
        refresh_link_topology(db, force=True)
        nodes = LINK_TOPOLOGY['nodes']
    return nodes, LINK_TOPOLOGY['physical_pairs']


# Cap how many nearby repeaters a planned repeater is path-loss tested against,
# to bound the synchronous compute performed inside a single viewshed job.
MAX_PLANNED_LINK_PEERS = max(
    1,
    min(200, int(os.environ.get('MAX_PLANNED_LINK_PEERS', '60'))),
)


def compute_planned_links(db, lat: float, lon: float, elevation_m: float,
                          radius_m: Optional[float],
                          antenna_height_m: float = ANTENNA_HEIGHT_M,
                          *,
                          stage_seconds: Optional[dict[str, float]] = None,
                          deadline_monotonic: Optional[float] = None,
                          cancellation_check: Optional[Callable[[], None]] = None,
                          ) -> list[dict]:
    """Predict viable RF links from a hypothetical repeater at (lat, lon) to nearby
    real positioned repeaters, using the same ITM path-loss model as physical links.

    Returns a list of {peer_id, peer_name, itm_path_loss_db, itm_viable, distance_km}
    for peers the planned repeater would reach, ordered best (lowest loss) first.
    """
    check_job_budget(deadline_monotonic, cancellation_check)
    timings = stage_seconds if stage_seconds is not None else {}
    origin = {'lat': lat, 'lon': lon, 'radius_m': radius_m, 'elevation_m': elevation_m}
    origin_radius_km = physical_candidate_radius_km(radius_m)

    peer_lookup_started = time.monotonic()
    all_nodes, _physical_pairs = get_link_topology(db)
    support_tree: Optional[cKDTree] = SUPPORT_CONTEXT['tree']
    support_node_ids: list[str] = SUPPORT_CONTEXT['node_ids']
    if support_tree is not None and support_node_ids:
        origin_xy = project_xy_km([lat], [lon])[0]
        nearby_indices = support_tree.query_ball_point(
            origin_xy,
            # The tree uses one UK reference latitude. Over-query so its
            # equirectangular longitude scale cannot exclude a true candidate;
            # node_dist_km applies the exact bounded filter below.
            r=MAX_PHYSICAL_LINK_RADIUS_KM * 1.35,
        )
        candidate_ids = [
            support_node_ids[int(index)]
            for index in nearby_indices
            if int(index) < len(support_node_ids)
        ]
    else:
        candidate_ids = list(all_nodes)

    candidates: list[tuple[float, str, dict]] = []
    for peer_id in candidate_ids:
        peer = all_nodes.get(peer_id)
        if peer is None:
            continue
        if peer.get('lat') is None or peer.get('lon') is None:
            continue
        dist_km = node_dist_km(origin, peer)
        reach_km = max(origin_radius_km, physical_candidate_radius_km(peer.get('radius_m')))
        if dist_km > reach_km:
            continue
        candidates.append((dist_km, peer_id, peer))
    candidates.sort(key=lambda c: c[0])
    candidates = candidates[:MAX_PLANNED_LINK_PEERS]
    timings['planned_peer_lookup'] = round(
        time.monotonic() - peer_lookup_started,
        6,
    )
    check_job_budget(deadline_monotonic, cancellation_check)
    if not candidates:
        return []

    links: list[dict] = []
    with tempfile.TemporaryDirectory() as tmp:
        tile_started = time.monotonic()
        terrain_paths: dict[str, Path] = {}
        terrain_candidates: list[tuple[float, str, dict]] = []
        for dist_km, peer_id, peer in candidates:
            check_job_budget(deadline_monotonic, cancellation_check)
            try:
                paths = ensure_tiles_for_link(
                    SRTM_DIR,
                    lat,
                    lon,
                    peer['lat'],
                    peer['lon'],
                    log,
                )
                for path in paths:
                    terrain_paths[str(path)] = path
                terrain_candidates.append((dist_km, peer_id, peer))
            except PermanentOutOfScope as exc:
                log.info(f'Planned link {peer_id[:8]}… is outside terrain scope: {exc}')
        timings['planned_tile_acquisition'] = round(
            time.monotonic() - tile_started,
            6,
        )
        if not terrain_candidates:
            return []

        check_job_budget(deadline_monotonic, cancellation_check)
        vrt_started = time.monotonic()
        vrt = f'{tmp}/planned-links.vrt'
        build_vrt(terrain_paths.values(), vrt)
        timings['planned_vrt_construction'] = round(
            time.monotonic() - vrt_started,
            6,
        )

        link_started = time.monotonic()
        for dist_km, peer_id, peer in terrain_candidates:
            check_job_budget(deadline_monotonic, cancellation_check)
            try:
                peer_elev = peer.get('elevation_m')
                if peer_elev is None:
                    peer_elev = sample_elevation(vrt, peer['lat'], peer['lon'])
                result = compute_path_loss(
                    lat, lon, elevation_m,
                    peer['lat'], peer['lon'], peer_elev,
                    vrt,
                    antenna_height_m_tx=antenna_height_m,
                    antenna_height_m_rx=peer.get('antenna_height_m', ANTENNA_HEIGHT_M),
                )
                if not result.viable:
                    continue
                links.append({
                    'peer_id': peer_id,
                    'peer_name': peer.get('name'),
                    'itm_path_loss_db': round(float(result.path_loss_db), 1),
                    'itm_viable': True,
                    'distance_km': round(dist_km, 1),
                })
            except Exception as exc:
                raise RetryableTerrainError(
                    f'planned link {peer_id[:8]} path loss failed: {exc}'
                ) from exc
        timings['planned_per_link_calculation'] = round(
            time.monotonic() - link_started,
            6,
        )
    check_job_budget(deadline_monotonic, cancellation_check)
    links.sort(key=lambda link: link['itm_path_loss_db'])
    return links


def store_planned_links(db, node_id: str, links: list[dict]) -> None:
    with db.cursor() as cur:
        cur.execute(
            'UPDATE node_coverage SET predicted_links = %s::jsonb WHERE node_id = %s',
            (json.dumps(links), node_id),
        )
    db.commit()


def publish_link_update(r_client, a_id: str, b_id: str, obs_count: int, path_loss_db: Optional[float],
                        itm_viable: Optional[bool], count_a_to_b: int, count_b_to_a: int,
                        multibyte_obs: int) -> None:
    r_client.publish(LIVE_CHANNEL, json.dumps({
        'type': 'link_update',
        'data': {
            'node_a_id': a_id,
            'node_b_id': b_id,
            'observed_count': obs_count,
            'itm_path_loss_db': path_loss_db,
            'itm_viable': itm_viable,
            'count_a_to_b': count_a_to_b,
            'count_b_to_a': count_b_to_a,
            'multibyte_observed_count': multibyte_obs,
        },
        'ts': int(time.time() * 1000),
    }))


def upsert_link_pair(db, a_id: str, b_id: str, inc_atob: int, inc_btoa: int, inc_multibyte: int):
    obs_delta = inc_atob + inc_btoa
    with db.cursor() as cur:
        cur.execute(
            '''INSERT INTO node_links
                   (node_a_id, node_b_id, observed_count, last_observed,
                    count_a_to_b, count_b_to_a, multibyte_observed_count)
               VALUES (%s, %s, %s, NOW(), %s, %s, %s)
               ON CONFLICT (node_a_id, node_b_id) DO UPDATE
                 SET observed_count = node_links.observed_count + %s,
                     last_observed = CASE WHEN %s > 0 THEN NOW() ELSE node_links.last_observed END,
                     count_a_to_b = node_links.count_a_to_b + %s,
                     count_b_to_a = node_links.count_b_to_a + %s,
                     multibyte_observed_count = node_links.multibyte_observed_count + %s
               RETURNING observed_count, itm_computed_at, itm_path_loss_db, itm_viable,
                         count_a_to_b, count_b_to_a, multibyte_observed_count''',
            (
                a_id, b_id, obs_delta, inc_atob, inc_btoa, inc_multibyte,
                obs_delta, obs_delta, inc_atob, inc_btoa, inc_multibyte,
            ),
        )
        return cur.fetchone()


def ensure_physical_link_metrics(db, a_id: str, a: dict, b_id: str, b: dict):
    # Existing physical-pair jobs are read-only until a metric is actually
    # missing. Avoid the old zero-delta ON CONFLICT UPDATE, while retaining a
    # race-safe insert-and-reread path for a genuinely absent pair.
    select_pair_sql = '''SELECT observed_count, itm_computed_at, itm_path_loss_db, itm_viable,
                                count_a_to_b, count_b_to_a, multibyte_observed_count
                         FROM node_links
                         WHERE node_a_id = %s AND node_b_id = %s'''
    with db.cursor() as cur:
        cur.execute(select_pair_sql, (a_id, b_id))
        row = cur.fetchone()
        if row is None:
            cur.execute(
                '''INSERT INTO node_links
                     (node_a_id, node_b_id, observed_count, last_observed,
                      count_a_to_b, count_b_to_a, multibyte_observed_count)
                   VALUES (%s, %s, 0, NOW(), 0, 0, 0)
                   ON CONFLICT (node_a_id, node_b_id) DO NOTHING
                   RETURNING observed_count, itm_computed_at, itm_path_loss_db, itm_viable,
                             count_a_to_b, count_b_to_a, multibyte_observed_count''',
                (a_id, b_id),
            )
            row = cur.fetchone()
            if row is None:
                cur.execute(select_pair_sql, (a_id, b_id))
                row = cur.fetchone()
    obs_count = row[0] if row else 0
    itm_computed = row[1] if row else None
    path_loss_db = row[2] if row else None
    itm_viable = row[3] if row else None
    count_a_to_b = row[4] if row else 0
    count_b_to_a = row[5] if row else 0
    multibyte_obs = row[6] if row else 0

    missing_endpoint_elev = a.get('elevation_m') is None or b.get('elevation_m') is None
    if itm_computed is not None and not missing_endpoint_elev:
        return obs_count, path_loss_db, itm_viable, count_a_to_b, count_b_to_a, multibyte_obs

    with tempfile.TemporaryDirectory() as tmp:
        ensure_tiles_for_link(
            SRTM_DIR,
            a['lat'],
            a['lon'],
            b['lat'],
            b['lon'],
            log,
        )
        vrt = build_link_vrt(a['lat'], a['lon'], b['lat'], b['lon'], tmp, SRTM_DIR)
        try:
            a_elev = a.get('elevation_m')
            b_elev = b.get('elevation_m')
            if a_elev is None:
                a_elev = sample_elevation(vrt, a['lat'], a['lon'])
                a['elevation_m'] = a_elev
                with db.cursor() as cur:
                    cur.execute(
                        'UPDATE nodes SET elevation_m = %s WHERE node_id = %s AND elevation_m IS NULL',
                        (round(a_elev, 1), a_id),
                    )
            if b_elev is None:
                b_elev = sample_elevation(vrt, b['lat'], b['lon'])
                b['elevation_m'] = b_elev
                with db.cursor() as cur:
                    cur.execute(
                        'UPDATE nodes SET elevation_m = %s WHERE node_id = %s AND elevation_m IS NULL',
                        (round(b_elev, 1), b_id),
                    )
            result = compute_path_loss(
                a['lat'], a['lon'], a_elev,
                b['lat'], b['lon'], b_elev,
                vrt,
                antenna_height_m_tx=a.get('antenna_height_m', ANTENNA_HEIGHT_M),
                antenna_height_m_rx=b.get('antenna_height_m', ANTENNA_HEIGHT_M),
            )
            path_loss_db = round(result.path_loss_db, 1)
            itm_viable = result.viable
            with db.cursor() as cur:
                cur.execute(
                    '''UPDATE node_links
                       SET itm_path_loss_db = %s,
                           itm_viable = %s,
                           itm_computed_at = NOW(),
                           terrain_profile_json = %s::jsonb
                       WHERE node_a_id = %s AND node_b_id = %s''',
                    (path_loss_db, itm_viable, json.dumps(result.profile), a_id, b_id),
                )
            log.info(
                f'Link {a_id[:8]}…↔{b_id[:8]}…: '
                f'{path_loss_db:.1f} dB {"✓" if itm_viable else "✗"} '
                f'(obs={obs_count})'
            )
        except (PermanentOutOfScope, InvalidTerrainRequest):
            raise
        except Exception as exc:
            raise RetryableTerrainError(f'path loss computation failed: {exc}') from exc

    return obs_count, path_loss_db, itm_viable, count_a_to_b, count_b_to_a, multibyte_obs


def process_physical_link_job(db, r_client, job: dict):
    node_a_id = job.get('node_a_id')
    node_b_id = job.get('node_b_id')
    if not node_a_id or not node_b_id or node_a_id == node_b_id:
        return

    nodes, physical_pairs = get_link_topology(db, (node_a_id, node_b_id))
    a = nodes.get(node_a_id)
    b = nodes.get(node_b_id)
    if not a or not b:
        return

    obs_count, path_loss_db, itm_viable, count_a_to_b, count_b_to_a, multibyte_obs = ensure_physical_link_metrics(
        db, node_a_id, a, node_b_id, b,
    )
    if itm_viable:
        physical_pairs.add(link_pair_key(node_a_id, node_b_id))
    publish_link_update(r_client, node_a_id, node_b_id, obs_count, path_loss_db, itm_viable, count_a_to_b, count_b_to_a, multibyte_obs)


def process_observation_link_job(db, r_client, job: dict):
    """Resolve multibyte packet paths and annotate links.

    When both endpoints of a hop are uniquely resolved (exactly one node
    matched the hash prefix), the link is upserted even if it has no prior
    physical-link entry — multibyte evidence is precise enough to bootstrap
    new links directly.  Ambiguously-resolved hops retain the conservative
    behaviour of only annotating pre-existing physical links.
    """
    rx_node_id = job.get('rx_node_id')
    src_node_id = job.get('src_node_id')
    path_hashes = job.get('path_hashes', [])
    path_hash_size_bytes = int(job.get('path_hash_size_bytes') or 1)

    if not rx_node_id or not path_hashes or path_hash_size_bytes <= 1:
        return

    all_nodes, physical_pairs = get_link_topology(db, (rx_node_id,))

    def physical_link(a_id: str, b_id: str) -> bool:
        return link_pair_key(a_id, b_id) in physical_pairs

    rx = all_nodes.get(rx_node_id)
    if not rx:
        return

    def normalize_path_hash(value) -> str:
        return str(value or '').strip().upper()

    def node_matches_path_hash(node_id: str, path_hash: str) -> bool:
        return bool(path_hash) and node_id.upper().startswith(path_hash)

    def local_prefix_ambiguity_penalty(path_hash: str, target_id: str, target_node: dict, anchor_node: dict,
                                       pool: list[tuple[str, dict]]) -> float:
        target_dist = node_dist_km(target_node, anchor_node)
        raw = 0.0
        for cand_id, cand_node in pool:
            if cand_id == target_id:
                continue
            if not node_matches_path_hash(cand_id, path_hash):
                continue
            cand_dist = node_dist_km(cand_node, anchor_node)
            if cand_dist > PREFIX_AMBIGUITY_RADIUS_KM:
                continue
            dist_similarity = max(0.0, 1.0 - abs(cand_dist - target_dist) / PREFIX_AMBIGUITY_RADIUS_KM)
            proximity = max(0.0, 1.0 - cand_dist / PREFIX_AMBIGUITY_RADIUS_KM)
            raw += dist_similarity * proximity
        return min(0.24, raw * 0.12)

    # Each entry: (node_id, node_data, is_unique)
    # is_unique=True means exactly one node matched the hash — high-confidence.
    resolved: list[tuple[str, dict, bool]] = []
    prev_id = rx_node_id
    prev = rx
    visited = {rx_node_id}

    for raw_hash in reversed(path_hashes):
        path_hash = normalize_path_hash(raw_hash)
        if not path_hash:
            continue
        candidates = [
            (nid, nd) for nid, nd in all_nodes.items()
            if node_matches_path_hash(nid, path_hash)
            and nid not in visited
        ]
        if not candidates:
            continue

        is_unique = len(candidates) == 1

        best_id = None
        best = None
        best_score = float('-inf')
        for nid, nd in candidates:
            confirmed_bonus = 2.5 if physical_link(nid, prev_id) else 0.0
            distance_score = -node_dist_km(nd, prev) / 12.0
            ambiguity_penalty = local_prefix_ambiguity_penalty(path_hash, nid, nd, prev, candidates)
            score = confirmed_bonus + distance_score - ambiguity_penalty
            if score > best_score:
                best_score = score
                best_id, best = nid, nd

        if best_id is None or best is None:
            continue

        resolved.insert(0, (best_id, best, is_unique))
        visited.add(best_id)
        prev_id = best_id
        prev = best

    # rx_node_id is always uniquely known (it's the observer), so mark it unique.
    # src_node_id from the packet header is also authoritative if present.
    full: list[tuple[str, dict, bool]] = []
    if src_node_id and src_node_id in all_nodes:
        full.append((src_node_id, all_nodes[src_node_id], True))
    full.extend(resolved)
    full.append((rx_node_id, rx, True))
    if len(full) < 2:
        return

    for i in range(len(full) - 1):
        src_id, src, src_unique = full[i]
        dst_id, dst, dst_unique = full[i + 1]
        if src_id == dst_id:
            continue
        if src['lat'] is None or src['lon'] is None or dst['lat'] is None or dst['lon'] is None:
            continue

        if src_id < dst_id:
            a_id, a, b_id, b = src_id, src, dst_id, dst
            inc_atob, inc_btoa = 1, 0
        else:
            a_id, a, b_id, b = dst_id, dst, src_id, src
            inc_atob, inc_btoa = 0, 1

        # Allow new link creation only when both endpoints were uniquely resolved.
        # For ambiguous hops, stay conservative and only annotate existing links.
        both_unique = src_unique and dst_unique
        if not both_unique and not physical_link(a_id, b_id):
            continue

        # Keep directional observations, but reserve multibyte evidence for exact
        # endpoint matches so ambiguous two-byte prefixes do not inflate path history.
        row = upsert_link_pair(db, a_id, b_id, inc_atob, inc_btoa, 1 if both_unique else 0)
        obs_count = row[0] if row else 1
        path_loss_db = row[2] if row else None
        itm_viable = row[3] if row else None
        count_a_to_b = row[4] if row else inc_atob
        count_b_to_a = row[5] if row else inc_btoa
        multibyte_obs = row[6] if row else 1

        if itm_viable is None:
            obs_count, path_loss_db, itm_viable, count_a_to_b, count_b_to_a, multibyte_obs = ensure_physical_link_metrics(
                db, a_id, a, b_id, b,
            )
        if itm_viable:
            physical_pairs.add(link_pair_key(a_id, b_id))
        publish_link_update(r_client, a_id, b_id, obs_count, path_loss_db, itm_viable, count_a_to_b, count_b_to_a, multibyte_obs)


def process_link_job(db, r_client, job: dict):
    job_type = str(job.get('type') or 'observe').strip().lower()
    if job_type == 'physical_pair':
        process_physical_link_job(db, r_client, job)
        return
    process_observation_link_job(db, r_client, job)


def redis_connection():
    return redis.Redis.from_url(
        REDIS_URL,
        decode_responses=True,
        password=REDIS_PASSWORD,
    )


def process_claimed_link_job(db, r_client, claimed) -> None:
    job_id, token, job, _attempt = claimed
    try:
        with link_queue_v3.LeaseRenewer(redis_connection, job_id, token) as renewer:
            db.autocommit = False
            try:
                with db.cursor() as cur:
                    cur.execute(
                        'SELECT 1 FROM public.link_job_commits WHERE job_id = %s',
                        (job_id,),
                    )
                    already_committed = cur.fetchone() is not None
                    if not already_committed:
                        generation = str(job.get('generation') or '').strip()
                        if generation:
                            generation_suffix = generation.removeprefix('link_rebuild_')
                            if (
                                not generation.startswith('link_rebuild_')
                                or len(generation_suffix) != 16
                                or not all(ch in '0123456789abcdef' for ch in generation_suffix)
                            ):
                                raise ValueError('Invalid link rebuild generation')
                            cur.execute(
                                sql.SQL('SET LOCAL search_path = {}, public').format(
                                    sql.Identifier(generation),
                                )
                            )
                            # Physical-pair membership is graph-generation
                            # specific; never reuse the process-global public
                            # cache while building a staging graph.
                            LINK_TOPOLOGY['updated_at'] = 0.0
                        cur.execute(
                            "SELECT pg_advisory_xact_lock(hashtext('meshcore-link-graph-write'))"
                        )
                        process_link_job(db, r_client, job)
                        renewer.assert_owned()
                        if not link_queue_v3.owns_lease(r_client, job_id, token):
                            raise RuntimeError(f'LINK_LEASE_LOST:{job_id}')
                        payload_hash = hashlib.sha256(
                            json.dumps(job, separators=(',', ':'), sort_keys=True).encode()
                        ).hexdigest()
                        cur.execute(
                            '''INSERT INTO public.link_job_commits
                                 (job_id, job_type, generation, logical_job_id, payload_hash)
                               VALUES (%s, %s, %s, %s, %s)
                               ON CONFLICT (job_id) DO NOTHING''',
                            (
                                job_id,
                                str(job.get('type') or 'observe'),
                                generation or None,
                                job.get('logical_job_id'),
                                payload_hash,
                            ),
                        )
                db.commit()
            except Exception:
                db.rollback()
                raise
            finally:
                db.autocommit = True
                if job.get('generation'):
                    LINK_TOPOLOGY['updated_at'] = 0.0
        if not link_queue_v3.ack(r_client, job_id, token):
            log.warning(f'Link v3 ACK rejected for {job_id}')
        record_job('link', 'job', 'success')
    except Exception as exc:
        result = link_queue_v3.nack(r_client, job_id, token, type(exc).__name__)
        record_job('link', 'job', 'failure')
        log.exception(f'Link v3 job {job_id} failed; transition={result}')
        raise


def enqueue_physical_link_jobs_for_node(db, r_client, node_id: str, lat: float, lon: float, radius_m: Optional[float]) -> int:
    origin = {'lat': lat, 'lon': lon, 'radius_m': radius_m}
    origin_radius_km = physical_candidate_radius_km(radius_m)
    with db.cursor() as cur:
        cur.execute(
            '''
            SELECT n.node_id, n.lat, n.lon, nc.radius_m
            FROM nodes n
            LEFT JOIN node_coverage nc ON nc.node_id = n.node_id
            WHERE n.node_id <> %s
              AND n.lat IS NOT NULL
              AND n.lon IS NOT NULL
              AND n.lat BETWEEN %s AND %s
              AND n.lon BETWEEN %s AND %s
              AND NOT (ABS(n.lat) < 1e-9 AND ABS(n.lon) < 1e-9)
              AND (n.name IS NULL OR n.name NOT LIKE %s)
              AND (n.role IS NULL OR n.role = 2)
            ''',
            (node_id, UK_LAT_MIN, UK_LAT_MAX, UK_LON_MIN, UK_LON_MAX, '%🚫%'),
        )
        rows = cur.fetchall()

    queued = 0
    for peer_id, peer_lat, peer_lon, peer_radius_m in rows:
        peer = {'lat': peer_lat, 'lon': peer_lon, 'radius_m': peer_radius_m}
        if node_dist_km(origin, peer) > max(origin_radius_km, physical_candidate_radius_km(peer_radius_m)):
            continue
        [a_id, b_id] = sorted((node_id, peer_id))
        status, _ = link_queue_v3.admit_physical(r_client, a_id, b_id)
        if status in ('accepted', 'coalesced', 'duplicate'):
            queued += 1
        else:
            log.warning(f'Physical link job not admitted for {a_id[:8]}…↔{b_id[:8]}…: {status}')
    return queued


def enqueue_uncovered(db, r_client):
    """On startup, queue all nodes that have a position but no coverage yet."""
    # Remove any coverage that was previously computed for hidden or non-repeater nodes.
    with db.cursor() as cur:
        cur.execute("""
            DELETE FROM node_coverage WHERE node_id IN (
                SELECT node_id FROM nodes
                WHERE name LIKE '%🚫%' OR (role IS NOT NULL AND role != 2)
            )
        """)
    db.commit()

    with db.cursor() as cur:
        cur.execute('''
            SELECT n.node_id, n.lat, n.lon
            FROM nodes n
            LEFT JOIN node_coverage nc ON n.node_id = nc.node_id
            WHERE n.lat IS NOT NULL AND n.lon IS NOT NULL
              AND n.lat BETWEEN %s AND %s
              AND n.lon BETWEEN %s AND %s
              AND NOT (ABS(n.lat) < 1e-9 AND ABS(n.lon) < 1e-9)
              AND (nc.node_id IS NULL OR nc.model_version < %s OR n.elevation_m IS NULL)
              AND (n.name IS NULL OR n.name NOT LIKE %s)
              AND (n.role IS NULL OR n.role = 2)
        ''', (UK_LAT_MIN, UK_LAT_MAX, UK_LON_MIN, UK_LON_MAX, COVERAGE_MODEL_VERSION, '%🚫%',))
        rows = cur.fetchall()
    if rows:
        log.info(f'Queuing {len(rows)} existing node(s) for viewshed calculation (model v{COVERAGE_MODEL_VERSION})')
        for node_id, lat, lon in rows:
            status, _ = viewshed_queue_v2.admit(
                r_client,
                {
                    'version': 2,
                    'node_id': node_id,
                    'lat': lat,
                    'lon': lon,
                    'model_version': COVERAGE_MODEL_VERSION,
                },
            )
            if status not in ('accepted', 'coalesced', 'dead'):
                log.warning(f'Coverage startup job {node_id[:12]}… was not admitted: {status}')

def rebuild_pending_viewshed_set(r_client):
    """Rebuild the dedupe set from v2 state plus any legacy queue entries."""
    r_client.delete(JOB_PENDING_SET)
    node_ids = [
        str(job_id)
        for job_id, state in r_client.hgetall(viewshed_queue_v2.STATES).items()
        if state in ('queued', 'in_flight', 'dead')
    ]
    raw_jobs = r_client.lrange(JOB_QUEUE, 0, -1)
    for raw in raw_jobs:
        try:
            job = json.loads(raw)
        except Exception:
            continue
        node_id = str(job.get('node_id') or '').strip()
        if node_id:
            node_ids.append(node_id)
    if node_ids:
        r_client.sadd(JOB_PENDING_SET, *node_ids)
    log.info(f'Rebuilt viewshed pending set from queue ({len(set(node_ids))} unique node(s))')


def recover_planned_coverage_jobs(db, r_client) -> int:
    available = max(0, PLANNED_COVERAGE_QUEUE_MAX - int(r_client.llen(JOB_QUEUE)))
    with db.cursor() as cur:
        cur.execute('DELETE FROM planned_coverage_handles WHERE expires_at <= NOW()')
        cur.execute(
            '''SELECT job_id, lat, lon
                 FROM planned_coverage_jobs
                WHERE (
                        status = 'queued'
                        OR (status = 'running' AND heartbeat_at < NOW() - INTERVAL '60 seconds')
                      )
                  AND expires_at > NOW()
                ORDER BY created_at
                LIMIT %s''',
            (available,),
        )
        rows = cur.fetchall()
        cur.execute(
            '''UPDATE planned_coverage_jobs
                  SET status = 'queued', heartbeat_at = NULL, updated_at = NOW()
                WHERE status = 'running'
                  AND heartbeat_at < NOW() - INTERVAL '60 seconds'
                  AND expires_at > NOW()'''
        )
        cur.execute(
            '''DELETE FROM node_coverage coverage
                USING planned_coverage_jobs jobs
               WHERE coverage.node_id = jobs.job_id
                 AND coverage.is_planned = TRUE
                 AND jobs.expires_at <= NOW()'''
        )
        cur.execute(
            '''DELETE FROM planned_coverage_jobs jobs
                WHERE jobs.expires_at <= NOW()
                  AND NOT EXISTS (
                    SELECT 1 FROM planned_coverage_handles handles
                     WHERE handles.job_id = jobs.job_id
                  )'''
        )
    db.commit()
    queued = 0
    for job_id, lat, lon in rows:
        status, _ = viewshed_queue_v2.admit(
            r_client,
            {
                'version': 2,
                'node_id': job_id,
                'planned_job_id': job_id,
                'lat': lat,
                'lon': lon,
                'model_version': COVERAGE_MODEL_VERSION,
            },
        )
        if status in ('accepted', 'coalesced'):
            queued += 1
    if queued:
        log.warning(f'Recovered {queued} durable planned coverage job(s)')
    return queued


class PlannedJobHeartbeat:
    def __init__(self, job_id: str):
        self.job_id = job_id
        self.stop_event = threading.Event()
        self.thread = threading.Thread(target=self._run, name=f'planned-{job_id[:8]}', daemon=True)

    def _renew(self, conn) -> None:
        with conn.cursor() as cur:
            cur.execute(
                '''UPDATE planned_coverage_jobs
                      SET heartbeat_at = NOW(),
                          expires_at = GREATEST(expires_at, NOW() + (%s * INTERVAL '1 second')),
                          updated_at = NOW()
                    WHERE job_id = %s AND status = 'running' ''',
                (PLANNED_RESULT_TTL_S, self.job_id),
            )
            cur.execute(
                '''UPDATE planned_coverage_handles
                      SET expires_at = GREATEST(expires_at, NOW() + (%s * INTERVAL '1 second'))
                    WHERE job_id = %s''',
                (PLANNED_RESULT_TTL_S, self.job_id),
            )

    def _run(self):
        conn = None
        while not self.stop_event.wait(15):
            try:
                if conn is None or conn.closed:
                    conn = wait_for_db()
                self._renew(conn)
            except Exception as exc:
                log.warning(f'Planned job heartbeat reconnecting: {exc}')
                if conn is not None:
                    conn.close()
                conn = None
        if conn is not None:
            conn.close()

    def __enter__(self):
        self.thread.start()
        return self

    def __exit__(self, exc_type, exc, tb):
        self.stop_event.set()
        self.thread.join(timeout=5)


def process_planned_job(
    db,
    r_client,
    job: dict,
    *,
    cancellation_check: Optional[Callable[[], None]] = None,
) -> None:
    job_id = str(job['planned_job_id'])
    check_job_budget(None, cancellation_check)
    with db.cursor() as cur:
        cur.execute(
            '''UPDATE planned_coverage_jobs
                  SET status = 'running', heartbeat_at = NOW(), error = NULL, updated_at = NOW()
                WHERE job_id = %s AND expires_at > NOW()
              RETURNING job_id''',
            (job_id,),
        )
        if cur.fetchone() is None:
            return
    db.commit()
    try:
        with PlannedJobHeartbeat(job_id):
            process_job(
                db,
                r_client,
                job,
                cancellation_check=cancellation_check,
            )
        check_job_budget(None, cancellation_check)
        expires_at = dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=PLANNED_RESULT_TTL_S)
        with db.cursor() as cur:
            cur.execute(
                '''UPDATE planned_coverage_jobs
                      SET status = 'ready', heartbeat_at = NOW(), expires_at = %s,
                          error = NULL, updated_at = NOW()
                    WHERE job_id = %s''',
                (expires_at, job_id),
            )
            cur.execute(
                '''UPDATE planned_coverage_handles SET expires_at = %s WHERE job_id = %s''',
                (expires_at, job_id),
            )
            cur.execute(
                '''UPDATE node_coverage
                      SET is_planned = TRUE, expires_at = %s
                    WHERE node_id = %s''',
                (expires_at, job_id),
            )
        db.commit()
    except Cancelled:
        try:
            db.rollback()
        except Exception:
            pass
        raise
    except PermanentOutOfScope as exc:
        with db.cursor() as cur:
            cur.execute(
                '''UPDATE planned_coverage_jobs
                      SET status = 'permanent', error = %s, heartbeat_at = NOW(), updated_at = NOW()
                    WHERE job_id = %s''',
                (str(exc)[:500], job_id),
            )
        db.commit()
        log.info(f'Planned coverage job {job_id} has a permanent outcome: {exc}')
    except Exception as exc:
        with db.cursor() as cur:
            cur.execute(
                '''UPDATE planned_coverage_jobs
                      SET status = 'failed', error = %s, heartbeat_at = NOW(), updated_at = NOW()
                    WHERE job_id = %s''',
                (str(exc)[:500], job_id),
            )
        db.commit()
        raise

# ── Job processor ─────────────────────────────────────────────────────────────

RELEASE_VIEWSHED_PAYLOAD_SCRIPT = """
local current = redis.call('HGET', KEYS[1], ARGV[1])
if current and current == ARGV[2] then
  redis.call('HDEL', KEYS[1], ARGV[1])
  local bytes = math.max(0, tonumber(redis.call('HGET', KEYS[2], 'bytes') or '0') - string.len(ARGV[2]))
  redis.call('HSET', KEYS[2], 'bytes', bytes)
  return bytes
end
return tonumber(redis.call('HGET', KEYS[2], 'bytes') or '0')
"""


def release_viewshed_queue_payload(r_client, job: dict, raw_payload: str) -> int:
    """Release producer-side byte accounting immediately after BRPOP."""
    node_id = str(job.get('node_id') or '')
    if not node_id:
        return 0
    return int(r_client.eval(
        RELEASE_VIEWSHED_PAYLOAD_SCRIPT,
        2,
        JOB_PAYLOADS,
        JOB_COUNTERS,
        node_id,
        raw_payload,
    ))

def process_job(
    db,
    r_client,
    job: dict,
    *,
    cancellation_check: Optional[Callable[[], None]] = None,
):
    job_started = time.monotonic()
    deadline_monotonic = job_started + COVERAGE_JOB_DEADLINE_SECONDS
    stage_seconds: dict[str, float] = {}
    check_job_budget(deadline_monotonic, cancellation_check)
    node_id = str(job.get('node_id') or '').strip()
    if not node_id or len(node_id) > 128:
        raise InvalidJob('node_id is required and must be at most 128 characters')
    is_planned = bool(node_id.startswith('plan_') or job.get('planned_job_id'))
    try:
        lat = float(job['lat'])
        lon = float(job['lon'])
    except (KeyError, TypeError, ValueError) as exc:
        raise InvalidJob('lat and lon must be finite numbers') from exc
    if not all(math.isfinite(value) for value in (lat, lon)):
        raise InvalidJob('lat and lon must be finite numbers')
    try:
        if not is_viewshed_eligible_coordinate(lat, lon):
            raise PermanentOutOfScope(
                f'coordinate outside UK coverage bounds at ({lat:.4f}, {lon:.4f})'
            )
        # Skip hidden (🚫) or non-repeater nodes regardless of how the job arrived
        with db.cursor() as cur:
            cur.execute(
                '''SELECT n.name, n.role, nc.antenna_height_m
                   FROM nodes n
                   LEFT JOIN node_coverage nc ON nc.node_id = n.node_id
                   WHERE n.node_id = %s''',
                (node_id,),
            )
            row = cur.fetchone()
        node_antenna_height_m = ANTENNA_HEIGHT_M
        if row:
            name, role, stored_antenna_h = row
            if stored_antenna_h is not None:
                node_antenna_height_m = float(stored_antenna_h)
            if name and '🚫' in name:
                log.info(f'Skipping hidden node {node_id[:12]}…')
                return
            if role is not None and role != 2:
                log.info(f'Skipping non-repeater {node_id[:12]}… (role={role})')
                return

        if already_calculated(db, node_id, require_predicted_links=is_planned):
            if not is_planned and not side_effects_complete(r_client, node_id):
                # BUG-006: coverage exists but the post-commit side effects
                # (link jobs + notifications) never completed — replay them
                # instead of silently skipping.
                replay_coverage_side_effects(db, r_client, node_id)
            log.info(f'Coverage already exists for {node_id[:12]}…, skipping')
            return

        log.info(f'Viewshed: {node_id[:12]}… at ({lat:.4f}, {lon:.4f})')
        result = calculate_viewshed(
            node_id,
            lat,
            lon,
            antenna_height_m=node_antenna_height_m,
            deadline_monotonic=deadline_monotonic,
            cancellation_check=cancellation_check,
        )
        stage_seconds.update(result.stage_seconds)
        geom = result.geom
        strength_geoms = result.strength_geoms
        radius_m = result.radius_m
        elevation_m = result.elevation_m
        predicted: Optional[list[dict]] = None
        if is_planned:
            # Planned (hypothetical) repeater: predict viable links to nearby real
            # repeaters in-line so the result is ready when the user polls coverage.
            predicted = compute_planned_links(
                db, lat, lon, elevation_m, radius_m, antenna_height_m=node_antenna_height_m,
                stage_seconds=stage_seconds,
                deadline_monotonic=deadline_monotonic,
                cancellation_check=cancellation_check,
            )
        check_job_budget(deadline_monotonic, cancellation_check)
        db_write_started = time.monotonic()
        store_coverage(
            db,
            node_id,
            geom,
            strength_geoms,
            radius_m,
            elevation_m,
            antenna_height_m=node_antenna_height_m,
            commit=not is_planned,
        )
        if is_planned:
            assert predicted is not None
            store_planned_links(db, node_id, predicted)
            log.info(f'Planned repeater {node_id}: {len(predicted)} predicted link(s)')
        elif WORKER_MODE in ('all', 'link'):
            queued_links = enqueue_physical_link_jobs_for_node(db, r_client, node_id, lat, lon, radius_m)
            if queued_links > 0:
                log.info(f'Queued {queued_links} physical link job(s) for {node_id[:12]}…')
        stage_seconds['database_write'] = round(
            time.monotonic() - db_write_started,
            6,
        )
        check_job_budget(deadline_monotonic, cancellation_check)
        duration_seconds = round(time.monotonic() - job_started, 6)
        log.info(
            'coverage_job_metrics %s',
            json.dumps(
                {
                    'node_id': node_id,
                    'is_planned': is_planned,
                    'model': COVERAGE_MODEL,
                    'model_version': COVERAGE_MODEL_VERSION,
                    'duration_seconds': duration_seconds,
                    'peak_rss_mb': max(result.peak_rss_mb, peak_rss_mb()),
                    'stage_seconds': stage_seconds,
                    'predicted_link_count': len(predicted or []),
                },
                sort_keys=True,
                separators=(',', ':'),
            ),
        )
        log.info(f'Done in {duration_seconds:.1f}s — notifying frontend')

        _publish_coverage_notifications(r_client, node_id, geom, strength_geoms, elevation_m)
        if not is_planned:
            # BUG-006: only after EVERY side effect succeeds do we record the
            # completion marker, so a crash/Redis failure here is detected on
            # retry and the side effects replayed instead of skipped forever.
            mark_side_effects_complete(r_client, node_id)
    except PermanentOutOfScope as exc:
        store_permanent_coverage(
            db,
            node_id,
            str(exc),
            is_planned=is_planned,
        )
        raise


def process_claimed_viewshed_job(db, r_client, claimed) -> None:
    job_id, token, job, attempt = claimed
    permanent_outcome: Optional[PermanentOutOfScope] = None
    try:
        with viewshed_queue_v2.LeaseRenewer(redis_connection, job_id, token) as renewer:
            try:
                if job.get('planned_job_id'):
                    process_planned_job(
                        db,
                        r_client,
                        job,
                        cancellation_check=renewer.assert_owned,
                    )
                else:
                    process_job(
                        db,
                        r_client,
                        job,
                        cancellation_check=renewer.assert_owned,
                    )
            except PermanentOutOfScope as exc:
                # process_job persists this explicit terminal result before
                # raising. It is therefore safe to ACK after the lease fence.
                permanent_outcome = exc
            renewer.assert_owned()
            if not viewshed_queue_v2.owns_lease(r_client, job_id, token):
                raise RuntimeError(f'VIEWSHED_LEASE_LOST:{job_id}')
        transition = viewshed_queue_v2.ack(r_client, job_id, token)
        if transition == 'invalid':
            raise RuntimeError(f'VIEWSHED_ACK_REJECTED:{job_id}')
        if permanent_outcome is not None:
            log.info(f'Coverage job {job_id[:12]}… completed permanently: {permanent_outcome}')
        elif transition == 'requeued':
            log.info(f'Coverage job {job_id[:12]}… coalesced a newer position and was requeued')
        record_job('viewshed', 'job', 'success')
    except InvalidJob:
        try:
            db.rollback()
        except Exception:
            pass
        transition = viewshed_queue_v2.nack(
            r_client,
            job_id,
            token,
            permanent=True,
            reason='InvalidJob',
        )
        record_job('viewshed', 'job', 'dead')
        log.exception(
            f'Invalid coverage job {job_id[:12]}… moved to terminal state; '
            f'attempt={attempt} transition={transition}'
        )
    except Exception as exc:
        try:
            db.rollback()
        except Exception:
            pass
        transition = viewshed_queue_v2.nack(
            r_client,
            job_id,
            token,
            reason=type(exc).__name__,
        )
        record_job('viewshed', 'job', 'failure')
        log.exception(
            f'Coverage job {job_id[:12]}… failed; attempt={attempt} transition={transition}'
        )
        raise

# ── Main loop ─────────────────────────────────────────────────────────────────

def wait_for_db() -> psycopg2.extensions.connection:
    for attempt in range(30):
        try:
            conn = psycopg2.connect(DATABASE_URL, application_name=DB_APPLICATION_NAME)
            # autocommit=True prevents SELECT queries from holding open transactions
            # that would block schema DDL (CREATE EXTENSION etc.) on app restart.
            conn.autocommit = True
            conn.cursor().execute('SELECT 1')
            return conn
        except Exception:
            log.info(f'Waiting for DB… (attempt {attempt + 1}/30)')
            time.sleep(3)
    raise RuntimeError('DB never became ready')


def link_idle_wait_seconds(now: float, last_link_reap: float) -> float:
    """Block until the next maintenance deadline, capped by lease reaping."""
    deadlines = [
        5.0 - (now - last_link_reap),
        LINK_TOPOLOGY_REFRESH_S - (now - float(LINK_TOPOLOGY['updated_at'])),
        SUPPORT_REFRESH_S - (now - float(SUPPORT_CONTEXT['updated_at'])),
        CALIBRATION_REFRESH_S - (now - float(RF_CALIBRATION['updated_at'])),
    ]
    if RADIO_BOT_URL:
        deadlines.append(
            RADIO_NEIGHBOR_REFRESH_S - (now - float(RADIO_NEIGHBOR_SYNC['updated_at']))
        )
    return max(0.05, min(5.0, *deadlines))

def worker_loop():
    """Single worker process: owns its own DB and Redis connections."""
    name     = multiprocessing.current_process().name
    db       = wait_for_db()
    r_client = redis_connection()
    heartbeat_stop = threading.Event()
    if WORKER_MODE in ('all', 'link'):
        link_queue_v3.start_worker_heartbeat(redis_connection, heartbeat_stop)
    if WORKER_MODE in ('all', 'viewshed'):
        def viewshed_heartbeat():
            heartbeat_client = None
            while not heartbeat_stop.is_set():
                try:
                    if heartbeat_client is None:
                        heartbeat_client = redis_connection()
                    heartbeat_client.set(VIEWSHED_WORKER_HEARTBEAT, str(int(time.time())), ex=45)
                    heartbeat('viewshed')
                    heartbeat_stop.wait(10)
                except Exception as exc:
                    log.warning(f'Viewshed heartbeat reconnecting: {exc}')
                    if heartbeat_client is not None:
                        heartbeat_client.close()
                    heartbeat_client = None
                    heartbeat_stop.wait(2)
            if heartbeat_client is not None:
                heartbeat_client.close()
        threading.Thread(target=viewshed_heartbeat, name='viewshed-worker-heartbeat', daemon=True).start()
    sync_radio_neighbors = name in ('MainProcess', 'Worker-1')
    if sync_radio_neighbors and WORKER_MODE in ('all', 'link'):
        refresh_radio_neighbor_reports(db, r_client, force=True)
    refresh_rf_calibration(db, force=True)
    refresh_support_context(db, force=True)
    if WORKER_MODE in ('all', 'link'):
        refresh_link_topology(db, force=True)
    log.info(f'{name} ready')
    last_link_reap = 0.0
    last_viewshed_reap = 0.0
    last_planned_recovery = 0.0

    while True:
        try:
            if WORKER_MODE in ('link', 'viewshed'):
                heartbeat(WORKER_MODE)
            if WORKER_MODE == 'viewshed' and not VIEWSHED_ENABLED:
                time.sleep(300)
                continue
            if WORKER_MODE in ('all', 'viewshed') and VIEWSHED_ENABLED:
                now = time.time()
                if now - last_planned_recovery >= 30:
                    recover_planned_coverage_jobs(db, r_client)
                    last_planned_recovery = now
                if now - last_viewshed_reap >= 5:
                    recovered = viewshed_queue_v2.reap(r_client)
                    if recovered:
                        log.warning(f'Recovered {recovered} expired coverage lease(s)')
                    viewshed_queue_v2.cleanup_dead(r_client)
                    last_viewshed_reap = now

            if sync_radio_neighbors and WORKER_MODE in ('all', 'link'):
                refresh_radio_neighbor_reports(db, r_client)
            refresh_rf_calibration(db)
            refresh_support_context(db)
            if WORKER_MODE in ('all', 'link'):
                refresh_link_topology(db)
                now = time.time()
                if now - last_link_reap >= 5:
                    recovered = link_queue_v3.reap(r_client)
                    if recovered:
                        log.warning(f'Recovered {recovered} expired link lease(s)')
                    link_queue_v3.cleanup_completed(r_client)
                    link_queue_v3.cleanup_dead(r_client)
                    last_link_reap = now
                # Drain v3 first. Claim removes a ready entry and installs its
                # token/lease in one Lua transition.
                while True:
                    claimed = link_queue_v3.claim(r_client)
                    if claimed is None:
                        break
                    process_claimed_link_job(db, r_client, claimed)
                # Compatible cutover: drain legacy jobs only after v3.
                if not r_client.exists(link_queue_v3.REBUILD):
                    while True:
                        raw = r_client.rpop(LINK_JOB_QUEUE)
                        if raw is None:
                            break
                        process_link_job(db, r_client, json.loads(raw))

            if WORKER_MODE in ('all', 'viewshed') and VIEWSHED_ENABLED:
                claimed_viewshed = viewshed_queue_v2.claim(r_client)
                if claimed_viewshed is not None:
                    process_claimed_viewshed_job(db, r_client, claimed_viewshed)
                    continue

            if WORKER_MODE == 'viewshed':
                wait_queues = [JOB_QUEUE]
            elif WORKER_MODE == 'link':
                r_client.blpop(
                    link_queue_v3.WAKE,
                    timeout=link_idle_wait_seconds(time.time(), last_link_reap),
                )
                continue
            else:
                wait_queues = [JOB_QUEUE, LINK_JOB_QUEUE] if VIEWSHED_ENABLED else [LINK_JOB_QUEUE]

            item = r_client.brpop(wait_queues, timeout=1)
            if item is None:
                continue
            queue_name, raw = item
            if queue_name == LINK_JOB_QUEUE:
                process_link_job(db, r_client, json.loads(raw))
            else:
                job = json.loads(raw)
                release_viewshed_queue_payload(r_client, job, raw)
                try:
                    if job.get('planned_job_id'):
                        process_planned_job(db, r_client, job)
                    else:
                        process_job(db, r_client, job)
                finally:
                    r_client.srem(JOB_PENDING_SET, str(job.get('node_id') or ''))
        except psycopg2.errors.DeadlockDetected as exc:
            log.warning(f'{name}: database deadlock detected — rolling back and recovering: {exc}')
            try:
                db.rollback()
            except Exception:
                db = wait_for_db()
            time.sleep(random.uniform(0.25, 1.25))
        except (psycopg2.OperationalError, psycopg2.InterfaceError):
            log.warning(f'{name}: DB connection lost — reconnecting')
            try:
                if db is not None:
                    db.close()
            except Exception:
                pass
            db = wait_for_db()
            time.sleep(2)
        except Exception as exc:
            log.error(f'{name}: job error: {exc}')
            try:
                if db and not db.closed:
                    db.rollback()
            except Exception:
                try:
                    db = wait_for_db()
                except RuntimeError:
                    log.warning(f'{name}: DB still unavailable — retrying shortly')
            time.sleep(1)


def resolve_node_ref(db, ref: str) -> dict:
    ref = str(ref or '').strip()
    if not ref:
        raise ValueError('Empty node reference')

    with db.cursor() as cur:
        if len(ref) == 64 and all(ch in '0123456789abcdefABCDEF' for ch in ref):
            cur.execute(
                'SELECT node_id, name, lat, lon, elevation_m FROM nodes WHERE lower(node_id) = lower(%s)',
                (ref,),
            )
            rows = cur.fetchall()
        elif len(ref) >= 6 and all(ch in '0123456789abcdefABCDEF' for ch in ref):
            cur.execute(
                'SELECT node_id, name, lat, lon, elevation_m FROM nodes WHERE lower(node_id) LIKE lower(%s) ORDER BY node_id LIMIT 2',
                (f'{ref}%',),
            )
            rows = cur.fetchall()
        else:
            cur.execute(
                '''
                SELECT node_id, name, lat, lon, elevation_m
                FROM nodes
                WHERE lower(coalesce(name, '')) = lower(%s)
                   OR lower(coalesce(name, '')) LIKE lower(%s)
                ORDER BY node_id
                LIMIT 2
                ''',
                (ref, f'%{ref}%'),
            )
            rows = cur.fetchall()

    if len(rows) < 1:
        raise ValueError(f'No node matched "{ref}"')
    if len(rows) > 1:
        raise ValueError(f'Ambiguous node reference "{ref}"')
    node_id, name, lat, lon, elevation_m = rows[0]
    return {
        'node_id': node_id,
        'name': name,
        'lat': lat,
        'lon': lon,
        'elevation_m': elevation_m,
    }


def compute_pair_diagnostics(a: dict, b: dict) -> dict:
    if a.get('lat') is None or a.get('lon') is None or b.get('lat') is None or b.get('lon') is None:
        raise ValueError('Both nodes need coordinates')

    elev_a = float(a.get('elevation_m') or 0.0)
    elev_b = float(b.get('elevation_m') or 0.0)

    with tempfile.TemporaryDirectory() as tmp_dir:
        ensure_tiles_for_link(
            SRTM_DIR,
            float(a['lat']),
            float(a['lon']),
            float(b['lat']),
            float(b['lon']),
            log,
        )
        vrt = build_link_vrt(float(a['lat']), float(a['lon']), float(b['lat']), float(b['lon']), tmp_dir, SRTM_DIR)

        cos_mid = math.cos(math.radians((float(a['lat']) + float(b['lat'])) / 2))
        dlat = (float(b['lat']) - float(a['lat'])) * 111_320
        dlon = (float(b['lon']) - float(a['lon'])) * 111_320 * cos_mid
        d_total = math.sqrt(dlat ** 2 + dlon ** 2)
        n_samples = max(20, min(200, int(d_total / PROFILE_STEP_M)))

        ds = gdal.Open(vrt)
        if ds is None:
            raise RuntimeError('Failed to open generated VRT')
        gt = ds.GetGeoTransform()
        inv_gt = gdal.InvGeoTransform(gt)
        band = ds.GetRasterBand(1)

        heights: list[float] = []
        dists: list[float] = []
        for i in range(n_samples + 1):
            t = i / n_samples
            la = float(a['lat']) + t * (float(b['lat']) - float(a['lat']))
            lo = float(a['lon']) + t * (float(b['lon']) - float(a['lon']))
            px, py = gdal.ApplyGeoTransform(inv_gt, lo, la)
            px = int(np.clip(px, 0, ds.RasterXSize - 1))
            py = int(np.clip(py, 0, ds.RasterYSize - 1))
            data = band.ReadAsArray(px, py, 1, 1)
            h = max(0.0, float(data[0][0])) if data is not None else 0.0
            heights.append(h)
            dists.append(t * d_total)
        ds = None

    dists_arr = np.asarray(dists, dtype=np.float64)
    heights_arr = np.asarray(heights, dtype=np.float64)
    h_tx = elev_a + ANTENNA_HEIGHT_M
    h_rx = elev_b + ANTENNA_HEIGHT_M
    fspl = 20 * math.log10(4 * math.pi * d_total / LAMBDA_M)

    d1 = dists_arr[1:-1]
    d2 = d_total - d1
    valid = (d1 > 0) & (d2 > 0)
    d1 = d1[valid]
    d2 = d2[valid]
    profile_h = heights_arr[1:-1][valid]
    los_h = h_tx + (h_rx - h_tx) * (d1 / d_total)
    earth_bulge = (d1 * d2) / (2 * K_FACTOR * R_EARTH_M)
    excess_h = profile_h + earth_bulge - los_h
    with np.errstate(divide='ignore', invalid='ignore'):
        vs = excess_h * np.sqrt(2 * (d1 + d2) / (LAMBDA_M * d1 * d2))
    max_v = float(np.max(vs)) if vs.size else -999.0

    if max_v <= -0.78:
        diff_loss = 0.0
    else:
        diff_loss = max(0.0, 6.9 + 20 * math.log10(
            math.sqrt((max_v - 0.1) ** 2 + 1) + max_v - 0.1
        ))

    total_loss = fspl + diff_loss
    clear_los = max_v <= LINK_LOS_MAX_V
    usable_threshold_db = current_usable_path_loss_db()
    return {
        'node_a_id': a['node_id'],
        'node_a_name': a.get('name'),
        'node_b_id': b['node_id'],
        'node_b_name': b.get('name'),
        'distance_km': round(d_total / 1000.0, 3),
        'fspl_db': round(fspl, 3),
        'diffraction_loss_db': round(diff_loss, 3),
        'total_path_loss_db': round(total_loss, 3),
        'max_v': round(max_v, 6),
        'los_limit_v': LINK_LOS_MAX_V,
        'clear_los': clear_los,
        'usable_threshold_db': usable_threshold_db,
        'viable': bool(clear_los and total_loss < usable_threshold_db),
    }


def run_test_mode(args) -> None:
    db = wait_for_db()
    try:
        refresh_rf_calibration(db, force=True)
        refs: list[tuple[str, str, Optional[dict]]] = []

        if args.test_pairs:
            for raw in args.test_pairs.split(','):
                raw = raw.strip()
                if not raw:
                    continue
                left, sep, right = raw.partition(':')
                if not sep:
                    raise ValueError(f'Invalid pair "{raw}" — expected A:B')
                refs.append((left.strip(), right.strip(), None))

        if args.test_from_node:
            response = requests.get(f'{RADIO_BOT_URL}/state', timeout=10)
            response.raise_for_status()
            payload = response.json()
            monitors = payload.get('monitors') if isinstance(payload, dict) else None
            if not isinstance(monitors, list):
                raise RuntimeError('Radio bot state missing monitors list')
            source_ref = str(args.test_from_node).strip().lower()
            selected = None
            for monitor in monitors:
                if not isinstance(monitor, dict):
                    continue
                candidates = [
                    str(monitor.get('fullPublicKey') or '').strip().lower(),
                    str(monitor.get('targetHex') or '').strip().lower(),
                    str(monitor.get('nodeName') or '').strip().lower(),
                    str(monitor.get('label') or '').strip().lower(),
                ]
                if any(source_ref and source_ref in candidate for candidate in candidates if candidate):
                    selected = monitor
                    break
            if selected is None:
                raise ValueError(f'No radio monitor matched "{args.test_from_node}"')
            reporter_ref = str(selected.get('fullPublicKey') or selected.get('targetHex') or '').strip()
            neighbours = selected.get('lastNeighbours')
            if not reporter_ref or not isinstance(neighbours, list):
                raise RuntimeError('Selected radio monitor has no neighbour data')
            for neighbour in neighbours:
                if not isinstance(neighbour, dict):
                    continue
                peer_ref = str(neighbour.get('fullPublicKey') or '').strip()
                if len(peer_ref) != 64:
                    continue
                refs.append((reporter_ref, peer_ref, neighbour))

        if not refs:
            raise ValueError('No test pairs were provided')

        results = []
        for left_ref, right_ref, neighbour_meta in refs:
            a = resolve_node_ref(db, left_ref)
            b = resolve_node_ref(db, right_ref)
            result = compute_pair_diagnostics(a, b)
            if neighbour_meta is not None:
                reported_snr = neighbour_meta.get('snrDb')
                result['reported_snr_db'] = reported_snr
                result['reported_quality_band'] = radio_snr_band(reported_snr if isinstance(reported_snr, (int, float)) else None)
                result['reported_heard_seconds_ago'] = neighbour_meta.get('heardSecondsAgo')
                result['reported_adv_name'] = neighbour_meta.get('advName')
            results.append(result)

        print(json.dumps({
            'usable_threshold_db': current_usable_path_loss_db(),
            'signal_thresholds_db': current_signal_thresholds_db(),
            'results': results,
        }, indent=2))
    finally:
        db.close()

def main():
    parser = argparse.ArgumentParser(description='MeshCore viewshed/link worker')
    parser.add_argument('--test-pairs', help='Comma-separated node pairs in the form A:B,C:D')
    parser.add_argument('--test-from-node', help='Evaluate all current radio-bot neighbours for one repeater')
    args = parser.parse_args()
    if args.test_pairs or args.test_from_node:
        run_test_mode(args)
        return

    start_worker_metrics(WORKER_MODE)
    log.info(
        f'Viewshed worker starting (mode={WORKER_MODE}, '
        f'coverage_model={COVERAGE_MODEL}, model_version={COVERAGE_MODEL_VERSION}, '
        f'viewshed_enabled={VIEWSHED_ENABLED})'
    )
    SRTM_DIR.mkdir(parents=True, exist_ok=True)

    # Connect once just to enqueue any nodes that lack coverage, then hand off
    # to the worker processes (each gets its own connection).
    db = wait_for_db()
    log.info('Connected to DB')
    r = redis.Redis.from_url(REDIS_URL, decode_responses=True, password=REDIS_PASSWORD)
    r.ping()
    log.info('Connected to Redis')
    if WORKER_MODE in ('all', 'viewshed') and VIEWSHED_ENABLED:
        rebuild_pending_viewshed_set(r)
        recover_planned_coverage_jobs(db, r)
        backfill_elevations(db)
        enqueue_uncovered(db, r)
    elif WORKER_MODE == 'viewshed':
        log.info('Viewshed disabled; skipping startup coverage backfill')
    db.close()

    num_workers = int(os.environ.get('NUM_WORKERS', '2'))
    log.info(f'Launching {num_workers} worker process(es)')

    if num_workers <= 1:
        worker_loop()
        return

    procs = [
        multiprocessing.Process(target=worker_loop, name=f'Worker-{i + 1}', daemon=True)
        for i in range(num_workers)
    ]
    for p in procs:
        p.start()
    for p in procs:
        p.join()

if __name__ == '__main__':
    main()