Merge remote-tracking branch 'forgejo/master'

# Conflicts:
#	.env.example

.env.example

@@ -27,3 +27,24 @@ CAMERA_REQUEST_DELAY=2.0
 NOMINATIM_URL=https://nominatim.openstreetmap.org
 NOMINATIM_MIN_INTERVAL=1.1
 SNAPSHOT_TTL_SECONDS=300
+
+# ── NASA FIRMS (active fire / hotspot ingest) ──────────────────────────────
+# MAP_KEY is FREE — get one at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/
+# (1-minute signup, no payment). Leave blank to keep fire ingest idle.
+FIRMS_MAP_KEY=
+# NRT VIIRS S-NPP 375m active fire detection (near-real-time).
+FIRMS_DATASET=VIIRS_SNPP_NRT
+# Area to poll as "minlon,minlat,maxlon,maxlat". Narrow it to reduce payload
+# (e.g. CONUS "-125,24,-66,50"). Default covers most of the inhabited globe.
+FIRMS_BBOX=-180,-60,180,75
+# Poll cadence in seconds (~15 min). FIRMS NRT updates every ~5-10 min.
+FIRMS_INTERVAL=900
+# Set to 0 to disable the fire loop entirely.
+INGEST_FIRES=1
+
+# ── API keys (managed from the dashboard UI) ──────────────────────────────
+# Keys such as GEMINI_API_KEY and TELEGRAM_TOKEN are stored in the Postgres
+# `api_keys` table and managed from the dashboard's "Keys" tab
+# (GET/POST/DELETE /api/keys/{name}) — see app/keystore.py. The FIRMS ingestor
+# currently reads FIRMS_MAP_KEY from .env (above); wiring the Keys-UI store as
+# its lookup/fallback is a planned follow-up.

.gitignore

@@ -3,4 +3,6 @@ __pycache__/
 *.egg-info/
 .venv/
 venv/
+.venv*/
 .env
+.pytest_cache/

alembic/versions/002_fires.py

