Wire ingestion: long-running ingester (sources->NATS->Postgres), fix nats-py API (durable=/LIMITS), drop UUID source_id on RSS

app/ingestor.py

@@ -66,7 +66,7 @@ async def start_nats_consumer():
                 "events.earthquake", "events.disaster", "events.weather",
                 "events.fire", "events.satellite", "events.new", "events.alert",
             ],
-            retention=nats.js.api.RetentionPolicy.INTERESTS,
+            retention=nats.js.api.RetentionPolicy.LIMITS,
             max_msgs=1_000_000,
         )
         logger.info("Created NATS stream %s", NATS_STREAM)
@@ -76,7 +76,7 @@ async def start_nats_consumer():
     # Create durable consumer
     sub = await js.pull_subscribe(
         subject="events.>",
-        durable_name=NATS_DURABLE,
+        durable=NATS_DURABLE,
     )
 
     logger.info("NATS consumer started, durable=%s", NATS_DURABLE)

app/run_ingester.py

@@ -45,7 +45,7 @@ async def producer_loop() -> None:
         try:
             for url in RSS_URLS:
                 try:
-                    n = await ingest_rss_feed(url, source_id=url)
+                    n = await ingest_rss_feed(url)
                     logger.info("RSS %s -> %d items", url, n)
                 except Exception:  # noqa: BLE001
                     logger.exception("RSS fetch failed: %s", url)

app/sources.py

@@ -40,7 +40,7 @@ async def publish_event(subject: str, event: dict):
 
 # ─── RSS Feed Ingestor ──────────────────────────────────────────────────
 
-async def ingest_rss_feed(feed_url: str, source_id: str = None):
+async def ingest_rss_feed(feed_url: str):
     """Fetch and parse an RSS feed, publish items to NATS."""
     async with httpx.AsyncClient(timeout=30) as client:
         resp = await client.get(feed_url)
@@ -51,7 +51,6 @@ async def ingest_rss_feed(feed_url: str, source_id: str = None):
     for entry in feed.entries[:100]:  # max 100 per run
         event = {
             "source_type": "rss",
-            "source_id": source_id,
             "title": entry.get("title"),
             "body": entry.get("summary") or entry.get("description"),
             "url": entry.get("link"),
@@ -59,6 +58,7 @@ async def ingest_rss_feed(feed_url: str, source_id: str = None):
                 or datetime.now(timezone.utc).isoformat(),
             "tags": [t.get("term") for t in entry.get("tags", []) if t.get("term")],
             "raw": {
+                "feed_url": feed_url,
                 "feed_title": feed.feed.get("title"),
                 "author": entry.get("author"),
                 "categories": [c.get("term") for c in entry.get("categories", [])],

docker-compose.yml

@@ -70,6 +70,7 @@ services:
       INGEST_INTERVAL: ${INGEST_INTERVAL:-300}
       INGEST_EARTHQUAKES: ${INGEST_EARTHQUAKES:-1}
     command: ["python", "app/run_ingester.py"]
+    entrypoint: ["python", "app/run_ingester.py"]
 
   app:
     build: