matchmaking-v2/app/db/repo.py

"""Feed repository — upsert + read the per-user cached feed. All calls degrade to no-op/None
when the DB is off or unreachable (persistence is optional; the engine must never crash on it)."""
from __future__ import annotations

import hashlib
import json
import logging

from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert as pg_insert

from app.db.models import Base, OpportunityState, UserFeed
from app.db.session import engine, session_factory

logger = logging.getLogger(__name__)

# fields a caller may set on an opportunity's state (status + the generated document ids + activity flags)
_STATE_FIELDS = ("status", "tailored_resume_id", "tailored_version_id", "cover_letter_id", "seen", "viewed", "saved")

# the search-defining pref fields — same values → same signature → cursor advances (new jobs);
# change any → new signature → cursor resets to page 1.
_SIG_FIELDS = ("title", "role", "location", "industry", "experience", "years", "work_mode", "remote")


def search_signature(prefs: dict) -> str:
    """Stable short hash of the search-defining prefs (ignores transient flags / user_context)."""
    payload = {k: prefs.get(k) for k in _SIG_FIELDS}
    blob = json.dumps(payload, sort_keys=True, default=str)
    return hashlib.sha1(blob.encode()).hexdigest()[:16]


async def init_db() -> None:
    """Create tables if the DB is configured (best-effort; logged, never fatal)."""
    eng = engine()
    if eng is None:
        logger.info("DATABASE_URL unset — feed persistence disabled")
        return
    try:
        async with eng.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        logger.info("feed store ready")
    except Exception as e:  # noqa: BLE001
        logger.warning("feed store init failed (persistence off): %s", e)


async def save_feed(user_id: str, prefs: dict, result: dict, *, query_sig: str = "", cursors: dict | None = None) -> None:
    """Upsert the user's last completed search + the advanced per-board cursors. No-op if the DB is off."""
    factory = session_factory()
    if factory is None or not user_id:
        return
    row = {
        "user_id": user_id,
        "prefs": prefs or {},
        "opportunities": result.get("opportunities") or [],
        "sources": result.get("sources") or {},
        "engine": result.get("engine") or "",
        "scanned": int(result.get("scanned") or 0),
        "query_sig": query_sig,
        "cursors": cursors or {},
        "search_count": 1,
    }
    try:
        async with factory() as s:
            stmt = pg_insert(UserFeed).values(**row)
            stmt = stmt.on_conflict_do_update(
                index_elements=[UserFeed.user_id],
                set_={**{k: row[k] for k in ("prefs", "opportunities", "sources", "engine", "scanned", "query_sig", "cursors")},
                      # engagement: count every search run (insert starts at 1, each re-run +1)
                      "search_count": UserFeed.search_count + 1},
            )
            await s.execute(stmt)
            await s.commit()
    except Exception as e:  # noqa: BLE001
        logger.warning("save_feed failed (non-fatal): %s", e)


async def update_salary_band(user_id: str, salaries: list[float]) -> None:
    """Accumulate the salary band: each deck contributes its HIGHEST disclosed salary (a 'peak'); the
    band is the running AVERAGE of those peaks (₹L). No-op when the deck disclosed nothing (persists)."""
    sal = [float(x) for x in salaries if x]
    factory = session_factory()
    if factory is None or not user_id or not sal:
        return
    hi = max(sal)
    try:
        async with factory() as s:
            row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none()
            if row is None:
                return
            row.salary_peak_sum = (row.salary_peak_sum or 0.0) + hi   # this deck's peak
            row.salary_peak_count = (row.salary_peak_count or 0) + 1
            await s.commit()
    except Exception as e:  # noqa: BLE001
        logger.warning("update_salary_band failed (non-fatal): %s", e)


async def get_salary_band(user_id: str) -> float | None:
    """The salary band = average of each deck's peak salary (₹L), or None if no deck disclosed one."""
    factory = session_factory()
    if factory is None or not user_id:
        return None
    try:
        async with factory() as s:
            row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none()
    except Exception as e:  # noqa: BLE001
        logger.warning("get_salary_band failed (non-fatal): %s", e)
        return None
    if not row or not row.salary_peak_count:
        return None
    return round((row.salary_peak_sum or 0.0) / row.salary_peak_count, 1)


async def get_feed(user_id: str) -> dict | None:
    """The user's last cached feed as a search_complete-shaped dict, or None."""
    factory = session_factory()
    if factory is None or not user_id:
        return None
    try:
        async with factory() as s:
            row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none()
    except Exception as e:  # noqa: BLE001
        logger.warning("get_feed failed (non-fatal): %s", e)
        return None
    if not row:
        return None
    states = await get_opportunity_states(user_id)
    opps = annotate_and_filter(row.opportunities or [], states)
    return {
        "opportunities": opps,
        "sources": row.sources or {},
        "prefs": row.prefs or {},
        "engine": row.engine,
        "scanned": row.scanned,
        "shortlisted": len(opps),
        "cached": True,
        "updated_at": row.updated_at.isoformat() if row.updated_at else None,
    }


