src/cache.py

"""
Two-level DuckDB cache.

Level 1 — full route result (72-hour TTL, keyed by country+src+dst+vehicle_class hash).
Level 2 — per-plaza rate (72-hour TTL, keyed by plaza_id+vehicle_class).

OSM plaza reference table (refreshed every 7 days per country).
API usage log for TollGuru quota tracking.
"""

import duckdb
import json
import hashlib
import threading
from datetime import datetime, timedelta
from typing import Optional, List, Dict
import config

_lock = threading.Lock()


def _conn() -> duckdb.DuckDBPyConnection:
    return duckdb.connect(config.DB_PATH)


def init_db():
    with _lock:
        con = _conn()
        con.execute("""
            CREATE TABLE IF NOT EXISTS osm_plazas (
                plaza_id      VARCHAR PRIMARY KEY,
                plaza_name    VARCHAR,
                highway_number VARCHAR,
                latitude      DOUBLE,
                longitude     DOUBLE,
                country       VARCHAR,
                last_updated  TIMESTAMP
            )
        """)
        con.execute("""
            CREATE TABLE IF NOT EXISTS plaza_rate_cache (
                plaza_id       VARCHAR,
                vehicle_class  VARCHAR,
                rate           DOUBLE,
                fasttag_rate   DOUBLE,
                cash_rate      DOUBLE,
                currency       VARCHAR,
                fetched_at     TIMESTAMP,
                source         VARCHAR,
                country        VARCHAR,
                PRIMARY KEY (plaza_id, vehicle_class)
            )
        """)
        con.execute("""
            CREATE TABLE IF NOT EXISTS route_cache (
                route_key         VARCHAR PRIMARY KEY,
                country           VARCHAR,
                source_location   VARCHAR,
                destination_location VARCHAR,
                vehicle_class     VARCHAR,
                full_result_json  VARCHAR,
                cached_at         TIMESTAMP
            )
        """)
        con.execute("""
            CREATE TABLE IF NOT EXISTS api_usage_log (
                id           INTEGER,
                service      VARCHAR,
                event_type   VARCHAR,
                event_time   TIMESTAMP,
                details      VARCHAR
            )
        """)
        con.execute("""
            CREATE SEQUENCE IF NOT EXISTS api_log_seq START 1
        """)
        con.close()
    # Seed static plaza data so the app works even before Overpass responds
    try:
        from src.seed_data import ensure_seeds_in_db
        ensure_seeds_in_db("India")
        ensure_seeds_in_db("United States")
    except Exception:
        pass


# ── Route cache ──────────────────────────────────────────────────────────────

def make_route_key(country: str, source: str, destination: str, vehicle_class: str) -> str:
    from src.seed_data import SEED_DATA_VERSION
    raw = (
        f"{country.lower()}|{source.lower().strip()}|"
        f"{destination.lower().strip()}|{vehicle_class.lower()}|{SEED_DATA_VERSION}"
    )
    return hashlib.md5(raw.encode()).hexdigest()


def get_route_cache(route_key: str) -> Optional[Dict]:
    cutoff = datetime.utcnow() - timedelta(hours=config.ROUTE_CACHE_TTL_HOURS)
    con = _conn()
    row = con.execute(
        "SELECT full_result_json FROM route_cache WHERE route_key = ? AND cached_at > ?",
        [route_key, cutoff],
    ).fetchone()
    con.close()
    return json.loads(row[0]) if row else None


def set_route_cache(
    route_key: str,
    country: str,
    source: str,
    destination: str,
    vehicle_class: str,
    result: Dict,
):
    con = _conn()
    con.execute(
        """INSERT OR REPLACE INTO route_cache
           (route_key, country, source_location, destination_location,
            vehicle_class, full_result_json, cached_at)
           VALUES (?,?,?,?,?,?,?)""",
        [route_key, country, source, destination, vehicle_class,
         json.dumps(result, default=str), datetime.utcnow()],
    )
    con.close()


# ── Plaza rate cache (Level 2) ────────────────────────────────────────────────

def get_plaza_rate(plaza_id: str, vehicle_class: str) -> Optional[Dict]:
    cutoff = datetime.utcnow() - timedelta(hours=config.PLAZA_CACHE_TTL_HOURS)
    con = _conn()
    row = con.execute(
        """SELECT rate, fasttag_rate, cash_rate, currency, source
           FROM plaza_rate_cache
           WHERE plaza_id = ? AND vehicle_class = ? AND fetched_at > ?""",
        [plaza_id, vehicle_class, cutoff],
    ).fetchone()
    con.close()
    if row:
        return {"rate": row[0], "fasttag_rate": row[1], "cash_rate": row[2],
                "currency": row[3], "source": row[4]}
    return None


def set_plaza_rate(
    plaza_id: str,
    vehicle_class: str,
    rate: float,
    fasttag_rate: float,
    cash_rate: float,
    currency: str,
    source: str,
    country: str,
):
    con = _conn()
    con.execute(
        """INSERT OR REPLACE INTO plaza_rate_cache
           (plaza_id, vehicle_class, rate, fasttag_rate, cash_rate,
            currency, fetched_at, source, country)
           VALUES (?,?,?,?,?,?,?,?,?)""",
        [plaza_id, vehicle_class, rate, fasttag_rate, cash_rate,
         currency, datetime.utcnow(), source, country],
    )
    con.close()


# ── OSM plaza reference ───────────────────────────────────────────────────────

def get_osm_plazas(country: str) -> List[Dict]:
    con = _conn()
    rows = con.execute(
        """SELECT plaza_id, plaza_name, highway_number, latitude, longitude
           FROM osm_plazas WHERE country = ?""",
        [country],
    ).fetchall()
    con.close()
    return [
        {"plaza_id": r[0], "plaza_name": r[1], "highway_number": r[2],
         "latitude": r[3], "longitude": r[4]}
        for r in rows
    ]


def get_osm_last_updated(country: str) -> Optional[datetime]:
    con = _conn()
    row = con.execute(
        "SELECT MAX(last_updated) FROM osm_plazas WHERE country = ?", [country]
    ).fetchone()
    con.close()
    return row[0] if (row and row[0]) else None


def upsert_osm_plazas(plazas: List[Dict], country: str):
    con = _conn()
    con.execute("DELETE FROM osm_plazas WHERE country = ?", [country])
    now = datetime.utcnow()
    for p in plazas:
        con.execute(
            """INSERT INTO osm_plazas
               (plaza_id, plaza_name, highway_number, latitude, longitude, country, last_updated)
               VALUES (?,?,?,?,?,?,?)""",
            [p["plaza_id"], p["plaza_name"], p["highway_number"],
             p["latitude"], p["longitude"], country, now],
        )
    con.close()


# ── API usage log ─────────────────────────────────────────────────────────────

def log_api_event(service: str, event_type: str, details: str = ""):
    con = _conn()
    con.execute(
        """INSERT INTO api_usage_log (id, service, event_type, event_time, details)
           VALUES (nextval('api_log_seq'),?,?,?,?)""",
        [service, event_type, datetime.utcnow(), details],
    )
    con.close()


def is_tollguru_limited() -> bool:
    """True if TollGuru was rate-limited at any point this calendar month."""
    first_of_month = datetime.utcnow().replace(day=1, hour=0, minute=0, second=0, microsecond=0)
    con = _conn()
    row = con.execute(
        """SELECT COUNT(*) FROM api_usage_log
           WHERE service = 'tollguru' AND event_type = 'rate_limited' AND event_time >= ?""",
        [first_of_month],
    ).fetchone()
    con.close()
    return (row[0] > 0) if row else False