purge: DuckDB uses singular HOUR

app/mapper.py

@@ -1,5 +1,5 @@
 import os, json, duckdb, pandas as pd
-from datetime import datetime
+from datetime import datetime, timedelta
 from app.db import get_conn, ensure_raw_table
 from app.utils.detect_industry import _ALIAS
 
@@ -113,14 +113,16 @@ def canonify_df(org_id: str, hours_window: int = 24) -> pd.DataFrame:
     # ⏱  Safe timestamp filtering
     # --------------------------
     try:
+        # Compute cutoff in Python to avoid parameter placeholders inside INTERVAL
+        cutoff = datetime.now() - timedelta(hours=hours_window)
+        cutoff_str = cutoff.strftime("%Y-%m-%d %H:%M:%S")
         rows = conn.execute(
-            """
+            f"""
             SELECT row_data
             FROM raw_rows
             WHERE strptime(json_extract(row_data, '$.timestamp'), '%Y-%m-%d %H:%M:%S')
-                  >= now() - INTERVAL ? HOUR
-            """,
-            (hours_window,)
+                  >= TIMESTAMP '{cutoff_str}'
+            """
         ).fetchall()
     except Exception as e:
         print(f"[canonify] ⚠️ fallback to all rows due to timestamp parse error: {e}")

app/tasks/purge.py

@@ -1,5 +1,9 @@
 from app.db import get_conn, ensure_raw_table
+from datetime import datetime, timedelta
+
 def purge_old_raw(org_id: str, hours=6):
     conn = get_conn(org_id)
-    conn.execute("DELETE FROM raw_rows WHERE ingested_at < now() - INTERVAL ? HOURS", [hours])
+    cutoff = datetime.now() - timedelta(hours=hours)
+    cutoff_str = cutoff.strftime("%Y-%m-%d %H:%M:%S")
+    conn.execute(f"DELETE FROM raw_rows WHERE ingested_at < TIMESTAMP '{cutoff_str}'")
     conn.commit(); conn.close()