sources: SourceIndex chat writer reads and delete_chat, Qdrant and fake impls

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

crates/huginn-daemon/src/index/fakes.rs

@@ -8,7 +8,7 @@ use std::time::Duration;
 
 use anyhow::{Result, bail, ensure};
 
-use crate::sources::{ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta};
+use crate::sources::{ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, SourceIndex, SourcePoint, SourceRow, SourcesMeta, StoredChat};
 
 use super::{CollectionMeta, Described, Embedder, Hit, ModelIdentity, Point, VectorIndex, point_id};
 
@@ -264,6 +264,12 @@ pub(crate) struct SourcesState {
     pub(crate) deletes: Vec<(String, Option<Vec<String>>)>,
     /// A delete naming this repo refuses.
     pub(crate) fail_deletes_of: Option<String>,
+    /// Every chat upsert refuses while this is set.
+    pub(crate) fail_chat_upserts: bool,
+    /// Every chat delete refuses while this is set.
+    pub(crate) fail_chat_deletes: bool,
+    /// Every chat delete that went through: the room and the ids (`None` is all of the room).
+    pub(crate) chat_deletes: Vec<(String, Option<Vec<String>>)>,
     /// The repo filter and limit of every search, in order.
     pub(crate) searches: Vec<(Option<String>, u32)>,
 }
