improve signed reads, crypto hardening, and scraper transport

app/scrapers/truth_social.py

@@ -11,7 +11,6 @@ import re
 from datetime import datetime, timezone
 from typing import Optional
 
-import httpx
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -29,6 +28,16 @@ ARCHIVE_URL = "https://ix.cnn.io/data/truth-social/truth_archive.json"
 last_successful_poll_at: Optional[datetime] = None
 last_poll_error: Optional[str] = None
 
+# Conditional-GET validators from the last 200 response. The archive is 30k+
+# posts (~MBs of JSON); CNN only regenerates it every ~5 min, so most polls
+# can be answered with a 304 and skip download + parse entirely.
+_etag: Optional[str] = None
+_last_modified: Optional[str] = None
+
+# Sentinel returned by _fetch_archive on HTTP 304 — distinct from None
+# (fetch failure) so the poller can count it as a successful, no-work cycle.
+NOT_MODIFIED = object()
+
 
 def _strip_html(text: str) -> str:
     text = re.sub(r"<[^>]+>", " ", text)
@@ -52,16 +61,29 @@ def _parse_dt(iso: str) -> datetime:
         return datetime.now(timezone.utc).replace(tzinfo=None)
 
 
-async def _fetch_archive() -> Optional[list]:
+async def _fetch_archive(conditional: bool = True):
+    """Fetch the archive JSON. Returns a list, NOT_MODIFIED (304), or None
+    on error. `conditional=False` (backfill) always downloads the full body
+    so a poller-set ETag can't starve the startup backfill."""
+    global _etag, _last_modified
     headers = {
         "User-Agent": "Mozilla/5.0 (compatible; TrumpSignal/1.0)",
         "Accept": "application/json",
     }
+    if conditional:
+        if _etag:
+            headers["If-None-Match"] = _etag
+        if _last_modified:
+            headers["If-Modified-Since"] = _last_modified
     try:
-        async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
-            resp = await client.get(ARCHIVE_URL, headers=headers)
-            resp.raise_for_status()
-            return resp.json()
+        from app.services.http_client import get_client
+        resp = await get_client().get(ARCHIVE_URL, headers=headers, timeout=30)
+        if conditional and resp.status_code == 304:
+            return NOT_MODIFIED
+        resp.raise_for_status()
+        _etag = resp.headers.get("etag")
+        _last_modified = resp.headers.get("last-modified")
+        return resp.json()
     except Exception as exc:
         # Include type name — httpx often raises bare ConnectError/TimeoutException
         # with empty .args, which used to log as just "Failed to fetch CNN archive:"
@@ -71,9 +93,27 @@ async def _fetch_archive() -> Optional[list]:
         return None
 
 
-async def _process_entry(entry: dict, db: AsyncSession) -> Optional[Post]:
+async def _known_external_ids(entries: list, db: AsyncSession) -> set:
+    """One batch query for the dedup pass. On a typical poll all ~50 entries
+    already exist, so this replaces 50 per-entry SELECTs with 1."""
+    ids = [hashlib.md5(str(e["id"]).encode()).hexdigest() for e in entries]
+    if not ids:
+        return set()
+    rows = await db.execute(
+        select(Post.external_id).where(Post.external_id.in_(ids)))
+    return {r[0] for r in rows}
+
+
+async def _process_entry(entry: dict, db: AsyncSession,
+                         known_ids: Optional[set] = None) -> Optional[Post]:
     external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()
 
+    # Fast path: pre-fetched batch dedup set. New (unseen) ids still get the
+    # confirming SELECT below — protects against a concurrent insert by the
+    # other poller between the batch query and this entry.
+    if known_ids is not None and external_id in known_ids:
+        return None
+
     result = await db.execute(select(Post).where(Post.external_id == external_id))
     if result.scalar_one_or_none():
         return None
@@ -198,10 +238,40 @@ def _post_to_ws_payload(post: Post) -> dict:
     }
 
 
