fix: persist Chatwoot outbound message timeline

app/persistence.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timezone
 from typing import Any, Dict, Optional, Tuple
 import json
 
@@ -32,6 +33,95 @@ def _lock_source_identity(conn: Any, source_system: str, source_event_id: str |
         })
 
 
+def _message_created_at(value: Any) -> datetime | None:
+    """Normalize a source message timestamp without substituting processing time."""
+    if value in (None, ""):
+        return None
+    if isinstance(value, datetime):
+        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
+    if isinstance(value, (int, float)) or str(value).strip().replace(".", "", 1).isdigit():
+        try:
+            return datetime.fromtimestamp(float(value), tz=timezone.utc)
+        except (OverflowError, TypeError, ValueError):
+            return None
+    try:
+        parsed = datetime.fromisoformat(str(value).strip().replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)
+
+
+def save_factual_chatwoot_message(
+    *,
+    raw_event_id: str,
+    source_event_id: str | None,
+    conversation_id: str | None,
+    contact_id: str | None,
+    direction: str,
+    raw_body: str,
+    clean_body: str | None = None,
+    source_created_at: Any = None,
+    metadata: Dict[str, Any] | None = None,
+) -> tuple[str, bool]:
+    """Persist one factual public Chatwoot message, without workflow effects.
+
+    Returns ``(message_id, inserted)``. Canonical source identity and the
+    transaction advisory lock make webhook delivery and backfill idempotent.
+    """
+    if not settings.clientflow_persist:
+        return "", False
+    normalized_direction = str(direction or "").strip().lower()
+    if normalized_direction not in {"inbound", "outbound"}:
+        raise ValueError("direction must be inbound or outbound")
+    created_at = _message_created_at(source_created_at)
+    with engine.begin() as conn:
+        _lock_source_identity(conn, "chatwoot", source_event_id)
+        existing = None
+        if source_event_id:
+            existing = conn.execute(text("""
+                SELECT id::text FROM messages
+                WHERE source_system='chatwoot' AND source_event_id=:source_event_id
+                LIMIT 1
+            """), {"source_event_id": source_event_id}).first()
+        if not existing:
+            existing = conn.execute(text("""
+                SELECT id::text FROM messages
+                WHERE raw_event_id=CAST(:raw_event_id AS UUID) LIMIT 1
+            """), {"raw_event_id": raw_event_id}).first()
+        inserted = existing is None
+        if inserted:
+            row = conn.execute(text("""
+                INSERT INTO messages (
+                    raw_event_id, source_system, source_event_id, conversation_id,
+                    contact_id, direction, raw_body, clean_body, previous_context,
+                    metadata, created_at
+                ) VALUES (
+                    CAST(:raw_event_id AS UUID), 'chatwoot', :source_event_id,
+                    :conversation_id, :contact_id, :direction, :raw_body, :clean_body,
+                    NULL, CAST(:metadata AS JSONB),
+                    COALESCE(CAST(:created_at AS TIMESTAMPTZ), now())
+                ) RETURNING id::text
+            """), {
+                "raw_event_id": raw_event_id,
+                "source_event_id": source_event_id,
+                "conversation_id": conversation_id,
+                "contact_id": contact_id,
+                "direction": normalized_direction,
+                "raw_body": raw_body,
+                "clean_body": clean_body if clean_body is not None else raw_body,
+                "metadata": _json(metadata),
+                "created_at": created_at,
+            }).first()
+            message_id = str(row[0])
+        else:
+            message_id = str(existing[0])
+        conn.execute(text("""
+            UPDATE raw_events SET message_id=CAST(:message_id AS UUID)
+            WHERE id=CAST(:raw_event_id AS UUID) AND message_id IS NULL
+        """), {"message_id": message_id, "raw_event_id": raw_event_id})
+    return message_id, inserted
+
+
 def save_inbound_message(
     *,
     request: AnalyzeRequest,