async def set_opportunity_state(user_id: str, opportunity_id: str, **fields) -> None:
    """Upsert per-opportunity state (status and/or document ids). No-op if the DB is off.
    Only the provided fields are written, so partial updates don't clobber existing values."""
    factory = session_factory()
    if factory is None or not user_id or not opportunity_id:
        return
    vals = {k: fields[k] for k in _STATE_FIELDS if k in fields}
    if not vals:
        return
    row = {"user_id": user_id, "opportunity_id": opportunity_id, **vals}
    try:
        async with factory() as s:
            stmt = pg_insert(OpportunityState).values(**row)
            stmt = stmt.on_conflict_do_update(
                index_elements=[OpportunityState.user_id, OpportunityState.opportunity_id],
                set_=vals,
            )
            await s.execute(stmt)
            await s.commit()
    except Exception as e:  # noqa: BLE001
        logger.warning("set_opportunity_state failed (non-fatal): %s", e)


async def get_opportunity_states(user_id: str) -> dict[str, dict]:
    """{opportunity_id: {status, tailored_resume_id, tailored_version_id, cover_letter_id}}."""
    factory = session_factory()
    if factory is None or not user_id:
        return {}
    try:
        async with factory() as s:
            rows = (await s.execute(
                select(OpportunityState).where(OpportunityState.user_id == user_id)
            )).scalars().all()
    except Exception as e:  # noqa: BLE001
        logger.warning("get_opportunity_states failed (non-fatal): %s", e)
        return {}
    return {
        r.opportunity_id: {
            "status": r.status,
            "tailored_resume_id": r.tailored_resume_id,
            "tailored_version_id": r.tailored_version_id,
            "cover_letter_id": r.cover_letter_id,
        }
        for r in rows
    }


# engagement score weights — applies count most, then saves, views, searches (honest "working it")
def _activity_score(viewed: int, saved: int, applied: int, searches: int) -> int:
    return applied * 3 + saved * 2 + viewed * 1 + searches * 1


async def get_activity_stats(user_id: str) -> dict:
    """Real dashboard activity for `user_id`: the funnel counts (viewed/saved/applied), how many
    documents Mira tailored, search count, and the COHORT engagement percentile (this user's activity
    vs every Scout user in the matchmaking DB). All honest, DB-derived. Degrades to zeros if DB is off."""
    factory = session_factory()
    empty = {"matches": 0, "viewed": 0, "saved": 0, "applied": 0, "tailored": 0, "searches": 0, "percentile": 0, "cohort_size": 0}
    if factory is None or not user_id:
        return empty
    try:
        async with factory() as s:
            # per-user activity across ALL users (one grouped pass → funnel for this user + cohort rank)
            os_rows = (await s.execute(
                select(
                    OpportunityState.user_id,
                    func.count().filter(OpportunityState.seen.is_(True)).label("matches"),
                    func.count().filter(OpportunityState.viewed.is_(True)).label("viewed"),
                    func.count().filter(OpportunityState.saved.is_(True)).label("saved"),
                    func.count().filter(OpportunityState.status == "applied").label("applied"),
                    func.count().filter(
                        OpportunityState.tailored_version_id.isnot(None)
                        | OpportunityState.cover_letter_id.isnot(None)
                    ).label("tailored"),
                ).group_by(OpportunityState.user_id)
            )).all()
            feed_rows = (await s.execute(select(UserFeed.user_id, UserFeed.search_count))).all()
    except Exception as e:  # noqa: BLE001
        logger.warning("get_activity_stats failed (non-fatal): %s", e)
        return empty

    searches_by = {r.user_id: int(r.search_count or 0) for r in feed_rows}
    me, scores = dict(empty), []
    users = {r.user_id for r in os_rows} | set(searches_by)
    by_os = {r.user_id: r for r in os_rows}
    for uid in users:
        r = by_os.get(uid)
        mt, v, sv, ap, tl = (int(r.matches), int(r.viewed), int(r.saved), int(r.applied), int(r.tailored)) if r else (0, 0, 0, 0, 0)
        sc = searches_by.get(uid, 0)
        scores.append(_activity_score(v, sv, ap, sc))
        if uid == user_id:
            me = {"matches": mt, "viewed": v, "saved": sv, "applied": ap, "tailored": tl, "searches": sc}
    my_score = _activity_score(me["viewed"], me["saved"], me["applied"], me["searches"])
    cohort = len(scores) or 1
    # percentile = share of the cohort you're STRICTLY ahead of (so rank-1-of-3 → 67th → "top 33%",
    # not "top 0%"). UI shows top (100 - percentile)%.
    below = sum(1 for x in scores if x < my_score)
    me["percentile"] = round(100 * below / cohort)
    me["cohort_size"] = cohort
    return me