@@ -375,6 +381,7 @@ impl SourceIndex for FakeSources {
 
     fn upsert_chat(&mut self, points: &[ChatPoint]) -> Result<()> {
         let mut state = self.state.lock().unwrap();
+        ensure!(!state.fail_chat_upserts, "the vector store refused the chat upsert");
         for point in points {
             ensure!(!point.payload.messages.is_empty(), "a chat chunk carries no messages");
             state.chat.insert(point.id.clone(), point.clone());
@@ -401,6 +408,44 @@ impl SourceIndex for FakeSources {
         kept.sort_by_key(|payload| payload.messages.iter().map(|message| message.ts_ms).min());
         Ok(kept.into_iter().take(limit as usize).cloned().collect())
     }
+
+    fn chat_tail(&mut self, room_id: &str) -> Result<Option<StoredChat>> {
+        let state = self.state.lock().unwrap();
+        Ok(state.chat.iter().filter(|(_, point)| point.payload.room_id == room_id).max_by_key(|(id, point)| (span(&point.payload).1, (*id).clone())).map(stored))
+    }
+
+    fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result<Vec<StoredChat>> {
+        let state = self.state.lock().unwrap();
+        Ok(state
+            .chat
+            .iter()
+            .filter(|(_, point)| point.payload.room_id == room_id && point.payload.messages.iter().any(|message| event_ids.contains(&message.event_id)))
+            .map(stored)
+            .collect())
+    }
+
+    fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result<Vec<StoredChat>> {
+        let state = self.state.lock().unwrap();
+        Ok(state.chat.iter().filter(|(_, point)| point.payload.room_id == room_id && span(&point.payload).1 >= from_ms && span(&point.payload).0 < to_ms).map(stored).collect())
+    }
+
+    fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()> {
+        let mut state = self.state.lock().unwrap();
+        ensure!(!state.fail_chat_deletes, "the vector store refused the chat delete");
+        state.chat_deletes.push((room_id.to_owned(), ids.map(<[String]>::to_vec)));
+        state.chat.retain(|id, point| !(point.payload.room_id == room_id && ids.is_none_or(|ids| ids.contains(id))));
+        Ok(())
+    }
+}
+
+fn stored((id, point): (&String, &ChatPoint)) -> StoredChat {
+    StoredChat { id: id.clone(), payload: point.payload.clone() }
+}
+
+/// (start, end) of the times a chunk's messages make.
+fn span(payload: &ChatPayload) -> (u64, u64) {
+    let times = || payload.messages.iter().map(|message| message.ts_ms);
+    (times().min().unwrap_or(0), times().max().unwrap_or(0))
 }
 
 /// A chat chunk under a filter, by the span its messages make.
@@ -440,4 +485,16 @@ impl SourceIndex for FakeIndex {
     fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>> {
         self.sources.chat_scroll(filter, limit, within)
     }
+    fn chat_tail(&mut self, room_id: &str) -> Result<Option<StoredChat>> {
+        self.sources.chat_tail(room_id)
+    }
+    fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result<Vec<StoredChat>> {
+        self.sources.chat_chunks_holding(room_id, event_ids)
+    }
+    fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result<Vec<StoredChat>> {
+        self.sources.chat_chunks_overlapping(room_id, from_ms, to_ms)
+    }
+    fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()> {
+        self.sources.delete_chat(room_id, ids)
+    }
 }

crates/huginn-daemon/src/index/qdrant.rs

@@ -11,7 +11,7 @@ use huginn_mind::ChatMessage;
 
 use crate::sources::{
     COLLECTION, CORPUS_CHAT, CORPUS_FORGE, ChatFilter, ChatPayload, ChatPoint, ChatRow, DescribedSources, FRONT_PAGE_FILES, SourceIndex, SourcePayload, SourcePoint, SourceRow,
-    SourcesMeta,
+    SourcesMeta, StoredChat,
 };
 
 use super::{CollectionMeta, INDEX_CALL_TIMEOUT, Described, Hit, Http, Method, Point, VectorIndex};
@@ -455,6 +455,70 @@ impl SourceIndex for QdrantIndex {
         let found = answer["result"]["points"].as_array().context("Qdrant's scroll has no points")?;
         found.iter().map(|point| chat_payload_of(&point["payload"])).collect()
     }
+
+    fn chat_tail(&mut self, room_id: &str) -> Result<Option<StoredChat>> {
+        let order = json!({ "key": "ts_end_ms", "direction": "desc" });
+        Ok(self.scroll_chat(chat_room_filter(room_id, vec![]), Some((order, 1)))?.into_iter().next())
+    }
+
+    fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result<Vec<StoredChat>> {
+        if event_ids.is_empty() {
+            return Ok(Vec::new());
+        }
+        self.scroll_chat(chat_room_filter(room_id, vec![json!({ "key": "event_ids", "match": { "any": event_ids } })]), None)
+    }
+
+    fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result<Vec<StoredChat>> {
+        let span = vec![json!({ "key": "ts_end_ms", "range": { "gte": from_ms } }), json!({ "key": "ts_start_ms", "range": { "lt": to_ms } })];
+        self.scroll_chat(chat_room_filter(room_id, span), None)
+    }
+
+    fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()> {
+        let collection = self.sources.clone();
+        let body = json!({ "filter": chat_delete_filter(room_id, ids) });
+        self.ask(Method::Post, &format!("/collections/{collection}/points/delete?wait=true"), Some(&body), &[200])?;
+        Ok(())
+    }
+}
+
+impl QdrantIndex {
+    /// Every chat chunk `filter` keeps, with its point id, paged; or, given an
+    /// order and a limit, the first of them in that order.
+    fn scroll_chat(&mut self, filter: Value, ordered: Option<(Value, u64)>) -> Result<Vec<StoredChat>> {
+        let collection = self.sources.clone();
+        let mut out = Vec::new();
+        let mut offset = Value::Null;
+        loop {
+            let mut body = json!({ "filter": filter, "limit": SCROLL_PAGE, "with_payload": true, "with_vector": false });
+            if let Some((order, limit)) = &ordered {
+                body["order_by"] = order.clone();
+                body["limit"] = json!(limit);
+            } else if !offset.is_null() {
+                body["offset"] = offset;
+            }
+            let (_, answer) = self.ask(Method::Post, &format!("/collections/{collection}/points/scroll"), Some(&body), &[200])?;
+            for point in answer["result"]["points"].as_array().context("Qdrant's scroll has no points")? {
+                out.push(StoredChat { id: point["id"].as_str().context("a point id is not a string")?.to_owned(), payload: chat_payload_of(&point["payload"])? });
+            }
+            offset = answer["result"]["next_page_offset"].clone();
+            if ordered.is_some() || offset.is_null() {
+                return Ok(out);
+            }
+        }
+    }
+}
+
+/// What a chat delete removes: the room's chat points, all of them or those with `ids`. The corpus
+/// and room conditions hold either way, so a forge point or another room's point cannot match.
+fn chat_delete_filter(room_id: &str, ids: Option<&[String]>) -> Value {
+    chat_room_filter(room_id, ids.map(|ids| json!({ "has_id": ids })).into_iter().collect())
+}
+
+/// The chat points of one room, narrowed by `more` conditions.
+fn chat_room_filter(room_id: &str, more: Vec<Value>) -> Value {
+    let mut must = vec![json!({ "key": "corpus", "match": { "value": CORPUS_CHAT } }), json!({ "key": "room_id", "match": { "value": room_id } })];
+    must.extend(more);
+    json!({ "must": must })
 }
 
 #[cfg(test)]
@@ -517,6 +581,29 @@ mod source_filter_tests {
         assert!(chat_payload_of(&forge).is_err());
     }
 
+    /// The writer-side filters name the chat corpus and the room, so a forge point never matches.
+    #[test]
+    fn the_writer_filters_are_always_chat_corpus_and_the_room() {
+        let filter = chat_room_filter("!r:x", vec![json!({ "key": "event_ids", "match": { "any": ["$a"] } })]);
+        assert_eq!(
+            filter,
+            json!({ "must": [
+                { "key": "corpus", "match": { "value": "chat" } },
+                { "key": "room_id", "match": { "value": "!r:x" } },
+                { "key": "event_ids", "match": { "any": ["$a"] } },
+            ] })
+        );
+    }
+
+    #[test]
+    fn a_chat_delete_names_the_corpus_and_room_with_or_without_ids() {
+        let all = chat_delete_filter("!r:x", None);
+        assert_eq!(all["must"].as_array().unwrap().len(), 2);
+        let some = chat_delete_filter("!r:x", Some(&["p1".to_owned()]));
+        assert_eq!(some["must"][2], json!({ "has_id": ["p1"] }));
+        assert_eq!(some["must"][0], json!({ "key": "corpus", "match": { "value": "chat" } }));
+    }
+
     /// The chat filter is the chat corpus, the rooms as an any-of `should` split
     /// by id and alias, and the two ranges that keep an overlapping span.
     #[test]

crates/huginn-daemon/src/sources/mod.rs

@@ -179,6 +179,14 @@ impl ChatFilter {
     }
 }
 
