yaris-obd/server/modules/db.py

"""Postgres connection pool + small query helpers for the yaris-obd remote server."""
import os
from contextlib import contextmanager

from psycopg_pool import ConnectionPool

_POOL = None


def _dsn():
    return (
        f"host={os.environ.get('POSTGRES_HOST', 'db')} "
        f"port={os.environ.get('POSTGRES_PORT', '5432')} "
        f"dbname={os.environ.get('POSTGRES_DB', 'yarisobd')} "
        f"user={os.environ.get('POSTGRES_USER', 'yarisobd')} "
        f"password={os.environ.get('POSTGRES_PASSWORD', '')}"
    )


def get_pool():
    global _POOL
    if _POOL is None:
        _POOL = ConnectionPool(conninfo=_dsn(), min_size=1, max_size=5, open=True)
    return _POOL


@contextmanager
def get_cursor(commit=False):
    """Yield a cursor from the pool, optionally committing on clean exit."""
    pool = get_pool()
    with pool.connection() as conn:
        with conn.cursor() as cur:
            yield cur
            if commit:
                conn.commit()


def get_device_by_name(name):
    with get_cursor() as cur:
        cur.execute(
            "SELECT id, name, api_key_hash FROM devices WHERE name = %s",
            (name,),
        )
        row = cur.fetchone()
        if row is None:
            return None
        return {"id": row[0], "name": row[1], "api_key_hash": row[2]}


def get_device_by_id(device_id):
    """Get device by ID, including config + sync metadata."""
    with get_cursor() as cur:
        cur.execute(
            """
            SELECT id, name, api_key_hash, owner_name, car_model, device_name, avatar_path,
                   config_version, config_updated_at, created_at, last_seen_at,
                   last_config_sync_at, last_config_sync_version, force_sync_requested_at,
                   api_key_rotated_at
            FROM devices
            WHERE id = %s
            """,
            (device_id,),
        )
        row = cur.fetchone()
        if row is None:
            return None
        return {
            "id": row[0],
            "name": row[1],
            "api_key_hash": row[2],
            "owner_name": row[3],
            "car_model": row[4],
            "device_name": row[5],
            "avatar_path": row[6],
            "config_version": row[7],
            "config_updated_at": row[8],
            "created_at": row[9],
            "last_seen_at": row[10],
            "last_config_sync_at": row[11],
            "last_config_sync_version": row[12],
            "force_sync_requested_at": row[13],
            "api_key_rotated_at": row[14],
        }


def touch_device_last_seen(device_id):
    with get_cursor(commit=True) as cur:
        cur.execute(
            "UPDATE devices SET last_seen_at = now() WHERE id = %s",
            (device_id,),
        )


def list_devices():
    with get_cursor() as cur:
        cur.execute(
            """
            SELECT id, name, owner_name, car_model, device_name, avatar_path, config_version,
                   config_updated_at, created_at, last_seen_at, last_config_sync_at,
                   last_config_sync_version, force_sync_requested_at, api_key_rotated_at
            FROM devices
            ORDER BY created_at ASC, id ASC
            """
        )
        rows = cur.fetchall()
    return [
        {
            "id": row[0],
            "name": row[1],
            "owner_name": row[2],
            "car_model": row[3],
            "device_name": row[4],
            "avatar_path": row[5],
            "config_version": row[6],
            "config_updated_at": row[7],
            "created_at": row[8],
            "last_seen_at": row[9],
            "last_config_sync_at": row[10],
            "last_config_sync_version": row[11],
            "force_sync_requested_at": row[12],
            "api_key_rotated_at": row[13],
        }
        for row in rows
    ]


def create_device(name, api_key_hash):
    with get_cursor(commit=True) as cur:
        cur.execute(
            """
            INSERT INTO devices (name, api_key_hash, api_key_rotated_at)
            VALUES (%s, %s, now())
            RETURNING id
            """,
            (name, api_key_hash),
        )
        row = cur.fetchone()
    return row[0]


def rotate_device_api_key(device_id, api_key_hash):
    with get_cursor(commit=True) as cur:
        cur.execute(
            """
            UPDATE devices
            SET api_key_hash = %s, api_key_rotated_at = now()
            WHERE id = %s
            RETURNING id
            """,
            (api_key_hash, device_id),
        )
        row = cur.fetchone()
    return row[0] if row else None


def request_device_sync(device_id):
    with get_cursor(commit=True) as cur:
        cur.execute(
            """
            UPDATE devices
            SET config_version = config_version + 1,
                config_updated_at = now(),
                force_sync_requested_at = now()
            WHERE id = %s
            RETURNING config_version
            """,
            (device_id,),
        )
        row = cur.fetchone()
    return row[0] if row else None


def acknowledge_device_sync(device_id, config_version):
    with get_cursor(commit=True) as cur:
        cur.execute(
            """
            UPDATE devices
            SET last_config_sync_at = now(),
                last_config_sync_version = %s,
                force_sync_requested_at = CASE
                    WHEN config_version <= %s THEN NULL
                    ELSE force_sync_requested_at
                END
            WHERE id = %s
            """,
            (config_version, config_version, device_id),
        )


def delete_device(device_id):
    with get_cursor(commit=True) as cur:
        cur.execute("DELETE FROM devices WHERE id = %s RETURNING id", (device_id,))
        row = cur.fetchone()
    return row[0] if row else None


def insert_tracking_points(device_id, points):
    """Bulk-insert tracking points, skipping any that already exist
    (device_id, recorded_at) so a client can safely re-send a batch after a
    partial-failure retry without creating duplicates."""
    if not points:
        return 0
    with get_cursor(commit=True) as cur:
        cur.executemany(
            """
            INSERT INTO tracking_points
                (device_id, vehicle_name, driver_name, recorded_at, gps_lat, gps_lon, speed, odometer)
            VALUES (%(device_id)s, %(vehicle_name)s, %(driver_name)s, %(recorded_at)s,
                    %(gps_lat)s, %(gps_lon)s, %(speed)s, %(odometer)s)
            ON CONFLICT (device_id, recorded_at) DO NOTHING
            """,
            [
                {
                    "device_id": device_id,
                    "vehicle_name": p.get("vehicle_name"),
                    "driver_name": p.get("driver_name"),
                    "recorded_at": p["recorded_at"],
                    "gps_lat": p.get("gps_lat"),
                    "gps_lon": p.get("gps_lon"),
                    "speed": p.get("speed"),
                    "odometer": p.get("odometer"),
                }
                for p in points
            ],
        )
    return len(points)


def dashcam_file_exists(device_id, relative_path, sha256):
    with get_cursor() as cur:
        cur.execute(
            "SELECT 1 FROM dashcam_files WHERE device_id = %s AND relative_path = %s AND sha256 = %s",
            (device_id, relative_path, sha256),
        )
        return cur.fetchone() is not None


def insert_dashcam_file(device_id, relative_path, size_bytes, sha256, storage_path):
    with get_cursor(commit=True) as cur:
        cur.execute(
            """
            INSERT INTO dashcam_files (device_id, relative_path, size_bytes, sha256, storage_path)
            VALUES (%s, %s, %s, %s, %s)
            ON CONFLICT (device_id, relative_path, sha256) DO NOTHING
            """,
            (device_id, relative_path, size_bytes, sha256, storage_path),
        )


# ── User auth ──────────────────────────────────────────────────────────────

def get_user_by_id(user_id):
    with get_cursor() as cur:
        cur.execute("SELECT id, username, password_hash FROM users WHERE id = %s", (user_id,))
        row = cur.fetchone()
        return {"id": row[0], "username": row[1], "password_hash": row[2]} if row else None


def get_user_by_username(username):
    with get_cursor() as cur:
        cur.execute("SELECT id, username, password_hash FROM users WHERE username = %s", (username,))
        row = cur.fetchone()
        return {"id": row[0], "username": row[1], "password_hash": row[2]} if row else None


def create_user(username, password_hash):
    with get_cursor(commit=True) as cur:
        cur.execute(
            "INSERT INTO users (username, password_hash) VALUES (%s, %s) ON CONFLICT (username) DO NOTHING RETURNING id",
            (username, password_hash),
        )
        row = cur.fetchone()
        return row[0] if row else None


# ── Server-side trips ──────────────────────────────────────────────────────
# Gap threshold: if two consecutive points are more than this many seconds
# apart, they belong to different trips.
_TRIP_GAP_S = 300  # 5 minutes


