osint-dashboard/app/geofence.py

"""Geofences: GeoJSON polygons, ST_Intersects on ingest, WS alerts.

``/api/alerts`` is the dashboard entity/keyword table — geofence hits live
in ``geofence_alerts`` and fan out as WS type ``geofence_alert``.
"""

from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from typing import Any
from uuid import uuid4

from sqlalchemy import text

from database import async_session


def _rings_from_geojson(geojson: dict) -> list[list[list[float]]]:
    if not isinstance(geojson, dict):
        raise ValueError("geojson must be an object")
    gj = geojson
    if gj.get("type") == "Feature":
        gj = gj.get("geometry") or {}
    if gj.get("type") == "FeatureCollection":
        raise ValueError("FeatureCollection is not a single polygon")
    if gj.get("type") != "Polygon":
        raise ValueError("geojson must be a Polygon")
    coords = gj.get("coordinates")
    if not isinstance(coords, list) or not coords:
        raise ValueError("polygon has no rings")
    rings: list[list[list[float]]] = []
    for ring in coords:
        if not isinstance(ring, list) or len(ring) < 4:
            raise ValueError("polygon ring needs ≥4 positions (closed)")
        pts = []
        for pt in ring:
            if not isinstance(pt, (list, tuple)) or len(pt) < 2:
                raise ValueError("position must be [lon, lat]")
            pts.append([float(pt[0]), float(pt[1])])
        rings.append(pts)
    return rings


def validate_polygon_geojson(geojson: dict) -> dict:
    """Return a canonical Polygon GeoJSON or raise ValueError."""
    rings = _rings_from_geojson(geojson)
    return {"type": "Polygon", "coordinates": rings}


def _ring_contains(lon: float, lat: float, ring: list[list[float]]) -> bool:
    """Ray-cast even-odd rule. Ring is [lon, lat] positions."""
    inside = False
    n = len(ring)
    if n < 4:
        return False
    j = n - 1
    for i in range(n):
        xi, yi = ring[i][0], ring[i][1]
        xj, yj = ring[j][0], ring[j][1]
        intersects = ((yi > lat) != (yj > lat)) and (
            lon < (xj - xi) * (lat - yi) / ((yj - yi) or 1e-16) + xi
        )
        if intersects:
            inside = not inside
        j = i
    return inside


def point_in_geojson(lon: float, lat: float, geojson: dict) -> bool:
    """True if (lon, lat) is inside the outer ring and outside holes."""
    try:
        rings = _rings_from_geojson(geojson)
    except (ValueError, TypeError, KeyError):
        return False
    if not _ring_contains(lon, lat, rings[0]):
        return False
    for hole in rings[1:]:
        if _ring_contains(lon, lat, hole):
            return False
    return True


def matching_geofences(lon: float, lat: float, fences: list[dict]) -> list[dict]:
    hits = []
    for fence in fences:
        if not fence.get("active", True):
            continue
        gj = fence.get("geojson") or {}
        if point_in_geojson(lon, lat, gj):
            hits.append(fence)
    return hits


# In-process copy of active fences so ingest does not round-trip Postgres
# on every AIS frame. CRUD endpoints refresh this list.
_cache: list[dict] = []
_recent_hits: dict[tuple[str, str], datetime] = {}
_HIT_COOLDOWN = timedelta(minutes=5)


async def refresh_cache() -> list[dict]:
    global _cache
    async with async_session() as session:
        rows = (await session.execute(text(
            "SELECT id::text, name, geojson, active FROM geofences"
        ))).mappings().all()
    _cache = [
        {
            "id": r["id"],
            "name": r["name"],
            "geojson": r["geojson"] if isinstance(r["geojson"], dict)
            else json.loads(r["geojson"] or "{}"),
            "active": bool(r["active"]),
        }
        for r in rows
    ]
    return _cache


def cached_fences() -> list[dict]:
    return list(_cache)


async def list_geofences() -> list[dict]:
    if not _cache:
        try:
            await refresh_cache()
        except Exception:
            return []
    return cached_fences()


async def create_geofence(name: str, geojson: dict, active: bool = True) -> dict:
    polygon = validate_polygon_geojson(geojson)
    gid = str(uuid4())
    gj = json.dumps(polygon)
    async with async_session() as session:
        await session.execute(
            text(
                """
                INSERT INTO geofences (id, name, geojson, geom, active)
                VALUES (
                    :id, :name, CAST(:geojson AS jsonb),
                    ST_SetSRID(ST_GeomFromGeoJSON(:geojson), 4326),
                    :active
                )
                """
            ),
            {"id": gid, "name": name, "geojson": gj, "active": 1 if active else 0},
        )
        await session.commit()
    row = {"id": gid, "name": name, "geojson": polygon, "active": active}
    _cache.append(row)
    return row