async def get_active_window(user_id: str, *, tz_offset_min: int = 330) -> dict:
    """WHEN the user works their search — activity timestamps (views/saves/applies + searches)
    bucketed by local hour-of-day. Returns the peak 3-hour band + a 24-bucket histogram. Real signal,
    from this DB. tz_offset_min defaults to IST (+5:30); the product is India-centric.

    SQLite/Postgres-agnostic: we read raw UTC timestamps and bucket in Python (so it works in tests too)."""
    factory = session_factory()
    empty = {"by_hour": [0] * 24, "peak_start": None, "peak_end": None, "samples": 0}
    if factory is None or not user_id:
        return empty
    try:
        async with factory() as s:
            os_ts = (await s.execute(
                select(OpportunityState.updated_at).where(OpportunityState.user_id == user_id)
            )).scalars().all()
            feed_ts = (await s.execute(
                select(UserFeed.updated_at).where(UserFeed.user_id == user_id)
            )).scalars().all()
    except Exception as e:  # noqa: BLE001
        logger.warning("get_active_window failed (non-fatal): %s", e)
        return empty

    by_hour = [0] * 24
    for ts in [*os_ts, *feed_ts]:
        if ts is None:
            continue
        local_min = ts.hour * 60 + ts.minute + tz_offset_min
        by_hour[(local_min // 60) % 24] += 1
    samples = sum(by_hour)
    if not samples:
        return empty
    # peak 3-hour window (circular) by summed activity
    best_h, best_sum = 0, -1
    for h in range(24):
        win = by_hour[h] + by_hour[(h + 1) % 24] + by_hour[(h + 2) % 24]
        if win > best_sum:
            best_h, best_sum = h, win
    return {"by_hour": by_hour, "peak_start": best_h, "peak_end": (best_h + 3) % 24, "samples": samples}


def annotate_and_filter(opportunities: list[dict], states: dict[str, dict]) -> list[dict]:
    """Drop dismissed opportunities and annotate the rest with their persisted state (applied flag
    + already-generated document ids) so the UI can restore without re-tailoring."""
    out: list[dict] = []
    for o in opportunities:
        st = states.get(o.get("id"))
        if st and st.get("status") == "dismissed":
            continue                                   # never resurface
        if st:
            o = {**o,
                 "applied": st.get("status") == "applied",
                 "tailored_resume_id": st.get("tailored_resume_id"),
                 "tailored_version_id": st.get("tailored_version_id"),
                 "cover_letter_id": st.get("cover_letter_id")}
        out.append(o)
    return out


_CURSOR_TTL_HOURS = 24   # job boards refresh ~daily (live scrape, 7-day newest-first window) → reset every 24h


async def get_search_cursors(user_id: str, query_sig: str) -> dict:
    """Per-board start pages for this query, e.g. {'linkedin': 3}. RESETS to empty (every board → page 1)
    when EITHER the query changed OR the last search was >24h ago (so we re-surface the freshest daily
    postings instead of marching into stale depth). Each board advances its OWN page; dedup is the
    per-user seen-net, never cross-user."""
    from datetime import datetime, timezone
    factory = session_factory()
    if factory is None or not user_id:
        return {}
    try:
        async with factory() as s:
            row = (await s.execute(select(UserFeed).where(UserFeed.user_id == user_id))).scalar_one_or_none()
    except Exception as e:  # noqa: BLE001
        logger.warning("get_search_cursors failed (non-fatal): %s", e)
        return {}
    if not row or row.query_sig != query_sig:
        return {}                                 # new/changed query → all boards start at page 1
    if row.updated_at:
        age_h = (datetime.now(timezone.utc) - row.updated_at).total_seconds() / 3600
        if age_h >= _CURSOR_TTL_HOURS:
            return {}                             # >24h stale → reset to catch the newest jobs
    return dict(row.cursors or {})


async def get_seen_ids(user_id: str) -> set[str]:
    """Every opportunity the user has already been shown / acted on — excluded from new runs."""
    factory = session_factory()
    if factory is None or not user_id:
        return set()
    try:
        async with factory() as s:
            rows = (await s.execute(
                select(OpportunityState.opportunity_id).where(OpportunityState.user_id == user_id)
            )).scalars().all()
        return set(rows)
    except Exception as e:  # noqa: BLE001
        logger.warning("get_seen_ids failed (non-fatal): %s", e)
        return set()


async def mark_seen(user_id: str, opportunity_ids: list[str]) -> None:
    """Record that these opportunities were shown in a deck (so future runs skip them). No-op on DB off.
    Sets seen=true without touching any existing status/document ids."""
    factory = session_factory()
    if factory is None or not user_id or not opportunity_ids:
        return
    rows = [{"user_id": user_id, "opportunity_id": oid, "seen": True} for oid in opportunity_ids if oid]
    if not rows:
        return
    try:
        async with factory() as s:
            stmt = pg_insert(OpportunityState).values(rows)
            stmt = stmt.on_conflict_do_update(
                index_elements=[OpportunityState.user_id, OpportunityState.opportunity_id],
                set_={"seen": True},
            )
            await s.execute(stmt)
            await s.commit()
    except Exception as e:  # noqa: BLE001
        logger.warning("mark_seen failed (non-fatal): %s", e)