def compute_trips_for_device(device_id):
    """Recompute trip boundaries from tracking_points and upsert into trips table."""
    with get_cursor() as cur:
        cur.execute(
            """
            SELECT recorded_at, gps_lat, gps_lon, speed
            FROM tracking_points
            WHERE device_id = %s AND gps_lat IS NOT NULL AND gps_lon IS NOT NULL
            ORDER BY recorded_at ASC
            """,
            (device_id,),
        )
        rows = cur.fetchall()

    if not rows:
        return 0

    import math

    def haversine(lat1, lon1, lat2, lon2):
        R = 6371.0
        dlat = math.radians(lat2 - lat1)
        dlon = math.radians(lon2 - lon1)
        a = math.sin(dlat/2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon/2)**2
        return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))

    # Split into segments by gap
    segments = []
    current = [rows[0]]
    for prev, curr in zip(rows, rows[1:]):
        gap = (curr[0] - prev[0]).total_seconds()
        if gap > _TRIP_GAP_S:
            segments.append(current)
            current = [curr]
        else:
            current.append(curr)
    segments.append(current)

    # Build trip records
    trips = []
    for seg in segments:
        if len(seg) < 2:
            continue
        start_time = seg[0][0]
        end_time = seg[-1][0]
        duration_s = int((end_time - start_time).total_seconds())
        if duration_s < 30:
            continue
        dist = sum(
            haversine(seg[i][1], seg[i][2], seg[i+1][1], seg[i+1][2])
            for i in range(len(seg) - 1)
        )
        if dist < 0.05:  # skip <50m blips
            continue
        speeds = [r[3] for r in seg if r[3] is not None]
        trips.append({
            "start_time": start_time,
            "end_time": end_time,
            "distance_km": round(dist, 3),
            "duration_s": duration_s,
            "avg_speed_kmh": round(sum(speeds) / len(speeds), 1) if speeds else None,
            "max_speed_kmh": round(max(speeds), 1) if speeds else None,
            "start_lat": seg[0][1],
            "start_lon": seg[0][2],
            "end_lat": seg[-1][1],
            "end_lon": seg[-1][2],
            "point_count": len(seg),
        })

    with get_cursor(commit=True) as cur:
        for t in trips:
            cur.execute(
                """
                INSERT INTO trips (device_id, start_time, end_time, distance_km, duration_s,
                    avg_speed_kmh, max_speed_kmh, start_lat, start_lon, end_lat, end_lon,
                    point_count, computed_at)
                VALUES (%(device_id)s, %(start_time)s, %(end_time)s, %(distance_km)s, %(duration_s)s,
                    %(avg_speed_kmh)s, %(max_speed_kmh)s, %(start_lat)s, %(start_lon)s,
                    %(end_lat)s, %(end_lon)s, %(point_count)s, now())
                ON CONFLICT (device_id, start_time) DO UPDATE SET
                    end_time = EXCLUDED.end_time,
                    distance_km = EXCLUDED.distance_km,
                    duration_s = EXCLUDED.duration_s,
                    avg_speed_kmh = EXCLUDED.avg_speed_kmh,
                    max_speed_kmh = EXCLUDED.max_speed_kmh,
                    end_lat = EXCLUDED.end_lat,
                    end_lon = EXCLUDED.end_lon,
                    point_count = EXCLUDED.point_count,
                    computed_at = now()
                """,
                {"device_id": device_id, **t},
            )
    return len(trips)


def list_trips(device_id, limit=200):
    with get_cursor() as cur:
        cur.execute(
            """
            SELECT id, start_time, end_time, distance_km, duration_s,
                   avg_speed_kmh, max_speed_kmh, start_lat, start_lon,
                   end_lat, end_lon, point_count
            FROM trips
            WHERE device_id = %s
            ORDER BY start_time DESC
            LIMIT %s
            """,
            (device_id, limit),
        )
        rows = cur.fetchall()
    return [
        {
            "id": r[0],
            "start_datetime": r[1].isoformat(),
            "end_datetime": r[2].isoformat(),
            "distance_km": r[3],
            "duration_seconds": r[4],
            "average_speed_kmh": r[5],
            "max_speed_kmh": r[6],
            "start_lat": r[7],
            "start_lon": r[8],
            "end_lat": r[9],
            "end_lon": r[10],
            "point_count": r[11],
            "odo_start_km": None,
            "odo_end_km": None,
        }
        for r in rows
    ]