+async def dispatch_post(post: Post, db: AsyncSession) -> None:
+    """Broadcast + fan-out + trade for one freshly committed post. Shared by
+    both pollers so delivery doesn't depend on which source wins the race."""
+    await manager.broadcast(_post_to_ws_payload(post))
+    logger.info("Saved new post id=%d: %s", post.id, post.text[:60])
+    # Telegram fan-out (fire-and-forget). _dispatch filters internally:
+    # buy/short → per-subscriber + public channel; relevant-but-hold →
+    # public channel only; noise → dropped.
+    try:
+        from app.services.telegram import notify_signal
+        notify_signal(post)
+    except Exception as exc:
+        logger.warning("Telegram notify failed for post %d: %s", post.id, exc)
+    try:
+        from app.services.x_poster import notify_x_signal
+        notify_x_signal(post)
+    except Exception as exc:
+        logger.warning("X notify failed for post %d: %s", post.id, exc)
+    try:
+        from app.services.bot_engine import process_post
+        await process_post(post, db)
+    except Exception as exc:
+        logger.error("process_post failed for post %d: %s", post.id, exc)
+
+
 async def poll_truth_social(db_session_factory) -> None:
     global last_successful_poll_at, last_poll_error
     logger.info("Polling CNN Truth Social archive...")
     entries = await _fetch_archive()
+    if entries is NOT_MODIFIED:
+        # Archive unchanged since last poll — successful cycle, nothing to do.
+        last_successful_poll_at = datetime.now(timezone.utc)
+        last_poll_error = None
+        return
     if not entries:
         last_poll_error = "fetch_archive returned empty"
         return
@@ -212,39 +282,24 @@ async def poll_truth_social(db_session_factory) -> None:
 
     async with db_session_factory() as db:
         try:
-            new_posts = []
+            known_ids = await _known_external_ids(recent, db)
+            found_new = False
+            # Entries are newest-first. Commit + dispatch each new post
+            # IMMEDIATELY rather than after the whole batch — an actionable
+            # post must not wait behind the AI analysis of older entries.
             for entry in recent:
                 try:
-                    post = await _process_entry(entry, db)
+                    post = await _process_entry(entry, db, known_ids)
                     if post:
-                        new_posts.append(post)
+                        found_new = True
+                        await db.commit()
+                        await dispatch_post(post, db)
                 except Exception as exc:
                     logger.error("Error processing entry %s: %s", entry.get("id"), exc)
 
-            if new_posts:
-                await db.commit()
-                for post in new_posts:
-                    await manager.broadcast(_post_to_ws_payload(post))
-                    logger.info("Saved new post id=%d: %s", post.id, post.text[:60])
-                    # Telegram fan-out (fire-and-forget). _dispatch filters
-                    # internally: buy/short → per-subscriber + public channel;
-                    # relevant-but-hold → public channel only; noise → dropped.
-                    try:
-                        from app.services.telegram import notify_signal
-                        notify_signal(post)
-                    except Exception as exc:
-                        logger.warning("Telegram notify failed for post %d: %s", post.id, exc)
-                    try:
-                        from app.services.x_poster import notify_x_signal
-                        notify_x_signal(post)
-                    except Exception as exc:
-                        logger.warning("X notify failed for post %d: %s", post.id, exc)
-                    try:
-                        from app.services.bot_engine import process_post
-                        await process_post(post, db)
-                    except Exception as exc:
-                        logger.error("process_post failed for post %d: %s", post.id, exc)
-            else:
+            # Capture any remaining writes (entry-filter stub rows).
+            await db.commit()
+            if not found_new:
                 logger.info("No new posts found.")
             # Mark a successful poll cycle (separate from "found new posts").
             last_successful_poll_at = datetime.now(timezone.utc)
@@ -258,7 +313,7 @@ async def poll_truth_social(db_session_factory) -> None:
 async def backfill_history(db_session_factory, limit: int = 500) -> None:
     """One-time backfill of historical posts (no Claude analysis, no price impact)."""
     logger.info("Starting historical backfill (limit=%d)...", limit)
-    entries = await _fetch_archive()
+    entries = await _fetch_archive(conditional=False)
     if not entries:
         logger.error("Backfill failed: could not fetch archive")
         return
@@ -268,10 +323,10 @@ async def backfill_history(db_session_factory, limit: int = 500) -> None:
 
     async with db_session_factory() as db:
         try:
+            known_ids = await _known_external_ids(to_process, db)
             for entry in to_process:
                 external_id = hashlib.md5(str(entry["id"]).encode()).hexdigest()
-                result = await db.execute(select(Post).where(Post.external_id == external_id))
-                if result.scalar_one_or_none():
+                if external_id in known_ids:
                     continue
 
                 text = _strip_html(entry.get("content") or "").strip()