async def update_geofence(gid: str, *, name: str | None = None,
                          geojson: dict | None = None,
                          active: bool | None = None) -> dict | None:
    current = next((f for f in _cache if f["id"] == gid), None)
    if current is None:
        await refresh_cache()
        current = next((f for f in _cache if f["id"] == gid), None)
    if current is None:
        return None
    if name is not None:
        current["name"] = name
    if geojson is not None:
        current["geojson"] = validate_polygon_geojson(geojson)
    if active is not None:
        current["active"] = active
    gj = json.dumps(current["geojson"])
    async with async_session() as session:
        await session.execute(
            text(
                """
                UPDATE geofences SET
                    name = :name,
                    geojson = CAST(:geojson AS jsonb),
                    geom = ST_SetSRID(ST_GeomFromGeoJSON(:geojson), 4326),
                    active = :active,
                    updated_at = now()
                WHERE id = CAST(:id AS uuid)
                """
            ),
            {
                "id": gid,
                "name": current["name"],
                "geojson": gj,
                "active": 1 if current["active"] else 0,
            },
        )
        await session.commit()
    return current


async def delete_geofence(gid: str) -> bool:
    async with async_session() as session:
        result = await session.execute(
            text("DELETE FROM geofences WHERE id = CAST(:id AS uuid)"),
            {"id": gid},
        )
        await session.commit()
    _cache[:] = [f for f in _cache if f["id"] != gid]
    return bool(result.rowcount)


async def st_intersects(lon: float, lat: float) -> list[dict]:
    """PostGIS ST_Intersects against active geofences.

    Falls back to the in-memory GeoJSON test if the DB is unreachable so
    ingest never dies because a fence check failed.
    """
    try:
        async with async_session() as session:
            rows = (await session.execute(
                text(
                    """
                    SELECT id::text, name, geojson, active
                    FROM geofences
                    WHERE active = 1
                      AND ST_Intersects(
                            geom,
                            ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)
                          )
                    """
                ),
                {"lon": lon, "lat": lat},
            )).mappings().all()
        return [
            {
                "id": r["id"],
                "name": r["name"],
                "geojson": r["geojson"] if isinstance(r["geojson"], dict)
                else json.loads(r["geojson"] or "{}"),
                "active": True,
            }
            for r in rows
        ]
    except Exception:
        return matching_geofences(lon, lat, cached_fences())


async def record_and_notify(
    *,
    source_kind: str,
    entity_id: str,
    lat: float,
    lon: float,
    payload: dict[str, Any] | None = None,
) -> int:
    """Insert a geofence_alerts row per hit and WS-push to viewport clients.

    PostGIS ST_Intersects is the source of truth. The in-process GeoJSON
    cache is not a reject filter — the FIRMS ingester never fills it.
    """
    hits = await st_intersects(lon, lat)
    if not hits:
        return 0
    from ws_manager import manager

    sent = 0
    now = datetime.now(timezone.utc)
    fresh = []
    for fence in hits:
        key = (str(fence["id"]), str(entity_id))
        prev = _recent_hits.get(key)
        if prev is not None and now - prev < _HIT_COOLDOWN:
            continue
        _recent_hits[key] = now
        fresh.append(fence)
    if not fresh:
        return 0
    hits = fresh
    async with async_session() as session:
        for fence in hits:
            aid = str(uuid4())
            body = {
                "id": aid,
                "geofence_id": fence["id"],
                "geofence_name": fence.get("name"),
                "source_kind": source_kind,
                "entity_id": str(entity_id),
                "lat": lat,
                "lon": lon,
                "payload": payload or {},
                "created_at": now.isoformat(),
            }
            try:
                await session.execute(
                    text(
                        """
                        INSERT INTO geofence_alerts
                            (id, geofence_id, source_kind, entity_id, lat, lon, payload)
                        VALUES (
                            CAST(:id AS uuid), CAST(:geofence_id AS uuid),
                            :source_kind, :entity_id, :lat, :lon, CAST(:payload AS jsonb)
                        )
                        """
                    ),
                    {
                        "id": aid,
                        "geofence_id": fence["id"],
                        "source_kind": source_kind,
                        "entity_id": str(entity_id),
                        "lat": lat,
                        "lon": lon,
                        "payload": json.dumps(payload or {}),
                    },
                )
            except Exception:
                pass
            sent += await manager.publish_point(
                "geofence_alert", body, lat=lat, lon=lon,
            )
        try:
            await session.commit()
        except Exception:
            pass
    return sent