def get_trip_route(device_id, trip_id):
    with get_cursor() as cur:
        cur.execute(
            "SELECT start_time, end_time, distance_km, duration_s, avg_speed_kmh, max_speed_kmh FROM trips WHERE id = %s AND device_id = %s",
            (trip_id, device_id),
        )
        trip_row = cur.fetchone()
        if trip_row is None:
            return None
        cur.execute(
            """
            SELECT recorded_at, gps_lat, gps_lon, speed
            FROM tracking_points
            WHERE device_id = %s AND recorded_at BETWEEN %s AND %s
              AND gps_lat IS NOT NULL AND gps_lon IS NOT NULL
            ORDER BY recorded_at ASC
            """,
            (device_id, trip_row[0], trip_row[1]),
        )
        points = cur.fetchall()
    return {
        "trip": {
            "distance_km": trip_row[2],
            "duration_seconds": trip_row[3],
            "average_speed_kmh": trip_row[4],
            "max_speed_kmh": trip_row[5],
        },
        "points": [
            {
                "datetime": p[0].isoformat(),
                "latitude": p[1],
                "longitude": p[2],
                "speed": p[3] or 0,
                "speed_source": "gps",
            }
            for p in points
        ],
    }


def list_dashcam_files(device_id, limit=500):
    with get_cursor() as cur:
        cur.execute(
            """
            SELECT id, relative_path, size_bytes, uploaded_at, storage_path
            FROM dashcam_files
            WHERE device_id = %s
            ORDER BY uploaded_at DESC
            LIMIT %s
            """,
            (device_id, limit),
        )
        rows = cur.fetchall()
    return [
        {
            "id": r[0],
            "relative_path": r[1],
            "size_bytes": r[2],
            "uploaded_at": r[3].isoformat(),
            "storage_path": r[4],
        }
        for r in rows
    ]


def get_dashcam_file(device_id, file_id):
    with get_cursor() as cur:
        cur.execute(
            "SELECT id, relative_path, size_bytes, uploaded_at, storage_path FROM dashcam_files WHERE id = %s AND device_id = %s",
            (file_id, device_id),
        )
        r = cur.fetchone()
    if r is None:
        return None
    return {"id": r[0], "relative_path": r[1], "size_bytes": r[2], "uploaded_at": r[3].isoformat(), "storage_path": r[4]}


def delete_trip(device_id, trip_id):
    """Delete a computed trip record. Raw tracking_points are NOT removed."""
    with get_cursor(commit=True) as cur:
        cur.execute(
            "DELETE FROM trips WHERE id = %s AND device_id = %s RETURNING id",
            (trip_id, device_id),
        )
        return cur.fetchone() is not None


def delete_dashcam_file(device_id, file_id):
    """Delete dashcam file from DB and disk. Returns storage_path if deleted, None if not found."""
    with get_cursor(commit=True) as cur:
        cur.execute(
            "DELETE FROM dashcam_files WHERE id = %s AND device_id = %s RETURNING storage_path",
            (file_id, device_id),
        )
        row = cur.fetchone()
    return row[0] if row else None


def get_overview_stats(device_id):
    with get_cursor() as cur:
        cur.execute(
            """
            SELECT
                (SELECT COUNT(*) FROM trips WHERE device_id = %s) AS trip_count,
                (SELECT COALESCE(SUM(distance_km), 0) FROM trips WHERE device_id = %s) AS total_km,
                (SELECT COALESCE(SUM(duration_s), 0) FROM trips WHERE device_id = %s) AS total_s,
                (SELECT COALESCE(MAX(max_speed_kmh), 0) FROM trips WHERE device_id = %s) AS top_speed,
                (SELECT COUNT(*) FROM dashcam_files WHERE device_id = %s) AS video_count,
                COALESCE(
                    (SELECT MAX(odometer) FROM tracking_points WHERE device_id = %s AND odometer IS NOT NULL),
                    (SELECT SUM(distance_km) FROM trips WHERE device_id = %s)
                ) AS odometer_km
            """,
            (device_id,) * 7,
        )
        r = cur.fetchone()
    return {
        "trip_count": r[0],
        "total_distance_km": float(r[1]),
        "total_drive_time_s": int(r[2]),
        "top_speed_kmh": float(r[3]),
        "video_count": r[4],
        "odometer_km": float(r[5]) if r[5] is not None else None,
    }