+/// A chat chunk as the index holds it, with its point id: what the ingest job
+/// reads back to rewrite a chunk.
+#[derive(Clone, Debug, PartialEq)]
+pub struct StoredChat {
+    pub id: String,
+    pub payload: ChatPayload,
+}
+
 /// One chunk a vector search returned and how near it scored.
 #[derive(Clone, Debug, PartialEq)]
 pub struct SourceRow {
@@ -216,6 +224,16 @@ pub trait SourceIndex {
     /// Up to `limit` chat chunks `filter` keeps, oldest first by span start.
     /// Reads only `corpus=chat` points; the daemon's.
     fn chat_scroll(&mut self, filter: &ChatFilter, limit: u32, within: Duration) -> Result<Vec<ChatPayload>>;
+    /// The room's chunk with the greatest `ts_end_ms`, if it has any. The ingest job's alone.
+    fn chat_tail(&mut self, room_id: &str) -> Result<Option<StoredChat>>;
+    /// The room's chunks holding any of `event_ids`. The ingest job's alone.
+    fn chat_chunks_holding(&mut self, room_id: &str, event_ids: &[String]) -> Result<Vec<StoredChat>>;
+    /// The room's chunks whose span reaches into [`from_ms`, `to_ms`): `ts_end_ms >= from_ms`
+    /// and `ts_start_ms < to_ms`. The ingest job's alone.
+    fn chat_chunks_overlapping(&mut self, room_id: &str, from_ms: u64, to_ms: u64) -> Result<Vec<StoredChat>>;
+    /// Deletes a room's chat chunks: those with the point ids `ids`, or all of them when `None`.
+    /// Acknowledged before it returns. Never touches a forge point. The ingest job's alone.
+    fn delete_chat(&mut self, room_id: &str, ids: Option<&[String]>) -> Result<()>;
 }
 
 /// A repo the forge lists as indexable.