13faed0a06
Boards (signal-weighted, ~150/sweep): - Add TimesJobs (shahidirfan) — richest board: skills+salary+experience+offsite. Build/normalize + register; legacy city alias (New Delhi→Delhi, Bengaluru→Bangalore, Gurugram→Gurgaon) — verified live it returns 0 for "New Delhi", 35 for "Delhi". - Swap LinkedIn harvestapi → curious_coder (structured industry/jobFunction/applicants); harvestapi kept registered as linkedin_v1 fallback. - Drop Indeed from enabled (0% skills, redundant). Budgets: Foundit 40, TimesJobs 35, Naukri 30, LinkedIn 30, WorkIndia 15. Pool — cursorless freshness + safety: - REMOVE pagination/cursors: curious_coder only honors start=0 (start>0 → empty), proven; the rest are date feeds. Freshness now = boards' date-sort + the per-user seen-net + pool_save id-dedup. - Exhaustion guard: a refill adding < POOL_MIN_NEW_PER_REFILL new jobs flags the (user,query) exhausted → the gate relaxes the floor instead of block-fetching dupes (makes the 100/90 floor safe on niche queries). - Storage cap: pool_save trims beyond POOL_MAX_PER_QUERY freshest (kills DB swell). 72h TTL verified. - Thresholds 100/90 (deep pool, pay-for-volume). Scoring / cards: - Evidence-based fallback prose: when Haiku's cards stage gives nothing, the card uses Opus's REAL dimension notes (not a generic "Strong on X") + coverage logging + generic salvage parser. - "Skills & requirements" → "Skills fit" (consistent dimension labels). Bug fixes: - mark_seen dedups ids (was CardinalityViolationError on a duplicate id in one batch). Tests: deep stack contracts + committed board-sample fixture (no cache-pollution flakiness) + offline e2e + opt-in live e2e. 64 pass.
504 lines
23 KiB
Python
504 lines
23 KiB
Python
"""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, UserJobPool
|
|
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
|
|
# DEDUP the ids — a duplicate (user, oppid) in ONE insert trips PG's "ON CONFLICT cannot affect row
|
|
# a second time" (CardinalityViolationError). A deck can carry the same id twice (cross-board dupes).
|
|
seen_ids = list(dict.fromkeys(o for o in opportunity_ids if o))
|
|
rows = [{"user_id": user_id, "opportunity_id": oid, "seen": True} for oid in seen_ids]
|
|
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)
|
|
|
|
|
|
# ── Warm pool (the shelf) ────────────────────────────────────────────────────
|
|
async def pool_save(user_id: str, query_sig: str, jobs: list[dict]) -> int:
|
|
"""Bank fetched jobs into the per-(user, query) pool (upsert by job id; refresh pooled_at). Returns
|
|
the count of GENUINELY NEW job ids (not already pooled) — the exhaustion signal. Then TTL-sweeps stale
|
|
rows AND caps storage to POOL_MAX_PER_QUERY freshest (kills DB swell). 0 if the DB is off."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from app.config import get_settings
|
|
factory = session_factory()
|
|
if factory is None or not user_id or not jobs:
|
|
return 0
|
|
rows = [{"user_id": user_id, "query_sig": query_sig, "job_id": j["id"], "job": j}
|
|
for j in jobs if j.get("id")]
|
|
if not rows:
|
|
return 0
|
|
s = get_settings()
|
|
cutoff = datetime.now(timezone.utc) - timedelta(hours=s.POOL_TTL_HOURS)
|
|
incoming = {r["job_id"] for r in rows}
|
|
try:
|
|
async with factory() as sess:
|
|
# how many of these are NEW (not already pooled for this query)?
|
|
existing = set((await sess.execute(
|
|
select(UserJobPool.job_id).where(
|
|
(UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig)
|
|
& (UserJobPool.job_id.in_(incoming)))
|
|
)).scalars().all())
|
|
new_count = len(incoming - existing)
|
|
|
|
stmt = pg_insert(UserJobPool).values(rows)
|
|
stmt = stmt.on_conflict_do_update(
|
|
index_elements=[UserJobPool.user_id, UserJobPool.query_sig, UserJobPool.job_id],
|
|
set_={"job": stmt.excluded.job, "pooled_at": func.now()},
|
|
)
|
|
await sess.execute(stmt)
|
|
# TTL sweep (stale postings, likely filled).
|
|
await sess.execute(UserJobPool.__table__.delete().where(
|
|
(UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig)
|
|
& (UserJobPool.pooled_at < cutoff)))
|
|
# STORAGE cap: keep only the POOL_MAX_PER_QUERY freshest for this (user, query).
|
|
keep = (await sess.execute(
|
|
select(UserJobPool.job_id)
|
|
.where((UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig))
|
|
.order_by(UserJobPool.pooled_at.desc()).limit(s.POOL_MAX_PER_QUERY)
|
|
)).scalars().all()
|
|
if len(keep) >= s.POOL_MAX_PER_QUERY:
|
|
await sess.execute(UserJobPool.__table__.delete().where(
|
|
(UserJobPool.user_id == user_id) & (UserJobPool.query_sig == query_sig)
|
|
& (UserJobPool.job_id.notin_(keep))))
|
|
await sess.commit()
|
|
return new_count
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("pool_save failed (non-fatal): %s", e)
|
|
return 0
|
|
|
|
|
|
async def mark_pool_exhausted(user_id: str, query_sig: str) -> None:
|
|
"""Flag this (user, query) as tapped out (a refill added almost no new jobs) → the gate relaxes the
|
|
floor until POOL_EXHAUSTION_TTL passes. Upsert on the one-row-per-user feed."""
|
|
factory = session_factory()
|
|
if factory is None or not user_id:
|
|
return
|
|
try:
|
|
async with factory() as s:
|
|
stmt = pg_insert(UserFeed).values(user_id=user_id, query_sig=query_sig, exhausted_at=func.now())
|
|
stmt = stmt.on_conflict_do_update(
|
|
index_elements=[UserFeed.user_id],
|
|
set_={"exhausted_at": func.now(), "query_sig": query_sig},
|
|
)
|
|
await s.execute(stmt)
|
|
await s.commit()
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("mark_pool_exhausted failed (non-fatal): %s", e)
|
|
|
|
|
|
async def is_pool_exhausted(user_id: str, query_sig: str) -> bool:
|
|
"""True if this exact query was marked exhausted within POOL_EXHAUSTION_TTL (boards refresh ~daily,
|
|
so we re-try after it expires). False on any other query / DB off / expired."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from app.config import get_settings
|
|
factory = session_factory()
|
|
if factory is None or not user_id:
|
|
return False
|
|
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("is_pool_exhausted failed (non-fatal): %s", e)
|
|
return False
|
|
if not row or row.query_sig != query_sig or not row.exhausted_at:
|
|
return False
|
|
ttl = get_settings().POOL_EXHAUSTION_TTL_HOURS
|
|
return (datetime.now(timezone.utc) - row.exhausted_at).total_seconds() < ttl * 3600
|
|
|
|
|
|
async def pool_load_fresh_unseen(user_id: str, query_sig: str, seen: set[str]) -> list[dict]:
|
|
"""The banked jobs for this query that are still FRESH (within TTL) and UNSEEN (not already shown),
|
|
freshest first. This is the count the reorder/floor gate reads. Empty list if the DB is off."""
|
|
from datetime import datetime, timedelta, timezone
|
|
from app.config import get_settings
|
|
factory = session_factory()
|
|
if factory is None or not user_id:
|
|
return []
|
|
s = get_settings()
|
|
cutoff = datetime.now(timezone.utc) - timedelta(hours=s.POOL_TTL_HOURS)
|
|
try:
|
|
async with factory() as sess:
|
|
rows = (await sess.execute(
|
|
select(UserJobPool.job_id, UserJobPool.job)
|
|
.where((UserJobPool.user_id == user_id)
|
|
& (UserJobPool.query_sig == query_sig)
|
|
& (UserJobPool.pooled_at >= cutoff))
|
|
.order_by(UserJobPool.pooled_at.desc())
|
|
.limit(s.POOL_MAX_PER_QUERY)
|
|
)).all()
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("pool_load failed (non-fatal): %s", e)
|
|
return []
|
|
return [r.job for r in rows if r.job_id not in seen]
|
|
|
|
|
|
async def save_cursors(user_id: str, query_sig: str, cursors: dict) -> None:
|
|
"""Persist advanced per-board cursors WITHOUT touching the saved feed — used by the background
|
|
refill (which sweeps + banks but has no new deck to show). Upsert on the one-row-per-user feed."""
|
|
factory = session_factory()
|
|
if factory is None or not user_id:
|
|
return
|
|
try:
|
|
async with factory() as s:
|
|
stmt = pg_insert(UserFeed).values(user_id=user_id, query_sig=query_sig, cursors=cursors or {})
|
|
stmt = stmt.on_conflict_do_update(
|
|
index_elements=[UserFeed.user_id],
|
|
set_={"cursors": cursors or {}, "query_sig": query_sig},
|
|
)
|
|
await s.execute(stmt)
|
|
await s.commit()
|
|
except Exception as e: # noqa: BLE001
|
|
logger.warning("save_cursors failed (non-fatal): %s", e)
|