@@ -0,0 +1,60 @@
+"""fires hypertable: NASA FIRMS active fire/hotspot detections
+
+Revision ID: 002_fires
+Revises: 001_initial
+Create Date: 2026-08-24
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '002_fires'
+down_revision = '001_initial'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # The PK IS the idempotency natural key (latitude, longitude, acq_time,
+    # satellite). acq_time is the hypertable partitioning column and is part of
+    # the PK, which satisfies TimescaleDB's requirement that every unique index
+    # include all partitioning columns.
+    op.create_table(
+        'fires',
+        sa.Column('latitude', sa.Float(), nullable=False),
+        sa.Column('longitude', sa.Float(), nullable=False),
+        sa.Column('brightness', sa.Float(), nullable=False),      # bright_ti4, K
+        sa.Column('confidence', sa.String(10), nullable=False),   # n/l/h or %
+        sa.Column('acq_time', sa.DateTime(timezone=True), nullable=False),
+        sa.Column('satellite', sa.String(16), nullable=False),
+        sa.Column('instrument', sa.String(16)),
+        sa.Column('bright_ti5', sa.Float()),
+        sa.Column('frp', sa.Float()),
+        sa.Column('daynight', sa.String(1)),
+        sa.Column('scan', sa.Float()),
+        sa.Column('track', sa.Float()),
+        sa.Column('version', sa.String(32)),
+        sa.Column('raw', sa.JSON()),
+        sa.Column('ingested_at', sa.DateTime(timezone=True),
+                  server_default=sa.func.now(), nullable=False),
+        sa.PrimaryKeyConstraint(
+            'latitude', 'longitude', 'acq_time', 'satellite',
+            name='pk_fires_natural_key',
+        ),
+    )
+
+    # 1-day chunks keep retention/drop_chunks operations cheap and scans local.
+    # (TimescaleDB auto-creates an index on the partition column acq_time.)
+    op.execute(
+        "SELECT create_hypertable('fires', 'acq_time', "
+        "if_not_exists => TRUE, chunk_time_interval => INTERVAL '1 day')"
+    )
+    op.create_index('ix_fires_bbox', 'fires', ['longitude', 'latitude'])
+
+
+def downgrade() -> None:
+    op.drop_index('ix_fires_bbox', table_name='fires')
+    op.execute("SELECT drop_hypertable('fires', if_exists => TRUE)")
+    op.drop_table('fires')

app/config.py

@@ -40,3 +40,19 @@ MINIO_ACCESS_KEY = os.getenv("MINIO_ACCESS_KEY", "")
 MINIO_SECRET_KEY = os.getenv("MINIO_SECRET_KEY", "")
 # "false" / "0" / "no" (case-insensitive) → plain HTTP (e.g. local compose).
 MINIO_SECURE = os.getenv("MINIO_SECURE", "false").lower() not in ("false", "0", "no")
+
+# ── NASA FIRMS (active fire / hotspot ingest) ───────────────────────────────
+# MAP_KEY is free; obtain one at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/
+# and set FIRMS_MAP_KEY in .env. Until it is set, the fire ingestor logs a
+# warning and stays idle (no crash).
+FIRMS_MAP_KEY = os.getenv("FIRMS_MAP_KEY", "")
+# NRT VIIRS S-NPP active fire/hotspot detection (375m).
+FIRMS_DATASET = os.getenv("FIRMS_DATASET", "VIIRS_SNPP_NRT")
+# Area bounding box as "minlon,minlat,maxlon,maxlat". Default covers most of
+# the inhabited globe; narrow it (e.g. CONUS "-125,24,-66,50") to shrink
+# payloads and the Postgres write volume.
+FIRMS_BBOX = os.getenv("FIRMS_BBOX", "-180,-60,180,75")
+# Poll cadence in seconds. FIRMS NRT updates every ~5-10 min; 900 = 15 min.
+FIRMS_INTERVAL = int(os.getenv("FIRMS_INTERVAL", "900"))
+# Outbound HTTP timeout for the FIRMS CSV download.
+FIRMS_TIMEOUT = float(os.getenv("FIRMS_TIMEOUT", "60"))

app/database.py

@@ -1,10 +1,22 @@
+import os
+
 from sqlalchemy import MetaData, event, text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from sqlalchemy.pool import NullPool
 
 from config import DATABASE_URL
 
+# Pooled connections are bound to the event loop they were created on. Long-lived
+# servers (uvicorn, the ingester) want a pool; short-lived / test processes that
+# open a fresh event loop per unit (asyncio.run, pytest) must not reuse pooled
+# connections across loops, so allow a NullPool (new connection per session).
+_NULL_POOL = os.getenv("DB_NULL_POOL", "").lower() in ("1", "true", "yes")
+
 engine = create_async_engine(
-    DATABASE_URL, echo=False, pool_size=5, max_overflow=10, pool_recycle=300
+    DATABASE_URL,
+    echo=False,
+    **({"poolclass": NullPool} if _NULL_POOL else
+       {"pool_size": 5, "max_overflow": 10, "pool_recycle": 300}),
 )
 async_session = async_sessionmaker(
     engine, class_=AsyncSession, expire_on_commit=False

app/fire_sources.py

@@ -0,0 +1,192 @@
+"""NASA FIRMS active fire / hotspot ingestor.
+
+Fetches the VIIRS active-fire detection CSV for an area from NASA FIRMS and
+publishes each hotspot to NATS JetStream on the ``events.fire`` subject. The
+NATS consumer (app/ingestor.py) routes those messages to the ``fires``
+TimescaleDB hypertable, deduped by the natural key (lat, lon, acq_time,
+satellite).
+
+Zero-cost data: the FIRMS MAP_KEY is free (https://firms.modaps.eosdis.nasa.gov/api/map_key_info/)
+and the CSV area endpoint is unlimited for personal/research use.
+
+Endpoint (documented at https://firms.modaps.eosdis.nasa.gov/api/area/csv/):
+    https://firms.modaps.eosdis.nasa.gov/api/area/csv/{MAP_KEY}/{DATASET}/{bbox}
+    bbox = "minlon,minlat,maxlon,maxlat"  (e.g. "-125,24,-66,50")
+
+CSV columns (VIIRS): latitude, longitude, bright_ti4, scan, track, acq_date,
+acq_time, satellite, instrument, confidence, version, bright_ti5, frp, daynight.
+``acq_time`` is an integer HHMM in UTC (seconds are 0); ``acq_date`` is the
+UTC date (YYYY-MM-DD).
+"""
+
+from __future__ import annotations
+
+import csv
+import io
+import json
+import logging
+from datetime import datetime, timezone
+
+import httpx
+import nats
+
+from config import FIRMS_MAP_KEY, FIRMS_DATASET, FIRMS_BBOX, FIRMS_TIMEOUT, NATS_URL
+
+logger = logging.getLogger("osint.firms")
+
+# ── FIRMS API ─────────────────────────────────────────────────────────────
+
+FIRMS_AREA_CSV = (
+    "https://firms.modaps.eosdis.nasa.gov/api/area/csv/{key}/{dataset}/{bbox}"
+)
+
+# The canonical VIIRS CSV header FIRMS returns. Used to (a) locate the real
+# header row if FIRMS ever prepends a legend line and (b) validate a download.
+FIRMS_CSV_COLUMNS = (
+    "latitude", "longitude", "bright_ti4", "scan", "track", "acq_date",
+    "acq_time", "satellite", "instrument", "confidence", "version",
+    "bright_ti5", "frp", "daynight",
+)
+
+
+def _to_float(value: object) -> float | None:
+    """Best-effort float conversion; returns None on empty/garbage values."""
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def normalize_acq_time(acq_date: object, acq_time: object) -> datetime | None:
+    """Combine FIRMS acq_date + acq_time into a timezone-aware UTC datetime.
+
+    acq_date is 'YYYY-MM-DD'; acq_time is an integer HHMM in UTC with seconds
+    truncated. Returns None when the values can't be parsed (dropped).
+    """
+    if not acq_date or acq_time is None or acq_time == "":
+        return None
+    try:
+        hhmm = str(int(acq_time)).zfill(4)
+        return datetime.strptime(
+            f"{acq_date} {hhmm}", "%Y-%m-%d %H%M"
+        ).replace(tzinfo=timezone.utc)
+    except (ValueError, TypeError):
+        return None
+
+
+def parse_firms_csv(text: str) -> list[dict]:
+    """Parse a FIRMS area CSV payload into normalized fire messages.
+
+    Returns one dict per hotspot with the fields stored in the ``fires`` table
+    (acq_time already combined into a UTC ISO timestamp). Rows that don't look
+    like valid VIIRS detections are skipped rather than failing the whole poll.
+    """
+    rows = list(csv.reader(io.StringIO(text)))
+    if not rows:
+        return []
+
+    # Locate the real header row. FIRMS normally returns the CSV header first,
+    # but occasionally prepends a legend/info line, so scan until we see the
+    # canonical header.
+    header_idx = 0
+    for i, row in enumerate(rows):
+        if row and row[0].strip().lower() == "latitude" and len(row) >= 4:
+            header_idx = i
+            break
+    header = [c.strip().lower() for c in rows[header_idx]]
+    # Guard against a header that isn't actually the FIRMS one.
+    if "latitude" not in header or "longitude" not in header:
+        logger.warning("FIRMS payload does not look like a hotspot CSV (first row: %r)", header[:6])
+        return []
+
+    points: list[dict] = []
+    for row in rows[header_idx + 1:]:
+        if len(row) < len(header):
+            continue
+        rec = dict(zip(header, row))
+        lat = _to_float(rec.get("latitude"))
+        lon = _to_float(rec.get("longitude"))
+        if lat is None or lon is None:
+            continue
+        acq_time = normalize_acq_time(rec.get("acq_date"), rec.get("acq_time"))
+        if acq_time is None:
+            continue
+        points.append({
+            "latitude": lat,
+            "longitude": lon,
+            "brightness": _to_float(rec.get("bright_ti4")),
+            "confidence": str(rec.get("confidence") or "").strip(),
+            "acq_time": acq_time.isoformat(),
+            "satellite": str(rec.get("satellite") or "").strip(),
+            "instrument": str(rec.get("instrument") or "").strip(),
+            "bright_ti5": _to_float(rec.get("bright_ti5")),
+            "frp": _to_float(rec.get("frp")),
+            "daynight": str(rec.get("daynight") or "").strip(),
+            "scan": _to_float(rec.get("scan")),
+            "track": _to_float(rec.get("track")),
+            "version": str(rec.get("version") or "").strip(),
+        })
+    return points
+
+
+async def publish_fire_batch(points: list[dict]) -> int:
+    """Publish a batch of hotspot messages to NATS JetStream in one connection.
+
+    Fires are bulk data (thousands of detections per poll), so opening a NATS
+    connection per message — as the RSS/GDELT path does — would be wasteful.
+    One connection, one JetStream context, one close, with the batch flushed
+    before returning so a crash mid-write can't silently drop half a poll.
+    """
+    if not points:
+        return 0
+    nc = await nats.connect(NATS_URL)
+    js = nc.jetstream()
+    try:
+        for pt in points:
+            msg = {"source_type": "fire", **pt}
+            await js.publish("events.fire", json.dumps(msg).encode())
+    finally:
+        await nc.close()
+    return len(points)
+
+
+async def ingest_fires(bbox: str | None = None) -> int:
+    """Fetch the FIRMS hotspot CSV for an area and publish it to NATS.
+
+    Returns the number of hotspot messages published. Idle (0) and a logged
+    warning when FIRMS_MAP_KEY is not set, so the ingester keeps running for
+    the other sources.
+    """
+    if not FIRMS_MAP_KEY:
+        logger.warning(
+            "FIRMS_MAP_KEY not set — fire ingest disabled. "
+            "Get a free key at https://firms.modaps.eosdis.nasa.gov/api/map_key_info/"
+        )
+        return 0
+
+    area = bbox or FIRMS_BBOX
+    url = FIRMS_AREA_CSV.format(
+        key=FIRMS_MAP_KEY, dataset=FIRMS_DATASET, bbox=area
+    )
+
+    async with httpx.AsyncClient(timeout=FIRMS_TIMEOUT) as client:
+        resp = await client.get(url)
+        resp.raise_for_status()
+        text = resp.text
+
+    # FIRMS returns HTTP 200 with a plain-text error for some failure modes
+    # (bad key, invalid bbox); surface the first line for debuggability.
+    if "latitude" not in text.lower()[:4096]:
+        first_line = text.strip().splitlines()[0][:200] if text.strip() else "(empty)"
+        logger.warning("FIRMS CSV download returned no hotspot data (%s)", first_line)
+        return 0
+
+    points = parse_firms_csv(text)
+    published = await publish_fire_batch(points)
+    logger.info(
+        "FIRMS: fetched %d hotspot(s) for bbox=%s (%s), published %d",
+        len(points), area, FIRMS_DATASET, published,
+    )
+    return published

app/ingestor.py

@@ -9,9 +9,11 @@ from datetime import datetime, timezone
 
 import nats
 from nats.errors import TimeoutError
+from sqlalchemy.dialects.postgresql import insert as pg_insert
 
 from database import async_session
 from models import events as events_table
+from models import fires as fires_table
 from config import NATS_URL
 
 logger = logging.getLogger("osint.ingestor")
@@ -22,8 +24,86 @@ NATS_STREAM = "events"
 NATS_DURABLE = "osint-ingestor"
 
 
+# ── Active fire / hotspot routing ─────────────────────────────────────────
+
+def _fire_row_from_msg(msg: dict) -> dict | None:
+    """Map a NATS fire message to a ``fires`` table row (pre-DB, unit-testable).
+
+    Returns None (dropped) when the idempotency key fields — latitude,
+    longitude, acq_time, satellite — are missing or unparseable.
+    """
+    row = {
+        "latitude": msg.get("latitude"),
+        "longitude": msg.get("longitude"),
+        "brightness": msg.get("brightness"),
+        "confidence": msg.get("confidence"),
+        "acq_time": msg.get("acq_time"),
+        "satellite": msg.get("satellite"),
+        "instrument": msg.get("instrument"),
+        "bright_ti5": msg.get("bright_ti5"),
+        "frp": msg.get("frp"),
+        "daynight": msg.get("daynight"),
+        "scan": msg.get("scan"),
+        "track": msg.get("track"),
+        "version": msg.get("version"),
+        "raw": msg.get("raw"),
+    }
+    # Idempotency key must be complete; brightness/confidence may be absent in
+    # malformed feeds but lat/lon/time/satellite are required for dedupe.
+    if (
+        row["latitude"] is None or row["longitude"] is None
+        or row["acq_time"] is None or not row["satellite"]
+    ):
+        logger.warning(
+            "dropping malformed fire message (missing natural key): %s",
+            {k: msg.get(k) for k in ("latitude", "longitude", "acq_time", "satellite")},
+        )
+        return None
+    if isinstance(row["acq_time"], str):
+        try:
+            row["acq_time"] = datetime.fromisoformat(row["acq_time"])
+        except ValueError:
+            logger.warning("dropping fire message with bad acq_time %r", row["acq_time"])
+            return None
+    if not isinstance(row["acq_time"], datetime):
+        row["acq_time"] = datetime.fromisoformat(str(row["acq_time"]))
+    return row
+
+
+async def ingest_fire_row(msg: dict) -> bool:
+    """Persist one FIRMS hotspot, idempotently.
+
+    The (latitude, longitude, acq_time, satellite) primary key doubles as the
+    dedupe key: ON CONFLICT DO NOTHING means a hotspot re-delivered on a later
+    15-minute poll is silently ignored. Returns True if a new row was inserted,
+    False if it was a duplicate (or dropped).
+    """
+    row = _fire_row_from_msg(msg)
+    if row is None:
+        return False
+    async with async_session() as session:
+        stmt = (
+            pg_insert(fires_table)
+            .values(**row)
+            .on_conflict_do_nothing(constraint="pk_fires_natural_key")
+        )
+        result = await session.execute(stmt)
+        await session.commit()
+    inserted = bool(result.rowcount)
+    if inserted:
+        logger.info(
+            "ingested fire %.5f,%.5f %s satellite=%s",
+            row["latitude"], row["longitude"], row["acq_time"], row["satellite"],
+        )
+    return inserted
+
+
 async def ingest_event(msg: dict):
     """Ingest a single event from NATS into PostgreSQL."""
+    # Active fire/hotspot messages carry a dedicated schema and land in the
+    # `fires` hypertable (idempotent natural key), not the generic events feed.
+    if msg.get("source_type") == "fire":
+        return await ingest_fire_row(msg)
     # source_id links to a feed_sources UUID; tolerate non-UUID / missing values
     # (e.g. legacy messages that carried a URL) by coercing to None.
     raw_source_id = msg.get("source_id")

app/keystore.py

@@ -0,0 +1,212 @@
+"""OSINT Dashboard — API key store (keyv-style Postgres table).
+
+Keys live in the ``api_keys`` table, shared between the dashboard app and the
+ingest services (both connect to the same Postgres). Only the app WRITES keys
+via the management API; ingest services read them with :func:`get_api_key`.
+
+Storage: the table is created lazily with ``CREATE TABLE IF NOT EXISTS`` on
+first use in each process (no alembic migration, so it can never fork the
+migration chain or block container startup). DDL is idempotent and safe when
+the app and ingester containers race on first boot.
+
+Security contract:
+  * Values are stored in Postgres, never in the container image or frontend.
+  * ``GET /api/keys`` returns only set/missing status plus a masked
+    "****last4" hint. Full values are NEVER returned by the API.
+  * Registered keys get cheap format validation on save (regex), e.g. the
+    FIRMS map key must be a 32-char hex string.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+from datetime import datetime, timezone
+
+from sqlalchemy import Column, DateTime, String, Table, Text, func, select, text
+
+from database import async_session, engine, metadata
+
+# ── Table definition (bound to the shared metadata; created lazily) ───────
+api_keys = Table(
+    "api_keys",
+    metadata,
+    Column("name", String(128), primary_key=True),
+    Column("value", Text, nullable=False),
+    Column("created_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
+    Column("updated_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
+)
+
+_CREATE_TABLE_SQL = text(
+    """
+    CREATE TABLE IF NOT EXISTS api_keys (
+        name       VARCHAR(128) PRIMARY KEY,
+        value      TEXT NOT NULL,
+        created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+        updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
+    )
+    """
+)
+
+# ── Registry of keys the ingest services understand ───────────────────────
+# Each entry: human description (shown in the UI) + optional ``pattern`` for
+# cheap format validation and an ``example`` for error/placeholder text.
+KEY_REGISTRY: dict[str, dict] = {
+    "FIRMS_MAP_KEY": {
+        "description": "NASA FIRMS API key — active fire / satellite ingest.",
+        "pattern": r"^[0-9a-fA-F]{32}$",
+        "example": "32-char hex string (e.g. 5f3c…9a02)",
+    },
+    "GEMINI_API_KEY": {
+        "description": "Google Gemini API key — LLM event analysis / summarization.",
+        "pattern": r"^AIza[0-9A-Za-z_-]{35}$",
+        "example": "AIza… (Google API key, 39 chars)",
+    },
+    "TELEGRAM_TOKEN": {
+        "description": "Telegram bot token — push alert notifications to a channel.",
+        "pattern": r"^\d{8,10}:[0-9A-Za-z_-]{35}$",
+        "example": "123456789:AA… (bot token from @BotFather)",
+    },
+}
+
+# Any stored key must at least be a sane UPPER_SNAKE name.
+_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]{1,63}$")
+
+
+class KeyFormatError(ValueError):
+    """Raised when a key name or value fails format validation."""
+
+
+# ── Lazy table bootstrap ──────────────────────────────────────────────────
+
+_ensure_lock = asyncio.Lock()
+_ensured = False
+
+
+async def ensure_api_keys_table() -> None:
+    """Create the api_keys table if it doesn't exist (idempotent, per process)."""
+    global _ensured
+    if _ensured:
+        return
+    async with _ensure_lock:
+        if _ensured:
+            return
+        async with engine.begin() as conn:
+            await conn.execute(_CREATE_TABLE_SQL)
+        _ensured = True
+
+
+# ── Validation ────────────────────────────────────────────────────────────
+
+def is_registered(name: str) -> bool:
+    """True if ``name`` has an entry in the registry (gets format validation)."""
+    return name in KEY_REGISTRY
+
+
+def validate_name(name: str) -> None:
+    """Reject malformed key names (must be UPPER_SNAKE)."""
+    if not _NAME_RE.match(name or ""):
+        raise KeyFormatError(
+            "Invalid key name — use UPPER_SNAKE_CASE (e.g. FIRMS_MAP_KEY)."
+        )
+
+
+def validate_value(name: str, value: str) -> None:
+    """Cheap format validation for registered keys; rejects empty values."""
+    if not value or not value.strip():
+        raise KeyFormatError("Value must not be empty.")
+    info = KEY_REGISTRY.get(name)
+    pattern = (info or {}).get("pattern")
+    if pattern and not re.match(pattern, value.strip()):
+        example = (info or {}).get("example", "a matching value")
+        raise KeyFormatError(f"'{name}' has an invalid format — expected {example}.")
+
+
+def mask_value(value: str | None) -> str:
+    """Mask a stored value as '****last4' (or '****' when shorter than 4)."""
+    if not value:
+        return ""
+    v = value.strip()
+    tail = v[-4:] if len(v) >= 4 else ""
+    return f"****{tail}"
+
+
+# ── Store operations ──────────────────────────────────────────────────────
+
+async def list_keys() -> list[dict]:
+    """Registry keys + any extra stored keys, with masked set-status only."""
+    await ensure_api_keys_table()
+    async with async_session() as session:
+        rows = (await session.execute(select(api_keys))).mappings().all()
+    stored = {r["name"]: r["value"] for r in rows}
+
+    names = list(KEY_REGISTRY) + [n for n in stored if n not in KEY_REGISTRY]
+    out = []
+    for name in names:
+        value = stored.get(name)
+        info = KEY_REGISTRY.get(name, {})
+        out.append(
+            {
+                "name": name,
+                "set": value is not None,
+                "masked": mask_value(value) if value is not None else None,
+                "description": info.get("description", ""),
+                "example": info.get("example", ""),
+                "validated": name in KEY_REGISTRY,
+            }
+        )
+    out.sort(key=lambda k: k["name"].lower())
+    return out
+
+
+async def set_key(name: str, value: str) -> dict:
+    """Upsert a key value. Validates registered formats; rejects bad names.
+
+    Returns the masked status (never the raw value).
+    """
+    validate_name(name)
+    validate_value(name, value)
+    value = value.strip()
+    now = datetime.now(timezone.utc)
+
+    await ensure_api_keys_table()
+    async with async_session() as session:
+        existing = (
+            await session.execute(select(api_keys).where(api_keys.c.name == name))
+        ).mappings().one_or_none()
+        if existing:
+            await session.execute(
+                api_keys.update()
+                .where(api_keys.c.name == name)
+                .values(value=value, updated_at=now)
+            )
+        else:
+            await session.execute(
+                api_keys.insert().values(name=name, value=value, updated_at=now)
+            )
+        await session.commit()
+    return {"name": name, "set": True, "masked": mask_value(value)}
+
+
+async def delete_key(name: str) -> bool:
+    """Remove a stored key. Returns True if something was deleted."""
+    await ensure_api_keys_table()
+    async with async_session() as session:
+        result = await session.execute(api_keys.delete().where(api_keys.c.name == name))
+        await session.commit()
+    return bool(result.rowcount)
+
+
+async def get_api_key(name: str) -> str | None:
+    """Read a stored key value — used by ingest services, never by the API.
+
+    Returns the raw value (or None when unset) so producers can pass it to
+    external APIs (FIRMS, Gemini, Telegram, …). Reads live from Postgres, so a
+    key set via the dashboard is picked up on the next poll — no restart.
+    """
+    await ensure_api_keys_table()
+    async with async_session() as session:
+        row = (
+            await session.execute(select(api_keys).where(api_keys.c.name == name))
+        ).mappings().one_or_none()
+    return row["value"] if row else None

app/main.py

@@ -26,18 +26,21 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from database import async_session, init_extensions
 from models import (
-    alerts, documents, entities, entity_events, events, feed_sources
+    alerts, documents, entities, entity_events, events, feed_sources, fires
 )
 from schemas import (
     AlertCreate, AlertOut, AlertSeverity, AlertType, AlertUpdate,
     DashboardSummary, EntityCreate, EntityKind, EntityOut,
-    EventCreate, EventOut,
+    EventCreate, EventOut, FireOut,
     FeedSourceCreate, FeedSourceOut,
+    KeyOut, KeyValueIn,
     SearchResult, SentimentSummary, SourceType,
     SearchQuery, TimelinePoint,
 )
 from ingestor import ingest_event, fetch_and_process
 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes, ingest_social_signals
+from fire_sources import ingest_fires
+from keystore import KeyFormatError, delete_key, list_keys, set_key
 
 logging.basicConfig(level=logging.INFO)
 logger = structlog.get_logger("osint.dashboard")
@@ -218,6 +221,64 @@ async def create_event(payload: EventCreate):
     return {"id": str(event_id)}
 
 
+# ── Active Fires / Hotspots (NASA FIRMS) ─────────────────────────────────
+
+@app.get("/api/fires", response_model=list[FireOut])
+async def list_fires(
+    bbox: str | None = Query(
+        None,
+        description="Comma-separated 'minlon,minlat,maxlon,maxlat' to bound the "
+                    "result set (e.g. '-125,24,-66,50'). Omit for all stored "
+                    "detections (most recent first).",
+    ),
+    since: datetime | None = Query(
+        None,
+        description="Only hotspots acquired at/after this UTC instant "
+                    "(ISO 8601, e.g. '2026-08-24T12:00:00Z').",
+    ),
+    limit: int = Query(2000, ge=1, le=10000),
+):
+    """List stored FIRMS active fire/hotspot detections as JSON.
+
+    This is the data contract for the map's fire heatmap overlay: the frontend
+    calls `GET /api/fires?bbox=...&since=...` and renders the returned points.
+    """
+    async with async_session() as session:
+        stmt = select(fires).order_by(fires.c.acq_time.desc())
+        if since:
+            stmt = stmt.where(fires.c.acq_time >= since)
+        if bbox:
+            parts = [p.strip() for p in bbox.split(",")]
+            if len(parts) != 4:
+                raise HTTPException(
+                    422, "bbox must be 'minlon,minlat,maxlon,maxlat' (4 comma-separated values)"
+                )
+            try:
+                minlon, minlat, maxlon, maxlat = (float(p) for p in parts)
+            except ValueError:
+                raise HTTPException(
+                    422, "bbox values must be floats: 'minlon,minlat,maxlon,maxlat'"
+                )
+            stmt = stmt.where(
+                and_(
+                    fires.c.longitude >= minlon, fires.c.longitude <= maxlon,
+                    fires.c.latitude >= minlat, fires.c.latitude <= maxlat,
+                )
+            )
+        stmt = stmt.limit(limit)
+        rows = (await session.execute(stmt)).mappings().all()
+    return [
+        FireOut(
+            latitude=r["latitude"], longitude=r["longitude"],
+            brightness=r["brightness"], confidence=r["confidence"],
+            acq_time=r["acq_time"], satellite=r["satellite"],
+            instrument=r["instrument"], bright_ti5=r["bright_ti5"],
+            frp=r["frp"], daynight=r["daynight"],
+        )
+        for r in rows
+    ]
+
+
 # ── Search ────────────────────────────────────────────────────────────────
 
 @app.post("/api/search", response_model=SearchResult)
@@ -432,6 +493,43 @@ async def list_documents(
     }
 
 
+# ── API Keys ────────────────────────────────────────────────────────────
+
+@app.get("/api/keys", response_model=list[KeyOut])
+async def list_api_keys():
+    """List known API keys with set/missing status — masked, never raw.
+
+    Registered keys (FIRMS_MAP_KEY, GEMINI_API_KEY, TELEGRAM_TOKEN)
+    are always included. Any extra stored keys are appended.
+    """
+    return await list_keys()
+
+
+@app.post("/api/keys/{name}", status_code=200)
+async def save_api_key(name: str, payload: KeyValueIn):
+    """Save (upsert) an API key value.
+
+    Registered key formats are validated (e.g. FIRMS_MAP_KEY must be a
+    32-char hex string). Unregistered names are accepted as long as they
+    use UPPER_SNAKE_CASE. The raw value is stored in Postgres and never
+    returned by any API endpoint — only the masked status is exposed.
+    """
+    try:
+        result = await set_key(name, payload.value)
+    except KeyFormatError as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
+    return result
+
+
+@app.delete("/api/keys/{name}")
+async def remove_api_key(name: str):
+    """Delete a stored API key (unsets it)."""
+    removed = await delete_key(name)
+    if not removed:
+        raise HTTPException(status_code=404, detail="Key not found")
+    return {"ok": True}
+
+
 # ── Ingestion Triggers ───────────────────────────────────────────────────
 
 @app.post("/api/ingest/rss")
@@ -455,6 +553,13 @@ async def trigger_earthquake_ingest():
     return {"status": "ok", "events_ingested": count}
 
 
+@app.post("/api/ingest/fires")
+async def trigger_fire_ingest(bbox: str | None = None):
+    """Trigger a NASA FIRMS active-fire poll (uses FIRMS_BBOX if bbox omitted)."""
+    count = await ingest_fires(bbox)
+    return {"status": "ok", "hotspots_published": count}
+
+
 @app.post("/api/ingest/social")
 async def trigger_social_ingest(query: str = "", max_items: int = 50):
     """Trigger social signals ingestion."""

app/models.py

@@ -5,7 +5,7 @@ from uuid import uuid4
 
 from sqlalchemy import (
     Column, Enum, Float, Index, Integer, String, Text,
-    DateTime, JSON, func, Table,
+    DateTime, JSON, func, Table, PrimaryKeyConstraint,
 )
 from sqlalchemy.dialects.postgresql import UUID, TSVECTOR
 
@@ -145,3 +145,40 @@ documents = Table(
     Column("event_id", UUID(as_uuid=True)),
     Column("uploaded_at", DateTime(timezone=True), server_default=func.now()),
 )
+
+
+# ── Active Fires / Hotspots (NASA FIRMS) ──────────────────────────────────
+
+# VIIRS active fire/hotspot detections from NASA FIRMS. The primary key IS the
+# natural key (latitude, longitude, acq_time, satellite) so repeated 15-minute
+# polls of the same detection are idempotent (INSERT ... ON CONFLICT DO NOTHING
+# silently ignores duplicates). acq_time is the partitioning column of the
+# TimescaleDB hypertable, and since it is part of the PK, TimescaleDB's
+# "all unique indexes must include the partitioning column" rule is satisfied.
+# No surrogate id is needed — the natural key is the identity of a detection.
+fires = Table(
+    "fires",
+    metadata,
+    Column("latitude", Float, nullable=False),
+    Column("longitude", Float, nullable=False),
+    Column("brightness", Float, nullable=False),          # bright_ti4, Kelvin
+    Column("confidence", String(10), nullable=False),     # VIIRS: n/l/h, MODIS: %
+    Column("acq_time", DateTime(timezone=True), nullable=False),  # UTC acquisition
+    Column("satellite", String(16), nullable=False),      # e.g. N (S-NPP), N20, N21
+    Column("instrument", String(16)),
+    Column("bright_ti5", Float),                          # 12µm brightness, Kelvin
+    Column("frp", Float),                                 # fire radiative power, MW
+    Column("daynight", String(1)),                        # D / N
+    Column("scan", Float),
+    Column("track", Float),
+    Column("version", String(32)),                        # e.g. 2.0NRT / 2.0URT
+    Column("raw", JSON),
+    Column("ingested_at", DateTime(timezone=True), server_default=func.now(), nullable=False),
+    PrimaryKeyConstraint(
+        "latitude", "longitude", "acq_time", "satellite",
+        name="pk_fires_natural_key",
+    ),
+)
+
+# Bounding-box index for `bbox=` filtering (lon first for max/min-lon scans).
+Index("ix_fires_bbox", fires.c.longitude, fires.c.latitude)

app/run_ingester.py

@@ -24,8 +24,9 @@ import sys
 
 sys.path.insert(0, sys_path)
 
-from config import NATS_URL  # noqa: E402
+from config import NATS_URL, FIRMS_INTERVAL, FIRMS_DATASET  # noqa: E402
 from sources import ingest_rss_feed, ingest_gdelt, ingest_earthquakes  # noqa: E402
+from fire_sources import ingest_fires  # noqa: E402
 from ingestor import ingest_event, start_nats_consumer  # noqa: E402
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
@@ -35,6 +36,7 @@ RSS_URLS = [u.strip() for u in os.getenv("RSS_URL", "").split(",") if u.strip()]
 INTERVAL = int(os.getenv("INGEST_INTERVAL", "300"))
 GDELT_QUERY = os.getenv("GDELT_QUERY", "")
 ENABLE_QUAKES = os.getenv("INGEST_EARTHQUAKES", "1").lower() in ("1", "true", "yes")
+ENABLE_FIRES = os.getenv("INGEST_FIRES", "1").lower() in ("1", "true", "yes")
 
 NATS_STREAM = "events"
 
@@ -84,12 +86,33 @@ async def consumer_loop() -> None:
                 logger.exception("failed to process message")
 
 
+async def fire_loop() -> None:
+    """Poll NASA FIRMS active fires on the ~15-minute cadence.
+
+    Runs as its own task so its slower cadence (FIRMS_INTERVAL, default 900s)
+    is independent of the RSS/GDELT/USGS producer cycle (INGEST_INTERVAL).
+    """
+    logger.info("fire loop starting (interval=%ss, dataset=%s)", FIRMS_INTERVAL, FIRMS_DATASET)
+    while True:
+        try:
+            n = await ingest_fires()
+            logger.info("FIRMS -> %d hotspots", n)
+        except Exception:  # noqa: BLE001 — keep the loop alive across transient failures
+            logger.exception("FIRMS fetch failed")
+        await asyncio.sleep(FIRMS_INTERVAL)
+
+
 async def main() -> None:
     logger.info(
-        "ingester starting (rss=%d feeds, gdelt_q=%r, quakes=%s, interval=%ss)",
-        len(RSS_URLS), GDELT_QUERY, ENABLE_QUAKES, INTERVAL,
+        "ingester starting (rss=%d feeds, gdelt_q=%r, quakes=%s, fires=%s, interval=%ss)",
+        len(RSS_URLS), GDELT_QUERY, ENABLE_QUAKES, ENABLE_FIRES, INTERVAL,
     )
-    await asyncio.gather(producer_loop(), consumer_loop())
+    tasks: list[asyncio.Task] = []
+    if ENABLE_FIRES:
+        # Fire ingest only starts once FIRMS_MAP_KEY is set (ingest_fires logs
+        # and idles otherwise).
+        tasks.append(asyncio.create_task(fire_loop()))
+    await asyncio.gather(producer_loop(), consumer_loop(), *tasks)
 
 
 if __name__ == "__main__":

app/schemas.py

@@ -215,6 +215,44 @@ class DocumentOut(BaseModel):
     uploaded_at: datetime
 
 
+# ─── API Keys ────────────────────────────────────────────────────────────
+
+class KeyValueIn(BaseModel):
+    """Body for POST /api/keys/{name} — the value to store.
+
+    The raw value is accepted only on write; it is never read back out.
+    """
+    value: str = Field(..., min_length=1, max_length=4096)
+
+
+class KeyOut(BaseModel):
+    """One key's status as exposed by GET /api/keys.
+
+    ``masked`` is the "****last4" hint; the full value is NEVER included.
+    """
+    name: str
+    set: bool
+    masked: Optional[str] = None
+    description: Optional[str] = None
+    example: Optional[str] = None
+    validated: bool = True
+
+
+# ─── Active fires / hotspots (NASA FIRMS) ────────────────────────────────
+
+class FireOut(BaseModel):
+    latitude: float
+    longitude: float
+    brightness: float  # bright_ti4 brightness temperature, Kelvin
+    confidence: str    # VIIRS: n (nominal) / l (low) / h (high)
+    acq_time: datetime
+    satellite: str
+    instrument: Optional[str] = None
+    bright_ti5: Optional[float] = None  # 12µm brightness temperature, Kelvin
+    frp: Optional[float] = None         # fire radiative power, MW
+    daynight: Optional[str] = None      # D / N
+
+
 # ─── Aggregations ────────────────────────────────────────────────────────
 
 class SentimentSummary(BaseModel):

app/static/index.html

@@ -40,6 +40,26 @@
         .sentiment-bar div { transition: width 0.3s; }
         .tab-bar { display: flex; gap: 0.5rem; margin-bottom: 1rem; }
         .tab-bar .btn.active { background: var(--accent); color: #0f172a; }
+        .key-grid { display: grid; grid-template-columns: repeat(auto-fit, minmax(320px, 1fr)); gap: 1rem; }
+        .key-card { background: var(--surface); border: 1px solid var(--border); border-radius: 10px; padding: 1rem 1.1rem; box-shadow: 0 0 18px rgba(56,189,248,0.06); transition: box-shadow .2s, border-color .2s; }
+        .key-card:hover { box-shadow: 0 0 22px rgba(56,189,248,0.14); }
+        .key-card.set { border-color: rgba(74,222,128,0.5); box-shadow: 0 0 18px rgba(74,222,128,0.14); }
+        .key-head { display: flex; justify-content: space-between; align-items: center; gap: 0.5rem; margin-bottom: 0.35rem; }
+        .key-name { font-family: ui-monospace, SFMono-Regular, Menlo, monospace; font-size: 0.9rem; font-weight: 600; color: var(--accent); word-break: break-all; }
+        .key-status { font-size: 0.72rem; font-family: ui-monospace, monospace; padding: 0.2rem 0.55rem; border-radius: 9999px; white-space: nowrap; }
+        .key-status.set { background: rgba(74,222,128,0.12); color: var(--green); box-shadow: 0 0 10px rgba(74,222,128,0.35); }
+        .key-status.missing { background: rgba(248,113,113,0.12); color: var(--red); box-shadow: 0 0 10px rgba(248,113,113,0.25); }
+        .key-desc { font-size: 0.82rem; color: var(--muted); margin-bottom: 0.7rem; }
+        .key-hint { font-size: 0.72rem; color: var(--muted); opacity: 0.8; margin-top: 0.15rem; }
+        .key-form { display: flex; gap: 0.5rem; }
+        .key-form input { flex: 1; background: var(--bg); border: 1px solid var(--border); border-radius: 6px; padding: 0.5rem 0.6rem; color: var(--text); font-size: 0.85rem; font-family: ui-monospace, monospace; }
+        .key-form input:focus { outline: none; border-color: var(--accent); box-shadow: 0 0 8px rgba(56,189,248,0.35); }
+        .key-form .btn { white-space: nowrap; }
+        .key-form .btn-danger { border-color: var(--red); color: var(--red); }
+        .key-form .btn-danger:hover { background: rgba(248,113,113,0.12); }
+        .key-msg { margin-top: 0.5rem; font-size: 0.8rem; min-height: 1rem; }
+        .key-msg.ok { color: var(--green); }
+        .key-msg.err { color: var(--red); }
     </style>
 </head>
 <body>
@@ -85,6 +105,7 @@
             <button class="btn" onclick="showTab('alerts')">Alerts</button>
             <button class="btn" onclick="showTab('entities')">Entities</button>
             <button class="btn" onclick="showTab('ingest')">Ingest</button>
+            <button class="btn" onclick="showTab('keys')">Keys</button>
         </div>
 
         <!-- Recent Events -->
@@ -144,6 +165,13 @@
             <div id="ingest-result" style="margin-top:1rem;color:var(--muted);font-size:0.9rem"></div>
         </div>
 
+        <!-- API Keys -->
+        <div class="section" id="tab-keys" style="display:none">
+            <h2>API Keys</h2>
+            <p class="sub" style="margin-bottom:0.75rem">Keys used by ingest services (FIRMS, Gemini, Telegram, ...). Values are stored in Postgres and never shown again — only set/missing status and the last 4 chars are displayed.</p>
+            <div class="key-grid" id="keys-grid"></div>
+        </div>
+
         <!-- Search Results -->
         <div class="section" id="search-results" style="display:none">
             <h2>Search Results (<span id="search-total">0</span>)</h2>
@@ -247,15 +275,83 @@ function sentimentBadge(label) {
     return `<span class="badge ${cls}">${label}</span>`;
 }
 
+// ── API Keys ────────────────────────────────────────────────────────────
+// Mirrors the backend keystore registry for instant client-side format hints.
+const KEY_PATTERNS = {
+    'FIRMS_MAP_KEY': /^[0-9a-fA-F]{32}$/,
+    'GEMINI_API_KEY': /^AIza[0-9A-Za-z_-]{35}$/,
+    'TELEGRAM_TOKEN': /^\d{8,10}:[0-9A-Za-z_-]{35}$/,
+};
+
+async function loadKeys() {
+    try {
+        const r = await fetch(`${API}/api/keys`);
+        const d = await r.json();
+        const grid = document.getElementById('keys-grid');
+        grid.innerHTML = d.map(k => {
+            const hint = k.validated && k.example ? `<div class="key-hint">expected: ${k.example}</div>` : '';
+            return `<div class="key-card ${k.set?'set':''}">
+                <div class="key-head">
+                    <span class="key-name">${k.name}</span>
+                    <span class="key-status ${k.set?'set':'missing'}">${k.set?'●':'○'} ${k.set ? (k.masked||'set') : 'missing'}</span>
+                </div>
+                <div class="key-desc">${k.description || 'Custom key (not in registry).'}${hint}</div>
+                <div class="key-form">
+                    <input type="password" id="key-in-${k.name}" placeholder="Paste new value…" autocomplete="off" spellcheck="false">
+                    <button class="btn" onclick="saveKey('${k.name}')">Save</button>
+                    ${k.set ? `<button class="btn btn-danger" onclick="clearKey('${k.name}')">Clear</button>` : ''}
+                </div>
+                <div class="key-msg" id="key-msg-${k.name}"></div>
+            </div>`;
+        }).join('');
+    } catch(e) { console.error('Keys load failed', e); }
+}
+
+function keyMsg(name, text, isErr) {
+    const el = document.getElementById('key-msg-'+name);
+    if (el) { el.textContent = text; el.className = 'key-msg ' + (isErr ? 'err' : 'ok'); }
+}
+
+async function saveKey(name) {
+    const input = document.getElementById('key-in-'+name);
+    const value = (input.value||'').trim();
+    if (!value) { keyMsg(name, 'Enter a value to save.', true); return; }
+    const pat = KEY_PATTERNS[name];
+    if (pat && !pat.test(value)) { keyMsg(name, 'Format check failed — see the expected format above.', true); return; }
+    try {
+        const r = await fetch(`${API}/api/keys/${encodeURIComponent(name)}`, {
+            method: 'POST',
+            headers: {'Content-Type': 'application/json'},
+            body: JSON.stringify({value})
+        });
+        const d = await r.json().catch(() => ({}));
+        if (!r.ok) { keyMsg(name, d.detail || 'Save failed', true); return; }
+        input.value = '';
+        keyMsg(name, `Saved (${d.masked || 'set'}).`, false);
+        loadKeys();
+    } catch(e) { keyMsg(name, 'Network error.', true); }
+}
+
+async function clearKey(name) {
+    if (!confirm(`Delete ${name}? This cannot be undone.`)) return;
+    try {
+        const r = await fetch(`${API}/api/keys/${encodeURIComponent(name)}`, {method:'DELETE'});
+        if (!r.ok) { keyMsg(name, 'Clear failed.', true); return; }
+        keyMsg(name, 'Key removed.', false);
+        loadKeys();
+    } catch(e) { keyMsg(name, 'Network error.', true); }
+}
+
 function showTab(name) {
-    ['recent','alerts','entities','ingest'].forEach(t => {
+    ['recent','alerts','entities','ingest','keys'].forEach(t => {
         document.getElementById('tab-'+t).style.display = t===name?'block':'none';
     });
     document.querySelectorAll('.tab-bar .btn').forEach((b,i) => {
-        b.classList.toggle('active', ['recent','alerts','entities','ingest'][i]===name);
+        b.classList.toggle('active', ['recent','alerts','entities','ingest','keys'][i]===name);
     });
     if (name==='alerts') loadAlerts();
     if (name==='entities') loadEntities();
+    if (name==='keys') loadKeys();
 }
 
 async function ingestRSS() {

docker-compose.yml

@@ -69,6 +69,12 @@ services:
       GDELT_QUERY: ${GDELT_QUERY:-}
       INGEST_INTERVAL: ${INGEST_INTERVAL:-300}
       INGEST_EARTHQUAKES: ${INGEST_EARTHQUAKES:-1}
+      # ── NASA FIRMS active fires ──
+      INGEST_FIRES: ${INGEST_FIRES:-1}
+      FIRMS_MAP_KEY: ${FIRMS_MAP_KEY:-}
+      FIRMS_DATASET: ${FIRMS_DATASET:-VIIRS_SNPP_NRT}
+      FIRMS_BBOX: ${FIRMS_BBOX:--180,-60,180,75}
+      FIRMS_INTERVAL: ${FIRMS_INTERVAL:-900}
     command: ["python", "app/run_ingester.py"]
     entrypoint: ["python", "app/run_ingester.py"]
 
@@ -94,6 +100,10 @@ services:
       MINIO_ACCESS_KEY: ${MINIO_ACCESS_KEY:-}
       MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:-}
       MINIO_SECURE: ${MINIO_SECURE:-false}
+      # ── NASA FIRMS (for the /api/ingest/fires trigger endpoint) ──
+      FIRMS_MAP_KEY: ${FIRMS_MAP_KEY:-}
+      FIRMS_DATASET: ${FIRMS_DATASET:-VIIRS_SNPP_NRT}
+      FIRMS_BBOX: ${FIRMS_BBOX:--180,-60,180,75}
     ports:
       - "127.0.0.1:8000:8000"
     healthcheck:

docs/firms.md

@@ -0,0 +1,145 @@
+# FIRMS active fire / hotspot heatmap — data source
+
+The OSINT map's fire overlay is fed by **NASA FIRMS** (Fire Information for
+Resource Management System). Two zero-cost options exist; this repo implements
+option A (ingested vector points served as JSON), and option B (GIBS raster
+tiles) is documented below for a no-storage frontend-only alternative.
+
+---
+
+## A. FIRMS area CSV → Postgres/TimescaleDB → `GET /api/fires`
+
+### Data flow
+
+```
+NASA FIRMS area CSV  ──►  app/fire_sources.py  ──►  NATS events.fire
+                                                          │
+                                              app/ingestor.py (ingest_fire_row)
+                                                          │
+                                          fires hypertable (idempotent PK)
+                                                          │
+                                          GET /api/fires?bbox=&since=  (JSON)
+```
+
+* `app/fire_sources.py` fetches the VIIRS active-fire CSV for a bounding box,
+  normalises each row (combining `acq_date` + `acq_time` into a UTC timestamp),
+  and publishes one message per hotspot to NATS JetStream subject `events.fire`.
+* The long-running ingester (`app/run_ingester.py`) polls FIRMS on its own
+  ~15-minute cadence (`FIRMS_INTERVAL`, default 900 s) and shares the existing
+  NATS consumer. `ingest_event` routes `source_type == "fire"` messages to
+  `ingest_fire_row`, which writes to the `fires` table.
+* **Idempotency:** the `fires` primary key IS the natural key
+  `(latitude, longitude, acq_time, satellite)`. Inserts use
+  `ON CONFLICT DO NOTHING`, so a hotspot re-delivered on a later poll is
+  silently ignored — no duplicates, no upsert churn.
+* The `fires` table is a **TimescaleDB hypertable** partitioned on `acq_time`
+  (1-day chunks), so retention is one `drop_chunks` call away.
+
+### Endpoint
+
+```
+GET /api/fires?bbox=<minlon,minlat,maxlon,maxlat>&since=<ISO-8601-UTC>&limit=<N>
+```
+
+| Query param | Meaning | Default |
+|---|---|---|
+| `bbox` | `"minlon,minlat,maxlon,maxlat"` to bound the result (e.g. `-125,24,-66,50`). | all stored detections |
+| `since` | only hotspots acquired at/after this UTC instant (e.g. `2026-08-24T12:00:00Z`). | none |
+| `limit` | max rows returned. | `2000` (max `10000`) |
+
+Response is a plain JSON array (frontend renders it as the heatmap overlay):
+
+```json
+[
+  {
+    "latitude": 39.45678,
+    "longitude": -121.12345,
+    "brightness": 341.4,        // bright_ti4, Kelvin
+    "confidence": "h",          // VIIRS: n (nominal) / l (low) / h (high)
+    "acq_time": "2026-08-24T18:10:00Z",
+    "satellite": "N",           // N (S-NPP), N20, N21
+    "instrument": "VIIRS",
+    "bright_ti5": 310.2,        // 12µm brightness, Kelvin
+    "frp": 12.4,                // fire radiative power, MW
+    "daynight": "D"             // D / N
+  }
+]
+```
+
+A manual poll can also be triggered with `POST /api/ingest/fires?bbox=...`.
+
+### Configuration (all via env / `.env`)
+
+| Var | Default | Notes |
+|---|---|---|
+| `FIRMS_MAP_KEY` | *(blank)* | **Required for live data.** Free key: <https://firms.modaps.eosdis.nasa.gov/api/map_key_info/> (1-minute signup). Until set, the fire loop logs a warning and stays idle — it never crashes the ingester. |
+| `FIRMS_DATASET` | `VIIRS_SNPP_NRT` | NRT VIIRS S-NPP 375 m active fire detection. |
+| `FIRMS_BBOX` | `-180,-60,180,75` | Poll area `"minlon,minlat,maxlon,maxlat"`. Narrow it (e.g. `-125,24,-66,50`) to cut payload and write volume. |
+| `FIRMS_INTERVAL` | `900` | Poll cadence in seconds (~15 min; FIRMS NRT refreshes every ~5–10 min). |
+| `INGEST_FIRES` | `1` | Set `0` to disable the fire loop entirely. |
+| `FIRMS_TIMEOUT` | `60` | Outbound HTTP timeout for the CSV download. |
+
+### Tests
+
+`tests/` — parser/normalisation/mapping unit tests run anywhere; DB-backed
+idempotency + API contract tests are marked `integration` and auto-skip without
+a reachable TimescaleDB+PostGIS (the repo's `localhost/osint-dashboard-pg`
+image or a local `postgis/postgis`):
+
+```bash
+DB_HOST=... DB_PORT=... DB_USER=osint DB_PASSWORD=... DB_NAME=osint_data \
+  pytest tests/ -v
+```
+
+`DB_NULL_POOL=1` is set by the test suite (fresh connection per event loop).
+
+### Live verification
+
+Live end-to-end verification (real FIRMS fetch → NATS → Postgres → API) is
+**blocked until `FIRMS_MAP_KEY` is set** in `.env`. Everything else — CSV
+parsing, idempotent storage, the API contract — is verified against a real
+TimescaleDB instance in the test suite.
+
+---
+
+## B. GIBS thermal-anomaly tiles — zero-cost, no key, no storage
+
+If you want a fire layer with **zero backend work** (raster tiles rendered by
+the map library directly, no ingest, no DB, no API key), NASA GIBS serves the
+same VIIRS S-NPP detections as WMTS tiles:
+
+* Layer: **`VIIRS_SNPP_Thermal_Anomalies_375m_All`** (375 m VIIRS S-NPP thermal
+  anomalies / active fires). Sibling layers exist for day/night-only views
+  (`..._Day`, `..._Night`).
+* REST tile URL (Web Mercator, EPSG:3857 — what Leaflet/MapLibre use):
+
+  ```
+  https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/VIIRS_SNPP_Thermal_Anomalies_375m_All/default/{Time}/GoogleMapsCompatible_Level{Z}/{Y}/{X}.png
+  ```
+
+* `{Time}` is a date like `2026-08-24` (or a time-of-day string); the list of
+  available times comes from the WMTS capabilities:
+  `https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/1.0.0/WMTSCapabilities.xml`
+  (search for the layer, read its `Dimension` → `Value`).
+* Leaflet/MapLibre example:
+
+  ```js
+  L.tileLayer(
+    'https://gibs.earthdata.nasa.gov/wmts/epsg3857/best/VIIRS_SNPP_Thermal_Anomalies_375m_All/default/{time}/GoogleMapsCompatible_Level{z}/{y}/{x}.png',
+    { attribution: 'NASA GIBS / FIRMS', maxZoom: 9 }
+  ).addTo(map);
+  ```
+
+**Trade-offs vs. option A:**
+
+| | A — FIRMS CSV ingest | B — GIBS WMTS tiles |
+|---|---|---|
+| Data in our DB | yes (queryable, filterable) | no (pixels only) |
+| Per-hotspot attributes (brightness, FRP, confidence) | yes | no (colour-coded only) |
+| Time range / `since` filtering server-side | yes | tile `{Time}` per snapshot |
+| Backend cost | ingest service + DB rows | none |
+| API key | free FIRMS_MAP_KEY | none |
+
+GIBS is the right choice when the map only needs "where are fires right now".
+The FIRMS ingest is right when you want to query, aggregate, or persist the
+detections (e.g. "fires near X in the last 24 h").

tests/conftest.py

@@ -0,0 +1,111 @@
+"""Shared test fixtures.
+
+Unit tests (parsing/mapping) run anywhere. DB-backed tests (idempotency, API)
+are marked `integration` and auto-skip when the test database is unreachable —
+set DB_HOST/DB_PORT/DB_USER/DB_PASSWORD/DB_NAME to point at a TimescaleDB+PostGIS
+instance (e.g. the `localhost/osint-dashboard-pg:test` image) to run them.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+# Make the app package importable from tests (repo root/app).
+APP_DIR = Path(__file__).resolve().parent.parent / "app"
+if str(APP_DIR) not in sys.path:
+    sys.path.insert(0, str(APP_DIR))
+
+# Point the app's config at the test database BEFORE any app module is imported
+# (database.py builds DATABASE_URL from env at import time).
+os.environ.setdefault("DB_HOST", "127.0.0.1")
+os.environ.setdefault("DB_PORT", "55432")
+os.environ.setdefault("DB_USER", "osint")
+os.environ.setdefault("DB_PASSWORD", "osint")
+os.environ.setdefault("DB_NAME", "osint_data")
+# Each test opens a fresh event loop (asyncio.run); a pooled connection can't be
+# reused across loops, so force a NullPool (new connection per session).
+os.environ.setdefault("DB_NULL_POOL", "1")
+
+import asyncpg  # noqa: E402
+
+pytestmark = []
+
+
+def _db_reachable() -> bool:
+    try:
+        import asyncio
+
+        async def _ping() -> bool:
+            try:
+                conn = await asyncpg.connect(
+                    host=os.environ["DB_HOST"],
+                    port=int(os.environ["DB_PORT"]),
+                    user=os.environ["DB_USER"],
+                    password=os.environ["DB_PASSWORD"],
+                    database=os.environ["DB_NAME"],
+                    timeout=3,
+                )
+                await conn.close()
+                return True
+            except Exception:
+                return False
+
+        return asyncio.run(_ping())
+    except Exception:
+        return False
+
+
+requires_db = pytest.mark.skipif(
+    not _db_reachable(),
+    reason="test database unreachable (set DB_* env or start the PG container)",
+)
+
+
+@pytest.fixture()
+def clean_fires():
+    """Truncate the fires table before and after a DB-backed test."""
+    import asyncio
+
+    async def _truncate():
+        conn = await asyncpg.connect(
+            host=os.environ["DB_HOST"],
+            port=int(os.environ["DB_PORT"]),
+            user=os.environ["DB_USER"],
+            password=os.environ["DB_PASSWORD"],
+            database=os.environ["DB_NAME"],
+        )
+        try:
+            await conn.execute("TRUNCATE fires")
+        finally:
+            await conn.close()
+
+    asyncio.run(_truncate())
+    yield
+    asyncio.run(_truncate())
+
+
+def make_fire_msg(**overrides) -> dict:
+    """A realistic VIIRS hotspot NATS message (as fire_sources publishes it)."""
+    msg = {
+        "source_type": "fire",
+        "latitude": 39.45678,
+        "longitude": -121.12345,
+        "brightness": 341.40,
+        "confidence": "h",
+        "acq_time": "2026-08-24T18:10:00+00:00",
+        "satellite": "N",
+        "instrument": "VIIRS",
+        "bright_ti5": 310.20,
+        "frp": 12.4,
+        "daynight": "D",
+        "scan": 0.45,
+        "track": 0.47,
+        "version": "2.0NRT",
+        "raw": None,
+    }
+    msg.update(overrides)
+    return msg

tests/test_api_fires.py

@@ -0,0 +1,91 @@
+"""Integration tests for GET /api/fires (bbox + since filters, JSON contract)."""
+
+import asyncio
+
+import httpx
+
+from conftest import make_fire_msg, requires_db
+
+from ingestor import ingest_fire_row
+from main import app
+
+BASE = "http://test"
+
+
+def _seed(*msgs) -> None:
+    """Insert fire rows through the real ingestor path (idempotent)."""
+
+    async def run():
+        for m in msgs:
+            await ingest_fire_row(m)
+
+    asyncio.run(run())
+
+
+def _get(path: str) -> httpx.Response:
+    return asyncio.run(_get_async(path))
+
+
+async def _get_async(path: str) -> httpx.Response:
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url=BASE) as client:
+        return await client.get(path)
+
+
+@requires_db
+def test_api_fires_bbox_filter(clean_fires):
+    _seed(
+        make_fire_msg(latitude=39.45678, longitude=-121.12345, satellite="N"),
+        make_fire_msg(latitude=34.00000, longitude=-118.20000, satellite="N20"),
+        make_fire_msg(latitude=10.00000, longitude=20.00000, satellite="N"),
+    )
+    # bbox covering only the California-ish points
+    resp = _get("/api/fires?bbox=-125,30,-115,42")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert isinstance(body, list)
+    assert len(body) == 2
+    lats = sorted(p["latitude"] for p in body)
+    assert lats == [34.0, 39.45678]
+
+
+@requires_db
+def test_api_fires_since_filter(clean_fires):
+    _seed(
+        make_fire_msg(acq_time="2026-08-24T18:10:00+00:00"),
+        make_fire_msg(acq_time="2026-08-24T19:10:00+00:00"),
+    )
+    resp = _get("/api/fires?since=2026-08-24T18:30:00Z")
+    assert resp.status_code == 200
+    body = resp.json()
+    assert len(body) == 1
+    assert body[0]["acq_time"].startswith("2026-08-24T19:10")
+
+
+@requires_db
+def test_api_fires_json_contract(clean_fires):
+    _seed(make_fire_msg())
+    body = _get("/api/fires").json()
+    assert len(body) == 1
+    point = body[0]
+    # The exact fields the frontend heatmap needs — nothing more, nothing less.
+    assert set(point.keys()) == {
+        "latitude", "longitude", "brightness", "confidence", "acq_time",
+        "satellite", "instrument", "bright_ti5", "frp", "daynight",
+    }
+    assert point["latitude"] == 39.45678
+    assert point["confidence"] == "h"
+    assert point["satellite"] == "N"
+
+
+@requires_db
+def test_api_fires_invalid_bbox_422(clean_fires):
+    assert _get("/api/fires?bbox=-125,30").status_code == 422
+    assert _get("/api/fires?bbox=-125,abc,-115,42").status_code == 422
+
+
+@requires_db
+def test_api_fires_empty(clean_fires):
+    resp = _get("/api/fires")
+    assert resp.status_code == 200
+    assert resp.json() == []

tests/test_fire_ingest.py

@@ -0,0 +1,87 @@
+"""Unit + integration tests for fire message routing and idempotent storage.
+
+The row-mapping logic (_fire_row_from_msg) is pure and tested everywhere; the
+DB-backed idempotency tests are `integration` and auto-skip without a live DB.
+"""
+
+import asyncio
+from datetime import datetime, timezone
+
+from conftest import make_fire_msg, requires_db
+
+from ingestor import _fire_row_from_msg, ingest_fire_row, ingest_event
+
+
+# ── Row mapping (pure, no DB) ────────────────────────────────────────────
+
+def test_fire_row_from_msg_maps_fields():
+    row = _fire_row_from_msg(make_fire_msg())
+    assert row is not None
+    assert row["latitude"] == 39.45678
+    assert row["longitude"] == -121.12345
+    assert row["brightness"] == 341.40
+    assert row["confidence"] == "h"
+    assert row["satellite"] == "N"
+    assert row["acq_time"] == datetime(2026, 8, 24, 18, 10, tzinfo=timezone.utc)
+    assert row["frp"] == 12.4
+    assert row["daynight"] == "D"
+
+
+def test_fire_row_from_msg_rejects_missing_natural_key():
+    for key in ("latitude", "longitude", "acq_time", "satellite"):
+        msg = make_fire_msg(**{key: None})
+        assert _fire_row_from_msg(msg) is None, f"{key}=None should be dropped"
+
+
+def test_fire_row_from_msg_rejects_bad_timestamp():
+    msg = make_fire_msg(acq_time="not-a-timestamp")
+    assert _fire_row_from_msg(msg) is None
+
+
+def test_ingest_event_routes_fire_away_from_events(monkeypatch):
+    """source_type='fire' must hit the fire path, never the generic events insert."""
+    calls = {"fire": 0, "events": 0}
+
+    async def fake_fire_row(msg):
+        calls["fire"] += 1
+        return True
+
+    async def boom(*a, **k):  # the generic events insert must not be reached
+        calls["events"] += 1
+        raise AssertionError("fire message leaked into the events insert path")
+
+    import ingestor
+    monkeypatch.setattr(ingestor, "ingest_fire_row", fake_fire_row)
+    monkeypatch.setattr(ingestor.events_table, "insert", boom)
+
+    asyncio.run(ingest_event(make_fire_msg()))
+    assert calls["fire"] == 1
+    assert calls["events"] == 0
+
+
+# ── Idempotent storage against a live database ───────────────────────────
+
+@requires_db
+def test_ingest_fire_row_idempotent(clean_fires):
+    async def run():
+        # First insert persists.
+        assert await ingest_fire_row(make_fire_msg()) is True
+        # Same natural key on a later 15-min poll -> silently ignored.
+        assert await ingest_fire_row(make_fire_msg(brightness=999.0)) is False
+        # Same point/time but a different satellite is a distinct detection.
+        assert await ingest_fire_row(make_fire_msg(satellite="N20")) is True
+        # Same point but a different acquisition time is a distinct detection.
+        assert await ingest_fire_row(
+            make_fire_msg(acq_time="2026-08-24T19:10:00+00:00")
+        ) is True
+
+    asyncio.run(run())
+
+
+@requires_db
+def test_ingest_fire_row_drops_malformed(clean_fires):
+    async def run():
+        assert await ingest_fire_row(make_fire_msg(latitude=None)) is False
+        assert await ingest_fire_row(make_fire_msg(acq_time="garbage")) is False
+
+    asyncio.run(run())

tests/test_fire_sources.py

@@ -0,0 +1,78 @@
+"""Unit tests for the NASA FIRMS CSV parser / timestamp normalization."""
+
+import asyncio
+from datetime import datetime, timezone
+
+from fire_sources import normalize_acq_time, parse_firms_csv, ingest_fires
+
+# Grounded against real FIRMS VIIRS area-CSV output.
+SAMPLE_CSV = """latitude,longitude,bright_ti4,scan,track,acq_date,acq_time,satellite,instrument,confidence,version,bright_ti5,frp,daynight
+-16.28359,29.40531,295.78,0.50,0.66,2025-06-06,1,N20,VIIRS,n,2.0NRT,284.11,1.17,N
+-16.28190,29.40279,303.31,0.50,0.66,2025-06-06,1,N20,VIIRS,n,2.0NRT,284.43,0.67,N
+-14.98900,28.36286,341.04,0.41,0.60,2025-06-06,1,N20,VIIRS,n,2.0NRT,279.77,4.59,N
+15.17397,-11.28343,341.40,0.45,0.47,2025-06-06,1425,N20,VIIRS,l,2.0NRT,315.18,7.91,D
+15.45870,-11.13616,339.25,0.46,0.47,2025-06-06,1425,N20,VIIRS,l,2.0NRT,312.05,9.24,D
+"""
+
+
+def test_normalize_acq_time_single_digit_hour():
+    # acq_time "1" (HHMM int) -> 00:01 UTC
+    dt = normalize_acq_time("2025-06-06", "1")
+    assert dt == datetime(2025, 6, 6, 0, 1, tzinfo=timezone.utc)
+
+
+def test_normalize_acq_time_four_digit():
+    dt = normalize_acq_time("2025-06-06", 1425)
+    assert dt == datetime(2025, 6, 6, 14, 25, tzinfo=timezone.utc)
+
+
+def test_normalize_acq_time_bad_values():
+    assert normalize_acq_time(None, 100) is None
+    assert normalize_acq_time("2025-06-06", None) is None
+    assert normalize_acq_time("2025-06-06", "") is None
+    assert normalize_acq_time("not-a-date", 100) is None
+
+
+def test_parse_firms_csv_happy_path():
+    points = parse_firms_csv(SAMPLE_CSV)
+    assert len(points) == 5
+
+    first = points[0]
+    assert first["latitude"] == -16.28359
+    assert first["longitude"] == 29.40531
+    assert first["brightness"] == 295.78
+    assert first["confidence"] == "n"
+    assert first["satellite"] == "N20"
+    assert first["instrument"] == "VIIRS"
+    assert first["frp"] == 1.17
+    assert first["daynight"] == "N"
+    assert first["acq_time"] == "2025-06-06T00:01:00+00:00"
+
+    # late-day acquisition (acq_time 1425) parses to 14:25 UTC
+    assert points[3]["acq_time"] == "2025-06-06T14:25:00+00:00"
+    assert points[3]["confidence"] == "l"
+
+
+def test_parse_firms_csv_skips_legend_line():
+    # FIRMS occasionally prepends a legend/info line before the real header.
+    with_legend = (
+        "Active Fire Data from VIIRS (S-NPP) — near real time\n"
+        + SAMPLE_CSV
+    )
+    points = parse_firms_csv(with_legend)
+    assert len(points) == 5
+    assert points[0]["latitude"] == -16.28359
+
+
+def test_parse_firms_csv_empty_and_garbage():
+    assert parse_firms_csv("") == []
+    assert parse_firms_csv("not a csv at all\njust text\n") == []
+    # Header present but a data row that is too short / has junk lat-lon
+    junk = SAMPLE_CSV.splitlines()[0] + "\n1,2\n"
+    assert parse_firms_csv(junk) == []
+
+
+def test_ingest_fires_idles_without_map_key(monkeypatch):
+    # No key -> returns 0 without attempting a network call.
+    monkeypatch.setattr("fire_sources.FIRMS_MAP_KEY", "")
+    assert asyncio.run(ingest_fires()) == 0