Continue AI features and improve protocol support

core/messaging/history.py

@@ -15,7 +15,11 @@ log = logs.get_logger("history")
 DEFAULT_PROMPT_HISTORY_MAX_MESSAGES = getattr(
     settings, "PROMPT_HISTORY_MAX_MESSAGES", 120
 )
-DEFAULT_PROMPT_HISTORY_MAX_CHARS = getattr(settings, "PROMPT_HISTORY_MAX_CHARS", 24000)
+DEFAULT_PROMPT_HISTORY_MAX_CHARS = getattr(
+    settings,
+    "PROMPT_HISTORY_MAX_CHARS",
+    24000,
+)
 DEFAULT_PROMPT_HISTORY_MIN_MESSAGES = getattr(
     settings, "PROMPT_HISTORY_MIN_MESSAGES", 24
 )
@@ -40,7 +44,8 @@ def _build_recent_history(messages, max_chars):
     total_chars = 0
     # Recency-first packing, then reorder to chronological output later.
     for msg in reversed(messages):
-        line = f"[{msg.ts}] <{msg.custom_author if msg.custom_author else msg.session.identifier.person.name}> {msg.text}"
+        author = msg.custom_author or msg.session.identifier.person.name
+        line = f"[{msg.ts}] <{author}> {msg.text}"
         line_len = len(line) + 1
         # Keep at least one line even if it alone exceeds max_chars.
         if selected and (total_chars + line_len) > max_chars:
@@ -147,6 +152,7 @@ async def store_message(session, sender, text, ts, outgoing=False):
         sender_uuid=sender,
         text=text,
         ts=ts,
+        delivered_ts=ts,
         custom_author="USER" if outgoing else None,
     )
 
@@ -161,6 +167,7 @@ async def store_own_message(session, text, ts, manip=None, queue=False):
         "custom_author": "BOT",
         "text": text,
         "ts": ts,
+        "delivered_ts": ts,
     }
     if queue:
         msg_object = QueuedMessage
@@ -177,3 +184,62 @@ async def store_own_message(session, text, ts, manip=None, queue=False):
 
 async def delete_queryset(queryset):
     await sync_to_async(queryset.delete, thread_sensitive=True)()
+
+
+async def apply_read_receipts(
+    user,
+    identifier,
+    message_timestamps,
+    read_ts=None,
+    source_service="signal",
+    read_by_identifier="",
+    payload=None,
+):
+    """
+    Persist delivery/read metadata for one identifier's messages.
+    """
+    ts_values = []
+    for item in message_timestamps or []:
+        try:
+            ts_values.append(int(item))
+        except Exception:
+            continue
+    if not ts_values:
+        return 0
+
+    try:
+        read_at = int(read_ts) if read_ts else None
+    except Exception:
+        read_at = None
+
+    rows = await sync_to_async(list)(
+        Message.objects.filter(
+            user=user,
+            session__identifier=identifier,
+            ts__in=ts_values,
+        ).select_related("session")
+    )
+    updated = 0
+    for message in rows:
+        dirty = []
+        if message.delivered_ts is None:
+            message.delivered_ts = read_at or message.ts
+            dirty.append("delivered_ts")
+        if read_at and (message.read_ts is None or read_at > message.read_ts):
+            message.read_ts = read_at
+            dirty.append("read_ts")
+        if source_service and message.read_source_service != source_service:
+            message.read_source_service = source_service
+            dirty.append("read_source_service")
+        if read_by_identifier and message.read_by_identifier != read_by_identifier:
+            message.read_by_identifier = read_by_identifier
+            dirty.append("read_by_identifier")
+        if payload:
+            receipt_data = dict(message.receipt_payload or {})
+            receipt_data[str(source_service)] = payload
+            message.receipt_payload = receipt_data
+            dirty.append("receipt_payload")
+        if dirty:
+            await sync_to_async(message.save)(update_fields=dirty)
+            updated += 1
+    return updated