osint-dashboard/alembic/versions/005_phase2.py

"""phase 2: geofences, 1-min track CAGGs, fire/aircraft hits

Revision ID: 005_phase2
Revises: 004_camera_enum
Create Date: 2026-08-28
"""

from alembic import op

revision = "005_phase2"
down_revision = "004_camera_enum"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute("CREATE EXTENSION IF NOT EXISTS postgis")
    op.execute("CREATE EXTENSION IF NOT EXISTS timescaledb")

    op.execute("""
        CREATE TABLE IF NOT EXISTS geofences (
            id UUID PRIMARY KEY,
            name TEXT NOT NULL,
            geojson JSONB NOT NULL,
            geom geometry(Polygon, 4326),
            active INTEGER NOT NULL DEFAULT 1,
            created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
        )
    """)
    op.execute("""
        CREATE INDEX IF NOT EXISTS ix_geofences_geom
        ON geofences USING gist (geom)
    """)

    op.execute("""
        CREATE TABLE IF NOT EXISTS geofence_alerts (
            id UUID PRIMARY KEY,
            geofence_id UUID NOT NULL,
            source_kind TEXT NOT NULL,
            entity_id TEXT NOT NULL,
            lat DOUBLE PRECISION,
            lon DOUBLE PRECISION,
            payload JSONB,
            created_at TIMESTAMPTZ NOT NULL DEFAULT now()
        )
    """)
    op.execute("""
        CREATE INDEX IF NOT EXISTS ix_geofence_alerts_created
        ON geofence_alerts (created_at DESC)
    """)

    op.execute("""
        CREATE TABLE IF NOT EXISTS vessel_positions (
            mmsi TEXT NOT NULL,
            ts TIMESTAMPTZ NOT NULL,
            lat DOUBLE PRECISION NOT NULL,
            lon DOUBLE PRECISION NOT NULL,
            heading DOUBLE PRECISION,
            speed DOUBLE PRECISION,
            label TEXT,
            extra JSONB,
            PRIMARY KEY (mmsi, ts)
        )
    """)
    op.execute("""
        SELECT create_hypertable(
            'vessel_positions', 'ts', if_not_exists => TRUE
        )
    """)
    op.execute("""
        CREATE INDEX IF NOT EXISTS ix_vessel_positions_bbox
        ON vessel_positions (lon, lat)
    """)

    op.execute("""
        CREATE TABLE IF NOT EXISTS aircraft_positions (
            hex TEXT NOT NULL,
            ts TIMESTAMPTZ NOT NULL,
            lat DOUBLE PRECISION NOT NULL,
            lon DOUBLE PRECISION NOT NULL,
            heading DOUBLE PRECISION,
            speed DOUBLE PRECISION,
            label TEXT,
            extra JSONB,
            PRIMARY KEY (hex, ts)
        )
    """)
    op.execute("""
        SELECT create_hypertable(
            'aircraft_positions', 'ts', if_not_exists => TRUE
        )
    """)
    op.execute("""
        CREATE INDEX IF NOT EXISTS ix_aircraft_positions_bbox
        ON aircraft_positions (lon, lat)
    """)

    op.execute("""
        CREATE TABLE IF NOT EXISTS fire_aircraft_hits (
            id UUID NOT NULL,
            fire_id TEXT NOT NULL,
            fire_lat DOUBLE PRECISION NOT NULL,
            fire_lon DOUBLE PRECISION NOT NULL,
            aircraft_hex TEXT NOT NULL,
            aircraft_type TEXT,
            aircraft_lat DOUBLE PRECISION NOT NULL,
            aircraft_lon DOUBLE PRECISION NOT NULL,
            distance_mi DOUBLE PRECISION NOT NULL,
            seen_at TIMESTAMPTZ NOT NULL,
            PRIMARY KEY (fire_id, aircraft_hex, seen_at)
        )
    """)
    op.execute("""
        CREATE INDEX IF NOT EXISTS ix_fire_aircraft_hits_seen
        ON fire_aircraft_hits (seen_at DESC)
    """)

    # 1-minute continuous aggregates (Timescale). last() keeps the newest
    # sample in each bucket — the DVR slider reads these, not the raw table.
    op.execute("""
        DO $$
        BEGIN
            IF NOT EXISTS (
                SELECT 1 FROM timescaledb_information.continuous_aggregates
                WHERE view_name = 'vessel_tracks_1min'
            ) THEN
                EXECUTE $v$
                    CREATE MATERIALIZED VIEW vessel_tracks_1min
                    WITH (timescaledb.continuous) AS
                    SELECT time_bucket('1 minute', ts) AS bucket,
                           mmsi,
                           last(lat, ts) AS lat,
                           last(lon, ts) AS lon,
                           last(heading, ts) AS heading,
                           last(speed, ts) AS speed,
                           last(label, ts) AS label
                    FROM vessel_positions
                    GROUP BY bucket, mmsi
                    WITH NO DATA
                $v$;
            END IF;
            IF NOT EXISTS (
                SELECT 1 FROM timescaledb_information.continuous_aggregates
                WHERE view_name = 'aircraft_tracks_1min'
            ) THEN
                EXECUTE $a$
                    CREATE MATERIALIZED VIEW aircraft_tracks_1min
                    WITH (timescaledb.continuous) AS
                    SELECT time_bucket('1 minute', ts) AS bucket,
                           hex,
                           last(lat, ts) AS lat,
                           last(lon, ts) AS lon,
                           last(heading, ts) AS heading,
                           last(speed, ts) AS speed,
                           last(label, ts) AS label
                    FROM aircraft_positions
                    GROUP BY bucket, hex
                    WITH NO DATA
                $a$;
            END IF;
        END
        $$;
    """)
    op.execute("""
        DO $$
        BEGIN
            PERFORM add_continuous_aggregate_policy(
                'vessel_tracks_1min',
                start_offset => INTERVAL '3 hours',
                end_offset => INTERVAL '1 minute',
                schedule_interval => INTERVAL '1 minute',
                if_not_exists => TRUE
            );
            PERFORM add_continuous_aggregate_policy(
                'aircraft_tracks_1min',
                start_offset => INTERVAL '3 hours',
                end_offset => INTERVAL '1 minute',
                schedule_interval => INTERVAL '1 minute',
                if_not_exists => TRUE
            );
        EXCEPTION WHEN OTHERS THEN
            NULL;
        END
        $$;
    """)


def downgrade() -> None:
    op.execute("DROP MATERIALIZED VIEW IF EXISTS aircraft_tracks_1min CASCADE")
    op.execute("DROP MATERIALIZED VIEW IF EXISTS vessel_tracks_1min CASCADE")
    op.execute("DROP TABLE IF EXISTS fire_aircraft_hits")
    op.execute("DROP TABLE IF EXISTS aircraft_positions")
    op.execute("DROP TABLE IF EXISTS vessel_positions")
    op.execute("DROP TABLE IF EXISTS geofence_alerts")
    op.execute("DROP TABLE IF EXISTS geofences")