From e01cdfc4d921a71a60826dfd1f173a7854b5a47b Mon Sep 17 00:00:00 2001 From: Eureka Date: Wed, 11 Nov 2026 19:34:02 +0000 Subject: [PATCH] sources: refresh_chat, retire_chat, RoomHistory port, steward purge windows, and the tests that falsify them Co-Authored-By: Claude Sonnet 5.5 --- crates/eureka-state/tests/cli.rs | 16 + crates/huginn-daemon/src/index/fakes.rs | 5 + crates/huginn-daemon/src/sources/chat.rs | 509 +++++++++++- .../huginn-daemon/src/sources/chat_purge.rs | 103 +++ .../huginn-daemon/src/sources/chat_tests.rs | 782 ++++++++++++++++++ crates/huginn-daemon/src/sources/mod.rs | 34 +- 6 files changed, 1435 insertions(+), 14 deletions(-) create mode 100644 crates/huginn-daemon/src/sources/chat_purge.rs create mode 100644 crates/huginn-daemon/src/sources/chat_tests.rs diff --git a/crates/eureka-state/tests/cli.rs b/crates/eureka-state/tests/cli.rs index ba446e5..b0a4614 100644 --- a/crates/eureka-state/tests/cli.rs +++ b/crates/eureka-state/tests/cli.rs @@ -1303,6 +1303,22 @@ impl huginn_daemon::sources::SourceIndex for Desk { anyhow::bail!("a daemon never writes the source index") } + fn chat_tail(&mut self, _room_id: &str) -> anyhow::Result> { + anyhow::bail!("a daemon never reads the ingest job's side of the source index") + } + + fn chat_chunks_holding(&mut self, _room_id: &str, _event_ids: &[String]) -> anyhow::Result> { + anyhow::bail!("a daemon never reads the ingest job's side of the source index") + } + + fn chat_chunks_overlapping(&mut self, _room_id: &str, _from_ms: u64, _to_ms: u64) -> anyhow::Result> { + anyhow::bail!("a daemon never reads the ingest job's side of the source index") + } + + fn delete_chat(&mut self, _room_id: &str, _ids: Option<&[String]>) -> anyhow::Result<()> { + anyhow::bail!("a daemon never writes the source index") + } + fn chat_nearest( &mut self, _vector: &[f32], diff --git a/crates/huginn-daemon/src/index/fakes.rs b/crates/huginn-daemon/src/index/fakes.rs index a0d8c77..30e864d 100644 --- a/crates/huginn-daemon/src/index/fakes.rs +++ b/crates/huginn-daemon/src/index/fakes.rs @@ -264,6 +264,8 @@ pub(crate) struct SourcesState { pub(crate) deletes: Vec<(String, Option>)>, /// A delete naming this repo refuses. pub(crate) fail_deletes_of: Option, + /// Every cursor write refuses once this many have been written. + pub(crate) fail_cursor_writes_after: Option, /// Every chat upsert refuses while this is set. pub(crate) fail_chat_upserts: bool, /// Every chat delete refuses while this is set. @@ -332,6 +334,9 @@ impl SourceIndex for FakeSources { fn set_cursors(&mut self, cursors: &BTreeMap) -> Result<()> { let mut state = self.state.lock().unwrap(); + if state.fail_cursor_writes_after.is_some_and(|after| state.cursor_writes.len() >= after) { + bail!("the vector store refused the cursor write"); + } state.cursor_writes.push(cursors.clone()); let Some(DescribedSources::Labelled { cursors: held, .. }) = &mut state.described else { bail!("no source collection") }; *held = cursors.clone(); diff --git a/crates/huginn-daemon/src/sources/chat.rs b/crates/huginn-daemon/src/sources/chat.rs index 6389634..3c895c0 100644 --- a/crates/huginn-daemon/src/sources/chat.rs +++ b/crates/huginn-daemon/src/sources/chat.rs @@ -1,6 +1,28 @@ //! Chat ingest: joined-room history into the shared source collection, beside //! the forge's repos. `refresh_chat` is the only writer of `corpus=chat` points -//! and of the cursors object's chat keys. +//! and of the cursors object's `chat:` keys; `chat_purge::request_purge` is the +//! only writer of `chat-purge:` keys, and `refresh_chat` the only applier. +//! +//! The Matrix client that implements `RoomHistory` is the next cut's and holds no +//! policy. What is in scope is `room_scope`'s; what may be embedded is +//! `huginn_mind::redact::scrub`'s; this file folds events into chunks and writes +//! them, write-then-delete and cursor-last, so any crash replays to the same state. + +use std::collections::{BTreeMap, BTreeSet, HashSet}; +use std::time::Instant; + +use anyhow::{Context, Result, bail, ensure}; +use huginn_mind::ChatMessage; +use huginn_mind::redact::{ScrubKind, scrub}; + +use super::chat_chunk::{chat_version, chunk_id, chunk_messages, chunk_payload, cut_body}; +use super::chat_purge::{PurgeWindow, covers, parse_windows, purge_key}; +use super::room_scope::{Estate, Refusal, RoomFacts, admit_room, refuse_event}; +use super::{CORPUS, ChatPoint, DescribedSources, SourceIndex, StoredChat}; +use crate::index::{BATCH, BULK_EMBED_TIMEOUT, Embedder, INDEX_CALL_TIMEOUT}; + +/// Events asked of the port per page. +pub const CHAT_PAGE: u32 = 200; /// A chat cursor is the key `chat:`; a forge repo name cannot contain /// a colon. The only definition of the prefix. @@ -39,3 +61,488 @@ pub struct RoomEvent { pub content_keys: Vec, pub kind: EventKind, } + +/// One page of a room's events after a cursor event, oldest first. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct EventPage { + pub events: Vec, + /// More events follow this page. + pub more: bool, +} + +/// The reader user's view of Synapse as ingest uses it. Holds no policy: it maps +/// the server's answers to these types and filters nothing. +pub trait RoomHistory { + /// The rooms the reader has joined. The only source of rooms; there is no allow list. + fn joined_rooms(&mut self) -> Result>; + fn facts(&mut self, room_id: &str) -> Result; + /// Up to `limit` events after the event `after` (from the room's start when `None`). + /// `Ok(None)` means `after` is unknown to the server. + fn events_after(&mut self, room_id: &str, after: Option<&str>, limit: u32) -> Result>; +} + +/// The cursors-object key of a room's chat cursor. +pub fn cursor_key(room_id: &str) -> String { + format!("{CHAT_CURSOR_PREFIX}{room_id}") +} + +/// The room a chat cursor key belongs to; `None` for any other key (a purge tombstone included). +pub fn cursor_room(key: &str) -> Option<&str> { + key.strip_prefix(CHAT_CURSOR_PREFIX) +} + +/// A cursor value, `:`; the event is empty before a room's first upsert. +pub fn parse_cursor(value: &str) -> Option<(&str, &str)> { + value.split_once(':') +} + +/// What one admitted room's run did, for the journal. Never carries a body or a sender. +#[derive(Clone, Debug, PartialEq)] +pub struct ChatReport { + pub room_id: String, + pub alias: Option, + pub first: Option, + pub last: Option, + pub events: u32, + pub chunks: u32, + pub scrubbed: u32, + pub purged: u32, + pub seconds: f64, +} + +impl std::fmt::Display for ChatReport { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + formatter, + "{} ({}) {}..{} events={} chunks={} scrubbed={} purged={} seconds={:.1}", + self.alias.as_deref().unwrap_or("-"), + self.room_id, + self.first.as_deref().unwrap_or("-"), + self.last.as_deref().unwrap_or("-"), + self.events, + self.chunks, + self.scrubbed, + self.purged, + self.seconds + ) + } +} + +/// A joined room the scope refused, whole, and why; `event_id` when an event did it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RefusedRoom { + pub room_id: String, + pub reason: Refusal, + pub event_id: Option, +} + +/// A credential the scrub replaced: where, never what. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ScrubHit { + pub room_id: String, + pub event_id: String, + pub kind: ScrubKind, +} + +/// What a chat refresh did. +#[derive(Clone, Debug, Default, PartialEq)] +pub struct ChatRefreshed { + pub rooms: Vec, + /// Rooms (or steps) that failed; the others went on. + pub failed: Vec, + pub refused: Vec, + pub scrubbed: Vec, +} + +impl ChatRefreshed { + /// Nothing failed and no joined room had to be refused: a joined estate, direct, + /// encrypted or bridged room means the reader's roster is wrong. A scrub hit does not make a run unclean. + pub fn is_clean(&self) -> bool { + self.failed.is_empty() && self.refused.is_empty() + } +} + +fn read_cursors(index: &mut S) -> Result> { + match index.describe_sources(INDEX_CALL_TIMEOUT).context("describing the source collection")? { + DescribedSources::Labelled { cursors, .. } => Ok(cursors), + _ => bail!("the source collection is gone; the forge refresh creates it"), + } +} + +fn put_cursor(index: &mut S, key: &str, value: &str) -> Result<()> { + let mut cursors = read_cursors(index)?; + if cursors.insert(key.to_owned(), value.to_owned()).as_deref() != Some(value) { + index.set_cursors(&cursors).context("writing a chat cursor")?; + } + Ok(()) +} + +fn drop_cursor(index: &mut S, key: &str) -> Result<()> { + let mut cursors = read_cursors(index)?; + if cursors.remove(key).is_some() { + index.set_cursors(&cursors).context("dropping a chat cursor")?; + } + Ok(()) +} + +/// A room's points, then its `chat:` cursor. Its purge tombstone stays. +fn purge_room(index: &mut S, room_id: &str) -> Result<()> { + index.delete_chat(room_id, None).with_context(|| format!("deleting the chunks of {room_id}"))?; + drop_cursor(index, &cursor_key(room_id)) +} + +/// Deletes the points and `chat:` cursor of every room not in `keep` (the reader left it, +/// or it is refused or excluded), and returns those rooms. Purge tombstones and forge +/// repos are never touched. With the empty set it removes the whole chat corpus. +pub fn retire_chat(index: &mut S, keep: &BTreeSet) -> Result> { + let rooms: Vec = read_cursors(index)?.keys().filter_map(|key| cursor_room(key)).filter(|room| !keep.contains(*room)).map(str::to_owned).collect(); + for room in &rooms { + purge_room(index, room)?; + } + Ok(rooms) +} + +/// A change to chunks already stored, in event order. +enum Op { + /// An edit by `sender` of `target`'s body; applies only to the original's sender. + Edit { target: String, sender: String, body: String }, + /// A held message's body is what the page says it is (a replay). + Set { target: String, body: String }, + Redact { target: String }, +} + +impl Op { + fn target(&self) -> &str { + match self { + Self::Edit { target, .. } | Self::Set { target, .. } | Self::Redact { target } => target, + } + } + + /// Applies to `messages`; whether they changed. + fn apply(&self, messages: &mut Vec) -> bool { + let Some(at) = messages.iter().position(|message| message.event_id == self.target()) else { return false }; + match self { + Self::Edit { sender, body, .. } if messages[at].sender == *sender && messages[at].body != *body => messages[at].body.clone_from(body), + Self::Set { body, .. } if messages[at].body != *body => messages[at].body.clone_from(body), + Self::Redact { .. } => { + messages.remove(at); + } + _ => return false, + } + true + } +} + +struct Folded { + pending: Vec, + ops: Vec, +} + +struct Room<'a> { + facts: &'a RoomFacts, + label: String, + windows: Vec, +} + +/// The page folded in order: the purge filter, then the credential scrub, then the +/// event's effect. After this no raw body exists. +fn fold(room: &Room, events: &[RoomEvent], report: &mut ChatReport, out: &mut ChatRefreshed) -> Folded { + let mut folded = Folded { pending: Vec::new(), ops: Vec::new() }; + let mut redacted: HashSet<&str> = HashSet::new(); + let mut scrubbed = |event: &RoomEvent, body: &str| { + let clean = scrub(body); + for kind in clean.hits { + report.scrubbed += 1; + out.scrubbed.push(ScrubHit { room_id: room.facts.room_id.clone(), event_id: event.event_id.clone(), kind }); + } + cut_body(&clean.text) + }; + for event in events { + match &event.kind { + EventKind::Message { body } => { + if covers(&room.windows, event.ts_ms) { + report.purged += 1; + } else if !redacted.contains(event.event_id.as_str()) { + let body = scrubbed(event, body); + if !body.trim().is_empty() { + folded.pending.push(ChatMessage { event_id: event.event_id.clone(), sender: event.sender.clone(), ts_ms: event.ts_ms, body }); + } + } + } + EventKind::Edit { target, body } => { + if covers(&room.windows, event.ts_ms) { + report.purged += 1; + continue; + } + let body = scrubbed(event, body); + match folded.pending.iter_mut().find(|message| message.event_id == *target) { + Some(message) if message.sender == event.sender && !body.trim().is_empty() => message.body = body, + Some(_) => {} + None => folded.ops.push(Op::Edit { target: target.clone(), sender: event.sender.clone(), body }), + } + } + EventKind::Redaction { target } => { + folded.pending.retain(|message| message.event_id != *target); + redacted.insert(target); + folded.ops.push(Op::Redact { target: target.clone() }); + } + EventKind::Other => {} + } + } + folded +} + +/// Chunks to rewrite: the stored chunk ids they replace and the messages that now make them. +type Rewrite = (Vec, Vec); + +/// Chunks `rewrites`, embeds and writes the new chunks (acknowledged), and only then deletes the old +/// ids that are not among the new. Returns how many chunks were written. +/// +/// A chunk written by an earlier run whose delete failed can still sit beside the chunk that replaced +/// it, and a rewrite of the stale one then names the same first event, hence the same id, as the +/// newer chunk. The newer chunk (`standing`, the other chunks the page found) stays; the stale one is deleted. +fn write_rewrites( + embedder: &mut E, + index: &mut S, + dimensions: u32, + room: &Room, + rewrites: Vec, + standing: &BTreeSet, +) -> Result { + let mut old: BTreeSet = BTreeSet::new(); + let mut points: Vec<(String, super::ChatPayload)> = Vec::new(); + for (ids, messages) in rewrites { + for group in chunk_messages(&room.label, &messages) { + let id = chunk_id(&room.facts.room_id, &group[0].event_id); + if standing.contains(&id) && !ids.contains(&id) { + continue; + } + points.push((id, chunk_payload(&room.facts.room_id, room.facts.alias.as_deref(), &room.label, group))); + } + old.extend(ids); + } + let new: BTreeSet<&String> = points.iter().map(|(id, _)| id).collect(); + let stale: Vec = old.iter().filter(|id| !new.contains(id)).cloned().collect(); + let written = u32::try_from(points.len()).unwrap_or(u32::MAX); + for slice in points.chunks(BATCH) { + let texts: Vec = slice.iter().map(|(_, payload)| payload.text.clone()).collect(); + let vectors = embedder.embed(&texts, BULK_EMBED_TIMEOUT).context("embedding chat chunks")?; + ensure!(vectors.len() == texts.len(), "the embedder returned {} vectors for {} chunks", vectors.len(), texts.len()); + let mut batch = Vec::with_capacity(slice.len()); + for ((id, payload), vector) in slice.iter().zip(vectors) { + ensure!(vector.len() == dimensions as usize, "the embedder returned a vector of {} where the collection holds {dimensions}", vector.len()); + batch.push(ChatPoint { id: id.clone(), vector, payload: payload.clone() }); + } + index.upsert_chat(&batch).context("writing chat chunks")?; + } + if !stale.is_empty() { + index.delete_chat(&room.facts.room_id, Some(&stale)).context("deleting superseded chat chunks")?; + } + Ok(written) +} + +/// Applies the room's purge windows to the chunks it already holds. +fn apply_purge(embedder: &mut E, index: &mut S, dimensions: u32, room: &Room, report: &mut ChatReport) -> Result<()> { + let mut hit: BTreeMap = BTreeMap::new(); + for window in &room.windows { + for chunk in index.chat_chunks_overlapping(&room.facts.room_id, window.from_ms, window.to_ms)? { + hit.insert(chunk.id.clone(), chunk); + } + } + let mut rewrites = Vec::new(); + for chunk in hit.into_values() { + let kept: Vec = chunk.payload.messages.iter().filter(|message| !covers(&room.windows, message.ts_ms)).cloned().collect(); + let removed = chunk.payload.messages.len() - kept.len(); + if removed > 0 { + report.purged += u32::try_from(removed).unwrap_or(u32::MAX); + rewrites.push((vec![chunk.id], kept)); + } + } + report.chunks += write_rewrites(embedder, index, dimensions, room, rewrites, &BTreeSet::new())?; + Ok(()) +} + +/// Folds one page into the stored chunks: edits and redactions rewrite the chunk that held +/// the target alone, new messages extend the open tail. +fn write_page(embedder: &mut E, index: &mut S, dimensions: u32, room: &Room, mut folded: Folded) -> Result { + let room_id = &room.facts.room_id; + let mut asked: Vec = folded.pending.iter().map(|message| message.event_id.clone()).collect(); + asked.extend(folded.ops.iter().map(|op| op.target().to_owned())); + if asked.is_empty() { + return Ok(0); + } + let holding = index.chat_chunks_holding(room_id, &asked)?; + // A message the index already holds is a replay: its page body is asserted, never appended again. + let held: HashSet<&str> = holding.iter().flat_map(|chunk| chunk.payload.messages.iter().map(|message| message.event_id.as_str())).collect(); + let (replayed, fresh): (Vec, Vec) = folded.pending.drain(..).partition(|message| held.contains(message.event_id.as_str())); + let mut ops: Vec = replayed.into_iter().map(|message| Op::Set { target: message.event_id, body: message.body }).collect(); + ops.append(&mut folded.ops); + + let tail = index.chat_tail(room_id)?; + let tail_id = tail.as_ref().map(|tail| tail.id.clone()); + let standing: BTreeSet = holding.iter().map(|chunk| chunk.id.clone()).chain(tail_id.clone()).collect(); + let mut tail_messages = tail.map(|tail| tail.payload.messages).unwrap_or_default(); + let mut tail_changed = false; + let mut rewrites: Vec = Vec::new(); + for chunk in holding { + let mut messages = chunk.payload.messages.clone(); + let mut changed = false; + for op in &ops { + changed |= op.apply(&mut messages); + } + if Some(&chunk.id) == tail_id.as_ref() { + (tail_messages, tail_changed) = (messages, changed); + } else if changed { + rewrites.push((vec![chunk.id], messages)); + } + } + if !fresh.is_empty() || tail_changed { + tail_messages.extend(fresh); + rewrites.push((tail_id.into_iter().collect(), tail_messages)); + } + write_rewrites(embedder, index, dimensions, room, rewrites, &standing) +} + +enum Outcome { + Done(ChatReport), + Refused(RefusedRoom), +} + +#[allow(clippy::too_many_arguments)] +fn ingest_room( + embedder: &mut E, + index: &mut S, + history: &mut H, + dimensions: u32, + facts: &RoomFacts, + out: &mut ChatRefreshed, +) -> Result { + let started = Instant::now(); + let room_id = facts.room_id.as_str(); + let (key, version) = (cursor_key(room_id), chat_version()); + let cursors = read_cursors(index)?; + let mut after: Option = match cursors.get(&key).map(|value| parse_cursor(value)) { + None => None, + Some(Some((held, event))) if held == version => Some(event.to_owned()).filter(|event| !event.is_empty()), + // Another chunker or scrub version, or a value nobody here wrote: this room alone starts over. + Some(_) => { + index.delete_chat(room_id, None).with_context(|| format!("deleting the chunks of {room_id} for a re-ingest"))?; + None + } + }; + if after.is_none() { + put_cursor(index, &key, &format!("{version}:"))?; + } + let windows = cursors.get(&purge_key(room_id)).map(|value| parse_windows(value)).transpose()?.unwrap_or_default(); + let label = facts.name.clone().or_else(|| facts.alias.clone()).unwrap_or_else(|| room_id.to_owned()); + let room = Room { facts, label, windows }; + let mut report = ChatReport { room_id: room_id.to_owned(), alias: facts.alias.clone(), first: None, last: None, events: 0, chunks: 0, scrubbed: 0, purged: 0, seconds: 0.0 }; + apply_purge(embedder, index, dimensions, &room, &mut report)?; + let mut restarted = false; + loop { + let Some(page) = history.events_after(room_id, after.as_deref(), CHAT_PAGE)? else { + ensure!(after.is_some() && !restarted, "the server does not know the start of {room_id}'s history"); + restarted = true; + purge_room(index, room_id)?; + put_cursor(index, &key, &format!("{version}:"))?; + after = None; + continue; + }; + let Some(last) = page.events.last().map(|event| event.event_id.clone()) else { break }; + // A refusing event refuses the room before anything of the page is chunked or embedded. + if let Some((event, reason)) = page.events.iter().find_map(|event| refuse_event(event).map(|reason| (event, reason))) { + purge_room(index, room_id)?; + return Ok(Outcome::Refused(RefusedRoom { room_id: room_id.to_owned(), reason, event_id: Some(event.event_id.clone()) })); + } + report.first.get_or_insert_with(|| page.events[0].event_id.clone()); + report.events += u32::try_from(page.events.len()).unwrap_or(u32::MAX); + let folded = fold(&room, &page.events, &mut report, out); + report.chunks += write_page(embedder, index, dimensions, &room, folded)?; + put_cursor(index, &key, &format!("{version}:{last}"))?; + report.last = Some(last.clone()); + after = Some(last); + if !page.more { + break; + } + } + report.seconds = started.elapsed().as_secs_f64(); + Ok(Outcome::Done(report)) +} + +/// Brings the chat corpus to the reader's joined rooms. The collection must already exist and be +/// Huginn's, built under this embedder's model: the forge refresh creates and rebuilds it, and this never does. +/// A joined room the scope refuses is purged and named; a room the reader left is retired; the rest +/// are ingested from their cursors. One room's failure is journaled and the others go on. +/// `journal` receives every line the run prints: reports, scrub hits, refusals and failures. +pub fn refresh_chat( + embedder: &mut E, + index: &mut S, + history: &mut H, + estate: &Estate, + journal: &mut dyn FnMut(String), +) -> Result { + let DescribedSources::Labelled { meta, .. } = index.describe_sources(INDEX_CALL_TIMEOUT).context("describing the source collection")? else { + bail!("the source collection is absent or not Huginn's; the forge refresh creates it and chat never does"); + }; + ensure!(meta.corpus == CORPUS, "the source collection was built for corpus {} and chat ingest writes {CORPUS}; the forge refresh rebuilds it", meta.corpus); + embedder.built_under(Some(&meta.model_identity())); + let identity = embedder.model_identity(BULK_EMBED_TIMEOUT).context("reading the embedding model's identity")?; + ensure!(meta.model_identity() == identity, "the source collection was built under {} {} and chat embeds under {} {}", meta.model, meta.model_digest, identity.name, identity.digest); + + let mut out = ChatRefreshed::default(); + let mut joined = history.joined_rooms().context("listing the reader's joined rooms")?; + let mut seen = HashSet::new(); + joined.retain(|room| seen.insert(room.clone())); + let mut keep: BTreeSet = joined.iter().cloned().collect(); + let mut admitted: Vec = Vec::new(); + for room_id in &joined { + let verdict = if estate.holds_id(room_id) { + Ok(Err(Refusal::Estate)) + } else { + history.facts(room_id).map(|facts| admit_room(&facts, estate).map(|()| facts)) + }; + match verdict { + Ok(Ok(facts)) => admitted.push(facts), + Ok(Err(reason)) => { + keep.remove(room_id); + match purge_room(index, room_id) { + Ok(()) => {} + Err(error) => { + journal(format!("huginn-sources: chat {room_id} could not be purged, the others go on: {error:#}")); + out.failed.push(room_id.clone()); + } + } + journal(format!("refused: {room_id} {reason}")); + out.refused.push(RefusedRoom { room_id: room_id.clone(), reason, event_id: None }); + } + Err(error) => { + journal(format!("huginn-sources: chat {room_id} facts failed, the others go on: {error:#}")); + out.failed.push(room_id.clone()); + } + } + } + if let Err(error) = retire_chat(index, &keep) { + journal(format!("huginn-sources: retiring rooms the reader left failed, the others go on: {error:#}")); + out.failed.push("retire".to_owned()); + } + for facts in admitted { + match ingest_room(embedder, index, history, identity.dimensions, &facts, &mut out) { + Ok(Outcome::Done(report)) => { + journal(report.to_string()); + out.rooms.push(report); + } + Ok(Outcome::Refused(refused)) => { + journal(format!("refused: {} {} [{}]", refused.room_id, refused.reason, refused.event_id.as_deref().unwrap_or(""))); + out.refused.push(refused); + } + Err(error) => { + journal(format!("huginn-sources: chat {} failed, the others go on: {error:#}", facts.room_id)); + out.failed.push(facts.room_id.clone()); + } + } + } + for hit in &out.scrubbed { + journal(format!("scrubbed: {} {} {}", hit.room_id, hit.event_id, hit.kind)); + } + Ok(out) +} diff --git a/crates/huginn-daemon/src/sources/chat_purge.rs b/crates/huginn-daemon/src/sources/chat_purge.rs new file mode 100644 index 0000000..fbc1778 --- /dev/null +++ b/crates/huginn-daemon/src/sources/chat_purge.rs @@ -0,0 +1,103 @@ +//! A steward's purge of a room's time window. The tombstone is the one chat +//! state that is not derived from Synapse: a durable request, kept in the +//! collection's cursors object under `chat-purge:`, that must outlive +//! the points it removes, so a re-ingest never brings the window back. +//! +//! `request_purge` is the only writer; `refresh_chat` reads the tombstone and +//! applies it at its next run. The code does not authenticate the steward: whoever +//! can run the job as the unit's user (the host admin) may ask, and the request +//! carries a ref in the job's journal for a trace. Tombstones live in the +//! collection, so a full collection rebuild forgets them; re-issue them from +//! their refs. + +use anyhow::{Context, Result, bail}; + +use super::chat::CHAT_PURGE_PREFIX; +use super::{DescribedSources, SourceIndex}; +use crate::index::INDEX_CALL_TIMEOUT; + +/// The most windows a room's tombstone holds. +pub const PURGE_WINDOWS_MAX: usize = 32; + +/// Messages with `from_ms <= ts_ms < to_ms` are purged. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct PurgeWindow { + pub from_ms: u64, + pub to_ms: u64, +} + +impl PurgeWindow { + pub fn covers(&self, ts_ms: u64) -> bool { + self.from_ms <= ts_ms && ts_ms < self.to_ms + } +} + +/// The tombstone key of a room. +pub fn purge_key(room_id: &str) -> String { + format!("{CHAT_PURGE_PREFIX}{room_id}") +} + +/// A tombstone value, `..[,.....]`. +pub fn parse_windows(value: &str) -> Result> { + value + .split(',') + .map(|part| { + let (from, to) = part.split_once("..").with_context(|| format!("purge window {part:?} is not .."))?; + let window = PurgeWindow { from_ms: from.parse().context("a purge window's start is not milliseconds")?, to_ms: to.parse().context("a purge window's end is not milliseconds")? }; + if window.from_ms >= window.to_ms { + bail!("purge window {part:?} is empty"); + } + Ok(window) + }) + .collect() +} + +/// Windows sorted, with overlapping and touching ones joined. +pub fn merge(mut windows: Vec) -> Vec { + windows.sort(); + let mut merged: Vec = Vec::new(); + for window in windows { + match merged.last_mut() { + Some(last) if window.from_ms <= last.to_ms => last.to_ms = last.to_ms.max(window.to_ms), + _ => merged.push(window), + } + } + merged +} + +pub fn encode(windows: &[PurgeWindow]) -> String { + windows.iter().map(|window| format!("{}..{}", window.from_ms, window.to_ms)).collect::>().join(",") +} + +/// Whether any window covers `ts_ms`. +pub fn covers(windows: &[PurgeWindow], ts_ms: u64) -> bool { + windows.iter().any(|window| window.covers(ts_ms)) +} + +/// Records that `window` of `room_id` must leave the index and never return. Writes the cursors +/// object only if the tombstone changed, and refuses a room's 33rd window. Writes no point and +/// embeds nothing; the next `refresh_chat` applies it. The job's `chat-purge` subcommand calls +/// it under the job's lock, because `set_cursors` replaces the cursors object whole. +/// Returns whether the tombstone changed. +pub fn request_purge(index: &mut S, room_id: &str, window: PurgeWindow) -> Result { + if window.from_ms >= window.to_ms { + bail!("a purge window must start before it ends"); + } + let DescribedSources::Labelled { mut cursors, .. } = index.describe_sources(INDEX_CALL_TIMEOUT).context("describing the source collection")? else { + bail!("the source collection is absent or not Huginn's; the forge refresh creates it"); + }; + let key = purge_key(room_id); + let mut windows = cursors.get(&key).map(|held| parse_windows(held)).transpose()?.unwrap_or_default(); + windows.push(window); + let windows = merge(windows); + if windows.len() > PURGE_WINDOWS_MAX { + bail!("room {room_id} already holds {PURGE_WINDOWS_MAX} purge windows; widen one instead"); + } + let encoded = encode(&windows); + if cursors.get(&key) == Some(&encoded) { + return Ok(false); + } + cursors.insert(key, encoded); + index.set_cursors(&cursors).context("writing the purge tombstone")?; + Ok(true) +} diff --git a/crates/huginn-daemon/src/sources/chat_tests.rs b/crates/huginn-daemon/src/sources/chat_tests.rs new file mode 100644 index 0000000..7eb6e60 --- /dev/null +++ b/crates/huginn-daemon/src/sources/chat_tests.rs @@ -0,0 +1,782 @@ +//! Chat ingest against a scripted room history, a fake vector store and the fake embedder. +//! The refusals come first: what must never be read, embedded or stored. + +use std::collections::{BTreeMap, BTreeSet}; +use std::time::Duration; + +use anyhow::{Result, bail}; +use huginn_mind::ChatMessage; +use huginn_mind::redact::ScrubKind; + +use super::chat::*; +use super::chat_chunk::{CHAT_BODY_CHARS_MAX, chat_version, chunk_id}; +use super::chat_purge::{PURGE_WINDOWS_MAX, PurgeWindow, request_purge}; +use super::room_scope::{ESTATE_ALIASES, Estate, Refusal, RoomFacts}; +use super::*; +use crate::index::MANAGED_BY; +use crate::index::fakes::{FakeEmbedder, FakeSources}; + +// ---- the rig ---- + +struct Scripted { + facts: RoomFacts, + events: Vec, +} + +#[derive(Default)] +struct FakeHistory { + joined: Vec, + rooms: BTreeMap, + calls: Vec, +} + +impl FakeHistory { + fn room(&mut self, id: &str, events: Vec) -> &mut Self { + self.rooms.insert(id.into(), Scripted { facts: RoomFacts { room_id: id.into(), ..RoomFacts::default() }, events }); + self.joined.push(id.into()); + self + } + + fn facts(&mut self, id: &str, edit: impl FnOnce(&mut RoomFacts)) -> &mut Self { + edit(&mut self.rooms.get_mut(id).unwrap().facts); + self + } + + fn push(&mut self, id: &str, events: Vec) { + self.rooms.get_mut(id).unwrap().events.extend(events); + } + + fn called(&self, prefix: &str, room: &str) -> bool { + self.calls.iter().any(|call| call == &format!("{prefix}:{room}")) + } +} + +impl RoomHistory for FakeHistory { + fn joined_rooms(&mut self) -> Result> { + self.calls.push("joined".into()); + Ok(self.joined.clone()) + } + + fn facts(&mut self, room_id: &str) -> Result { + self.calls.push(format!("facts:{room_id}")); + match self.rooms.get(room_id) { + Some(room) => Ok(room.facts.clone()), + None => bail!("no such room"), + } + } + + fn events_after(&mut self, room_id: &str, after: Option<&str>, limit: u32) -> Result> { + self.calls.push(format!("events:{room_id}")); + let Some(room) = self.rooms.get(room_id) else { bail!("no such room") }; + let from = match after { + None => 0, + Some(after) => match room.events.iter().position(|event| event.event_id == after) { + Some(at) => at + 1, + None => return Ok(None), + }, + }; + let rest = &room.events[from..]; + let take = rest.len().min(limit as usize); + Ok(Some(EventPage { events: rest[..take].to_vec(), more: take < rest.len() })) + } +} + +fn event(id: &str, sender: &str, ts_ms: u64, kind: EventKind) -> RoomEvent { + RoomEvent { event_id: id.into(), sender: sender.into(), ts_ms, content_keys: vec!["body".into(), "msgtype".into()], kind } +} + +fn said(id: &str, ts_ms: u64, body: &str) -> RoomEvent { + event(id, "@a:s", ts_ms, EventKind::Message { body: body.into() }) +} + +fn edit(id: &str, sender: &str, ts_ms: u64, target: &str, body: &str) -> RoomEvent { + event(id, sender, ts_ms, EventKind::Edit { target: target.into(), body: body.into() }) +} + +fn redaction(id: &str, ts_ms: u64, target: &str) -> RoomEvent { + event(id, "@a:s", ts_ms, EventKind::Redaction { target: target.into() }) +} + +struct Rig { + embedder: FakeEmbedder, + index: FakeSources, + estate: Estate, +} + +impl Rig { + fn new() -> Self { + let mut embedder = FakeEmbedder::new("d1"); + let index = FakeSources::default(); + let identity = crate::index::Embedder::model_identity(&mut embedder, Duration::from_secs(1)).unwrap(); + let meta = SourcesMeta { + managed_by: MANAGED_BY.into(), + corpus: CORPUS.into(), + model: identity.name, + model_digest: identity.digest, + dimensions: identity.dimensions, + chunker_version: chunk::CHUNKER_VERSION, + }; + index.plant(DescribedSources::Labelled { meta, cursors: BTreeMap::new() }); + Self { embedder, index, estate: Estate::default() } + } + + fn run(&mut self, history: &mut FakeHistory) -> (Result, Vec) { + let mut journal = Vec::new(); + let out = refresh_chat(&mut self.embedder, &mut self.index, history, &self.estate, &mut |line| journal.push(line)); + (out, journal) + } + + fn ok(&mut self, history: &mut FakeHistory) -> ChatRefreshed { + self.run(history).0.unwrap() + } + + fn points(&self) -> Vec { + self.index.state.lock().unwrap().chat.values().cloned().collect() + } + + fn ids(&self) -> BTreeSet { + self.index.state.lock().unwrap().chat.keys().cloned().collect() + } + + fn bodies(&self) -> Vec { + let mut bodies: Vec = self.points().iter().flat_map(|point| point.payload.messages.iter().map(|message| message.body.clone())).collect(); + bodies.sort(); + bodies + } + + fn held_ids(&self, room: &str) -> Vec { + let mut ids: Vec = + self.points().iter().filter(|point| point.payload.room_id == room).flat_map(|point| point.payload.messages.iter().map(|message| message.event_id.clone())).collect(); + ids.sort(); + ids + } + + fn cursor(&self, room: &str) -> Option { + self.index.cursors().get(&cursor_key(room)).cloned() + } + + fn embedded(&self) -> Vec { + self.embedder.texts() + } + + fn seed(&mut self, room: &str, event_id: &str, ts: u64) { + let mut seeded = vec![ChatPoint { + id: chunk_id(room, event_id), + vector: vec![1.0; 4], + payload: chunk_payload_of(room, &[(event_id, ts)]), + }]; + self.index.upsert_chat(&seeded.split_off(0)).unwrap(); + let mut cursors = self.index.cursors(); + cursors.insert(cursor_key(room), format!("{}:{event_id}", chat_version())); + self.index.set_cursors(&cursors).unwrap(); + } +} + +fn chunk_payload_of(room: &str, messages: &[(&str, u64)]) -> ChatPayload { + let messages: Vec = messages.iter().map(|(id, ts)| ChatMessage { event_id: (*id).into(), sender: "@a:s".into(), ts_ms: *ts, body: format!("seeded {id}") }).collect(); + ChatPayload { room_id: room.into(), room_alias: None, text: "seeded".into(), messages } +} + +fn tombstone(rig: &Rig, room: &str) -> Option { + rig.index.cursors().get(&format!("{CHAT_PURGE_PREFIX}{room}")).cloned() +} + +// ---- scope: refusals first ---- + +#[test] +fn only_joined_rooms_are_looked_at_and_a_join_rule_is_not_the_test() { + let mut history = FakeHistory::default(); + history.room("!a:s", vec![said("$a1", 1, "in a")]).room("!b:s", vec![said("$b1", 2, "in b")]); + // A room the reader has not joined: scripted, but never listed. + history.rooms.insert("!c:s".into(), Scripted { facts: RoomFacts { room_id: "!c:s".into(), ..RoomFacts::default() }, events: vec![said("$c1", 3, "in c")] }); + let mut rig = Rig::new(); + let done = rig.ok(&mut history); + assert!(done.is_clean()); + assert!(!history.calls.iter().any(|call| call.contains("!c:s")), "{:?}", history.calls); + assert_eq!(rig.held_ids("!a:s"), ["$a1"]); + assert_eq!(rig.held_ids("!b:s"), ["$b1"]); + assert!(rig.held_ids("!c:s").is_empty() && done.rooms.iter().all(|report| report.room_id != "!c:s")); + let code = include_str!("room_scope.rs"); + assert!(!code[..code.find("#[cfg(test)]").unwrap()].contains(&["join", "_rule"].concat())); +} + +#[test] +fn a_dm_room_is_refused_unread_unindexed_and_purged() { + let mut history = FakeHistory::default(); + history.room("!dm:s", vec![said("$d1", 1, "private")]).facts("!dm:s", |facts| facts.direct = true); + let mut rig = Rig::new(); + rig.seed("!dm:s", "$old", 5); + let (done, journal) = rig.run(&mut history); + let done = done.unwrap(); + assert!(!history.called("events", "!dm:s")); + assert!(rig.points().is_empty() && rig.cursor("!dm:s").is_none() && rig.embedded().is_empty()); + assert_eq!(done.refused, [RefusedRoom { room_id: "!dm:s".into(), reason: Refusal::Direct, event_id: None }]); + assert!(!done.is_clean() && journal.iter().any(|line| line == "refused: !dm:s direct room"), "{journal:?}"); +} + +#[test] +fn each_estate_room_is_refused_by_id_and_by_alias_and_an_excluded_room_in_the_index_is_purged() { + let mut rig = Rig::new(); + rig.estate = Estate { ids: ["!byid:s".to_owned()].into() }; + let mut history = FakeHistory::default(); + history.room("!byid:s", vec![said("$i1", 1, "x")]); + for (n, alias) in ESTATE_ALIASES.iter().map(|alias| alias.to_string()).chain(["#USERS:GAMECULT.ORG".to_owned()]).enumerate() { + history.room(&format!("!al{n}:s"), vec![said("$x", 1, "x")]).facts(&format!("!al{n}:s"), |facts| facts.alias = Some(alias)); + } + history.room("!alt:s", vec![said("$x", 1, "x")]).facts("!alt:s", |facts| facts.alt_aliases = vec!["#other:s".into(), "#patrons:gamecult.org".into()]); + history.room("!ok:s", vec![said("$ok1", 1, "fine")]); + history.room("!guide:s", vec![said("$g1", 1, "fine")]).facts("!guide:s", |facts| facts.alias = Some("#users-guide:gamecult.org".into())); + history.room("!club:s", vec![said("$c1", 1, "fine")]).facts("!club:s", |facts| facts.alias = Some("#patrons-club:gamecult.org".into())); + for room in ["!byid:s", "!al0:s", "!al1:s", "!al2:s", "!al3:s", "!alt:s"] { + rig.seed(room, "$old", 5); + } + let mut cursors = rig.index.cursors(); + cursors.insert(format!("{CHAT_PURGE_PREFIX}!al0:s"), "1..2".into()); + rig.index.set_cursors(&cursors).unwrap(); + let done = rig.ok(&mut history); + assert!(!history.called("facts", "!byid:s") && !history.called("events", "!byid:s")); + for room in ["!al0:s", "!al1:s", "!al2:s", "!al3:s", "!alt:s"] { + assert!(history.called("facts", room) && !history.called("events", room), "{room}"); + } + let refused: BTreeSet<&str> = done.refused.iter().map(|refused| refused.room_id.as_str()).collect(); + assert_eq!(refused, ["!al0:s", "!al1:s", "!al2:s", "!al3:s", "!alt:s", "!byid:s"].into()); + assert!(done.refused.iter().all(|refused| refused.reason == Refusal::Estate) && !done.is_clean()); + let held: BTreeSet = rig.points().iter().map(|point| point.payload.room_id.clone()).collect(); + assert_eq!(held, ["!guide:s".to_owned(), "!club:s".to_owned(), "!ok:s".to_owned()].into(), "lookalikes and plain rooms are in; every estate room is out"); + assert!(rig.index.cursors().keys().all(|key| !key.starts_with("chat:!al") && !key.starts_with("chat:!byid") && !key.starts_with("chat:!alt"))); + assert_eq!(tombstone(&rig, "!al0:s").as_deref(), Some("1..2"), "a tombstone is never touched by a refusal"); +} + +#[test] +fn an_encrypted_room_is_refused_and_an_encrypted_event_in_a_plain_room_is_never_stored() { + let mut history = FakeHistory::default(); + history.room("!enc:s", vec![said("$e1", 1, "x")]).facts("!enc:s", |facts| facts.encrypted = true); + history.room("!plain:s", vec![said("$p1", 1, "visible"), event("$p2", "@a:s", 2, EventKind::Other), said("$p3", 3, "also visible")]); + let mut rig = Rig::new(); + let done = rig.ok(&mut history); + assert_eq!(done.refused[0].reason, Refusal::Encrypted); + assert!(!history.called("events", "!enc:s")); + assert_eq!(rig.held_ids("!plain:s"), ["$p1", "$p3"]); + assert_eq!(rig.cursor("!plain:s").unwrap(), format!("{}:$p3", chat_version()), "the cursor passes an Other event"); +} + +#[test] +fn a_bridged_room_is_refused_whole_before_any_event_is_read_and_its_chunks_are_purged() { + let mut history = FakeHistory::default(); + for (n, state_type) in ["m.bridge", "uk.half-shot.bridge"].into_iter().enumerate() { + let room = format!("!br{n}:s"); + history.room(&room, vec![said("$x", 1, "x")]).facts(&room, |facts| facts.bridge = Some(state_type.into())); + } + let mut rig = Rig::new(); + rig.seed("!br0:s", "$old", 5); + let done = rig.ok(&mut history); + assert_eq!(done.refused.len(), 2); + assert!(done.refused.iter().all(|refused| matches!(refused.reason, Refusal::Bridged(_)))); + assert!(!history.calls.iter().any(|call| call.starts_with("events:"))); + assert!(rig.points().is_empty() && rig.index.cursors().is_empty()); +} + +#[test] +fn a_bridged_platforms_sender_or_content_refuses_the_room_before_anything_is_embedded_and_purges_what_it_held() { + let cases = [ + event("$bad", "@discord_123:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), + event("$bad", "@_discord_9:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), + event("$bad", "@discordbot:s", 2, EventKind::Message { body: "SECRET-BODY-TEXT".into() }), + RoomEvent { content_keys: vec!["body".into(), "fi.mau.discord".into()], ..said("$bad", 2, "SECRET-BODY-TEXT") }, + RoomEvent { content_keys: vec!["uk.half-shot.discord".into()], ..said("$bad", 2, "SECRET-BODY-TEXT") }, + ]; + for bad in cases { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$ok", 1, "before"), bad]); + history.room("!other:s", vec![said("$o1", 1, "other room")]); + let mut rig = Rig::new(); + rig.seed("!r:s", "$prev", 0); + let (done, journal) = rig.run(&mut history); + let done = done.unwrap(); + assert!(!rig.embedded().iter().any(|text| text.contains("SECRET-BODY-TEXT") || text.contains("before")), "nothing of the page is embedded"); + assert!(rig.held_ids("!r:s").is_empty() && rig.cursor("!r:s").is_none()); + assert_eq!(rig.held_ids("!other:s"), ["$o1"]); + assert_eq!(done.refused.len(), 1); + assert_eq!((done.refused[0].room_id.as_str(), done.refused[0].event_id.as_deref()), ("!r:s", Some("$bad"))); + assert!(!done.is_clean()); + assert!(journal.iter().all(|line| !line.contains("SECRET-BODY-TEXT")), "{journal:?}"); + } +} + +#[test] +fn a_lookalike_of_the_bridged_platform_is_kept() { + let mut history = FakeHistory::default(); + let kept = ["@discourse:s", "@disco:s", "@eureka:s"].map(|sender| event(&format!("${}", &sender[1..6]), sender, 1, EventKind::Message { body: "hi".into() })); + let mut keyed = said("$k", 2, "hi"); + keyed.content_keys.push("discourse".into()); + let mut events = kept.to_vec(); + events.push(keyed); + history.room("!r:s", events); + let mut rig = Rig::new(); + assert!(rig.ok(&mut history).is_clean()); + assert_eq!(rig.held_ids("!r:s").len(), 4); +} + +#[test] +fn the_collection_is_never_created_by_chat() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 1, "x")]); + for described in [None, Some(DescribedSources::Absent), Some(DescribedSources::Unlabelled)] { + let mut rig = Rig::new(); + match described { + None => rig.index.state.lock().unwrap().described = None, + Some(described) => rig.index.plant(described), + } + let error = format!("{:#}", rig.run(&mut history).0.unwrap_err()); + assert!(error.contains("forge refresh creates it"), "{error}"); + assert_eq!(rig.index.state.lock().unwrap().recreated, 0); + assert!(rig.points().is_empty() && rig.embedded().is_empty()); + } +} + +// ---- redactions, edits, replays ---- + +#[test] +fn a_redacted_event_is_absent_after_the_next_refresh_and_a_first_or_only_message_changes_or_drops_its_chunk() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta"), said("$c", 30, "gamma")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + let first_id = chunk_id("!r:s", "$a"); + assert_eq!(rig.ids(), [first_id.clone()].into()); + history.push("!r:s", vec![redaction("$r1", 40, "$b")]); + rig.ok(&mut history); + assert_eq!(rig.held_ids("!r:s"), ["$a", "$c"]); + assert!(!rig.points()[0].payload.text.contains("beta")); + history.push("!r:s", vec![redaction("$r2", 50, "$a")]); + rig.ok(&mut history); + assert_eq!(rig.ids(), [chunk_id("!r:s", "$c")].into(), "the first message left, so the id changed and the old one is gone"); + history.push("!r:s", vec![redaction("$r3", 60, "$c")]); + rig.ok(&mut history); + assert!(rig.points().is_empty(), "a chunk the redaction empties is deleted"); +} + +#[test] +fn a_redaction_in_the_same_page_as_its_message_never_reaches_the_embedder() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 10, "keep me"), said("$gone", 20, "REDACTED-BODY"), redaction("$r", 30, "$gone")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + assert!(rig.embedded().iter().all(|text| !text.contains("REDACTED-BODY"))); + assert_eq!(rig.held_ids("!r:s"), ["$a"]); +} + +#[test] +fn an_edit_replaces_the_body_in_place_and_a_stranger_an_unknown_target_or_an_edit_of_an_edit_change_nothing() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 10, "old words"), said("$b", 20, "other")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + let before = rig.points(); + history.push("!r:s", vec![edit("$e1", "@a:s", 30, "$a", "new words")]); + rig.ok(&mut history); + let after = rig.points(); + assert_eq!(after.len(), 1); + let first = &after[0].payload.messages[0]; + assert_eq!((first.event_id.as_str(), first.sender.as_str(), first.ts_ms, first.body.as_str()), ("$a", "@a:s", 10, "new words")); + assert_eq!(after[0].id, before[0].id); + assert!(after[0].payload.text.contains("new words") && !after[0].payload.text.contains("old words")); + assert!(rig.embedded().last().unwrap().contains("new words")); + let calls = rig.embedder.state.lock().unwrap().embedded.len(); + history.push("!r:s", vec![edit("$e2", "@stranger:s", 40, "$a", "hijack"), edit("$e3", "@a:s", 41, "$nothing", "lost"), edit("$e4", "@a:s", 42, "$e1", "edit of an edit")]); + rig.ok(&mut history); + assert_eq!(rig.points(), after); + assert_eq!(rig.embedder.state.lock().unwrap().embedded.len(), calls, "no embed call for edits that change nothing"); + assert_eq!(rig.cursor("!r:s").unwrap(), format!("{}:$e4", chat_version()), "the cursor is the last event of any kind"); +} + +#[test] +fn replay_after_a_crash_between_the_upsert_and_the_cursor_neither_duplicates_nor_loses() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta"), edit("$e", "@a:s", 25, "$a", "alpha two"), said("$c", 30, "gamma")]); + let mut rig = Rig::new(); + // One cursor write (the empty start) succeeds, then the page's upsert is acknowledged and the next write fails. + rig.index.state.lock().unwrap().fail_cursor_writes_after = Some(1); + assert!(!rig.run(&mut history).0.unwrap().failed.is_empty()); + let (ids, held) = (rig.ids(), rig.held_ids("!r:s")); + assert_eq!(held, ["$a", "$b", "$c"]); + rig.index.state.lock().unwrap().fail_cursor_writes_after = None; + let done = rig.ok(&mut history); + assert!(done.is_clean()); + assert_eq!((rig.ids(), rig.held_ids("!r:s")), (ids, held)); + assert!(rig.bodies().contains(&"alpha two".to_owned())); + assert_eq!(rig.cursor("!r:s").unwrap(), format!("{}:$c", chat_version())); +} + +#[test] +fn a_failed_upsert_leaves_the_cursor_and_a_failed_embed_or_delete_deletes_nothing_the_index_held() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + let (cursor, points) = (rig.cursor("!r:s"), rig.points()); + history.push("!r:s", vec![said("$c", 30, "gamma")]); + rig.index.state.lock().unwrap().fail_chat_upserts = true; + assert_eq!(rig.run(&mut history).0.unwrap().failed, ["!r:s"]); + assert_eq!((rig.cursor("!r:s"), rig.points()), (cursor.clone(), points.clone())); + rig.index.state.lock().unwrap().fail_chat_upserts = false; + rig.embedder.state.lock().unwrap().embed_fails = true; + assert_eq!(rig.run(&mut history).0.unwrap().failed, ["!r:s"]); + assert_eq!((rig.cursor("!r:s"), rig.points()), (cursor.clone(), points.clone())); + rig.embedder.state.lock().unwrap().embed_fails = false; + // A redaction of the first message writes a new chunk, then fails to delete the old one: the cursor stays. + history.push("!r:s", vec![redaction("$r", 40, "$a")]); + rig.index.state.lock().unwrap().fail_chat_deletes = true; + assert_eq!(rig.run(&mut history).0.unwrap().failed, ["!r:s"]); + assert_eq!(rig.cursor("!r:s"), cursor); + rig.index.state.lock().unwrap().fail_chat_deletes = false; + assert!(rig.ok(&mut history).is_clean()); + assert_eq!(rig.held_ids("!r:s"), ["$b", "$c"]); +} + +// ---- chunking through the job ---- + +#[test] +fn appending_to_the_tail_over_many_runs_equals_one_run_over_all() { + let all: Vec = (0..80).map(|n| said(&format!("$m{n:03}"), 1_000 * n + if n >= 40 { 5_000_000 } else { 0 }, &format!("message number {n} {}", "x".repeat((n as usize * 37) % 400)))).collect(); + let mut whole_history = FakeHistory::default(); + whole_history.room("!r:s", all.clone()); + let mut whole = Rig::new(); + whole.ok(&mut whole_history); + let mut grown_history = FakeHistory::default(); + grown_history.room("!r:s", vec![]); + let mut grown = Rig::new(); + for batch in all.chunks(7) { + grown_history.push("!r:s", batch.to_vec()); + grown.ok(&mut grown_history); + } + assert!(whole.points().len() > 2); + let shape = |rig: &Rig| rig.points().into_iter().map(|point| (point.id, point.payload)).collect::>(); + assert_eq!(shape(&grown), shape(&whole)); +} + +#[test] +fn a_long_message_is_cut_with_a_marker_and_every_point_satisfies_the_payload_contract() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 10, &"é".repeat(CHAT_BODY_CHARS_MAX * 2)), said("$b", 20, "short"), said("$c", 99_000_000, "later")]); + history.facts("!r:s", |facts| { + facts.alias = Some("#swarm:gamecult.org".into()); + facts.name = Some("Swarm".into()); + }); + let mut rig = Rig::new(); + rig.ok(&mut history); + let points = rig.points(); + assert_eq!(points.len(), 2); + for point in &points { + let payload = &point.payload; + assert_eq!((payload.room_id.as_str(), payload.room_alias.as_deref()), ("!r:s", Some("#swarm:gamecult.org"))); + assert_eq!(point.id, chunk_id("!r:s", &payload.messages[0].event_id)); + assert_eq!(payload.text, super::chat_chunk::embedded_text("Swarm", &payload.messages)); + assert!(!payload.messages.is_empty()); + } + let long = &points.iter().find(|point| point.payload.messages[0].event_id == "$a").unwrap().payload.messages[0]; + assert_eq!(long.body.chars().count(), CHAT_BODY_CHARS_MAX); + assert!(long.body.ends_with("[cut]")); +} + +#[test] +fn a_lost_cursor_event_or_a_version_bump_re_ingests_that_room_alone() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 10, "alpha"), said("$b", 20, "beta")]).room("!q:s", vec![said("$q1", 10, "quiet")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + let q_ids: BTreeSet = rig.points().iter().filter(|point| point.payload.room_id == "!q:s").map(|point| point.id.clone()).collect(); + let recreated = rig.index.state.lock().unwrap().recreated; + for bump in [Some("0.0".to_owned()), None] { + let mut cursors = rig.index.cursors(); + let value = match &bump { + Some(old) => format!("{old}:$b"), + None => format!("{}:$gone-from-the-server", chat_version()), + }; + cursors.insert(cursor_key("!r:s"), value); + rig.index.set_cursors(&cursors).unwrap(); + let before = rig.embedded().len(); + assert!(rig.ok(&mut history).is_clean()); + assert_eq!(rig.held_ids("!r:s"), ["$a", "$b"]); + assert!(rig.embedded().len() > before, "the room was re-embedded"); + assert_eq!(rig.points().iter().filter(|point| point.payload.room_id == "!q:s").map(|point| point.id.clone()).collect::>(), q_ids); + assert_eq!(rig.cursor("!r:s").unwrap(), format!("{}:$b", chat_version())); + } + assert_eq!(rig.index.state.lock().unwrap().recreated, recreated); +} + +#[test] +fn retire_chat_removes_rooms_no_longer_kept_and_given_none_all_chat_and_touches_nothing_else() { + let mut rig = Rig::new(); + rig.index.upsert_sources(&[SourcePoint { + id: "forge-1".into(), + vector: vec![1.0; 4], + payload: SourcePayload { repo: "Repo".into(), path: "a".into(), commit: "c".into(), blob: "b".into(), line_start: 1, line_end: 1, chunk_index: 0, ext: "rs".into(), text: "t".into() }, + }]) + .unwrap(); + for room in ["!a:s", "!b:s"] { + rig.seed(room, "$e", 1); + } + let mut cursors = rig.index.cursors(); + cursors.insert("Repo".into(), "c".into()); + cursors.insert(format!("{CHAT_PURGE_PREFIX}!a:s"), "1..2".into()); + rig.index.set_cursors(&cursors).unwrap(); + let retired = retire_chat(&mut rig.index, &["!a:s".to_owned()].into()).unwrap(); + assert_eq!(retired, ["!b:s"]); + assert_eq!(rig.held_ids("!a:s"), ["$e"]); + assert!(rig.held_ids("!b:s").is_empty() && rig.cursor("!b:s").is_none()); + assert_eq!(retire_chat(&mut rig.index, &BTreeSet::new()).unwrap(), ["!a:s"]); + assert!(rig.points().is_empty()); + assert_eq!(rig.index.state.lock().unwrap().points.len(), 1, "a forge point stays"); + let kept: Vec = rig.index.cursors().into_keys().collect(); + assert_eq!(kept, ["Repo".to_owned(), format!("{CHAT_PURGE_PREFIX}!a:s")]); + // A room the reader left is retired by the refresh itself. + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 1, "x")]); + rig.seed("!gone:s", "$e", 1); + rig.ok(&mut history); + assert!(rig.held_ids("!gone:s").is_empty()); +} + +// ---- the credential scrub through the job ---- + +fn planted() -> Vec<(ScrubKind, String)> { + vec![ + (ScrubKind::AuthorizationHeader, format!("Authorization: Bearer {}", "a1B2".repeat(8))), + (ScrubKind::Bearer, format!("Bearer {}", "z9Y8".repeat(6))), + (ScrubKind::KeyShape, format!("sk-{}", "q7W3".repeat(8))), + (ScrubKind::KeyShape, format!("sk_live_{}", "m4N5".repeat(6))), + (ScrubKind::KeyShape, format!("ghp_{}", "k2L9".repeat(9))), + (ScrubKind::KeyShape, format!("github_pat_{}_{}", "A1b2C3".repeat(4) + "ab", "d4E5f6".repeat(9) + "abcde")), + (ScrubKind::KeyShape, format!("glpat-{}", "r6T7".repeat(5))), + (ScrubKind::KeyShape, "xoxb-123456789012-987654321098-aBcDeFgHiJkLmNoP".into()), + (ScrubKind::KeyShape, "AKIA0123456789ABCDEF".into()), + (ScrubKind::KeyShape, format!("AIza{}", "Sy1_".repeat(9))), + (ScrubKind::Jwt, "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJmYWtlIn0.c2lnbmF0dXJlZmFrZQ".into()), + (ScrubKind::PemBlock, "-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEAfake\n-----END RSA PRIVATE KEY-----".into()), + (ScrubKind::PemBlock, "-----BEGIN PRIVATE KEY-----\nMIIBVQIBADANfakeNOEND".into()), + (ScrubKind::UrlUserinfo, "https://user:hunter2hunter2@host.example/r.git".into()), + (ScrubKind::UrlUserinfo, "https://tokenvalue12345@host.example/".into()), + ] +} + +/// The part of a planted secret no replacement can contain. +fn distinct(secret: &str) -> String { + let bare = secret.rsplit(['@', ' ', '\n']).next().filter(|_| !secret.contains("://")).unwrap_or(secret); + let source = if secret.contains("://") { secret.split("://").nth(1).unwrap().split('@').next().unwrap().rsplit(':').next().unwrap() } else { bare }; + source.chars().rev().take(12).collect::>().into_iter().rev().collect() +} + +#[test] +fn planted_credentials_never_reach_the_chunk_the_embedder_or_the_journal_in_a_message_or_an_edit() { + for (kind, secret) in planted() { + let needle = distinct(&secret); + assert!(needle.len() >= 8, "{secret}"); + for shape in ["see {} now", "see {}", "{}\nnext line"] { + let (dirty, clean) = (shape.replace("{}", &secret), "nothing here".to_owned()); + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$m", 10, &dirty), said("$e", 20, &clean), edit("$ed", "@a:s", 30, "$e", &dirty)]); + let mut rig = Rig::new(); + let (done, journal) = rig.run(&mut history); + let done = done.unwrap(); + assert!(done.is_clean(), "a scrub hit does not make a run unclean"); + let stored = format!("{:?}", rig.points().iter().map(|point| (&point.payload.messages, &point.payload.text)).collect::>()); + for seen in [stored, rig.embedded().join("\n"), journal.join("\n")] { + assert!(!seen.contains(&needle), "{kind}: {secret:?} leaked into {seen}"); + } + assert!(rig.bodies().iter().filter(|body| body.contains("")).count() >= 2, "{kind}: both bodies show the replacement"); + let hits: Vec<(&str, ScrubKind)> = done.scrubbed.iter().map(|hit| (hit.event_id.as_str(), hit.kind)).collect(); + assert!(hits.contains(&("$m", kind)) && hits.contains(&("$ed", kind)), "{kind}: {hits:?}"); + assert!(done.scrubbed.iter().all(|hit| hit.room_id == "!r:s")); + assert!(journal.contains(&format!("scrubbed: !r:s $m {kind}")), "{journal:?}"); + } + } +} + +#[test] +fn the_scrub_meets_edits_redactions_the_body_cut_replays_and_version_bumps() { + let token = format!("ghp_{}", "k2L9".repeat(9)); + let mut history = FakeHistory::default(); + history.room( + "!r:s", + vec![ + said("$clean", 10, "all clean"), + said("$dirty", 20, &format!("token {token} inside")), + edit("$adds", "@a:s", 30, "$clean", &format!("now with {token}")), + edit("$removes", "@a:s", 40, "$dirty", "token removed"), + said("$gone", 50, &format!("secret {token}")), + redaction("$r", 60, "$gone"), + said("$straddle", 70, &format!("{} {token} {}", "x".repeat(CHAT_BODY_CHARS_MAX - 10), "y".repeat(100))), + ], + ); + let mut rig = Rig::new(); + let done = rig.ok(&mut history); + assert!(rig.bodies().iter().all(|body| !body.contains("k2L9k2L9"))); + assert!(rig.embedded().iter().all(|text| !text.contains("k2L9k2L9"))); + let by_id: BTreeMap = rig.points().iter().flat_map(|point| point.payload.messages.iter().map(|message| (message.event_id.clone(), message.body.clone()))).collect(); + assert_eq!(by_id["$clean"], "now with "); + assert_eq!(by_id["$dirty"], "token removed"); + assert!(!by_id.contains_key("$gone")); + assert!(by_id["$straddle"].chars().count() <= CHAT_BODY_CHARS_MAX && !by_id["$straddle"].contains("ghp_")); + assert!(done.scrubbed.iter().any(|hit| hit.event_id == "$adds") && done.scrubbed.iter().any(|hit| hit.event_id == "$gone")); + let (ids, texts) = (rig.ids(), rig.points().into_iter().map(|point| point.payload.text).collect::>()); + // Replay from the start, as after a lost cursor: identical ids and text. + let mut cursors = rig.index.cursors(); + cursors.remove(&cursor_key("!r:s")); + rig.index.set_cursors(&cursors).unwrap(); + rig.ok(&mut history); + assert_eq!((rig.ids(), rig.points().into_iter().map(|point| point.payload.text).collect::>()), (ids, texts)); + // An older scrub version re-ingests the room and scrubs it again from the history. + let mut cursors = rig.index.cursors(); + cursors.insert(cursor_key("!r:s"), format!("{}.0:$straddle", huginn_mind::redact::SCRUB_VERSION - 1)); + rig.index.set_cursors(&cursors).unwrap(); + rig.ok(&mut history); + assert!(rig.bodies().iter().all(|body| !body.contains("k2L9k2L9"))); +} + +#[test] +fn ordinary_text_is_stored_byte_for_byte() { + let mut history = FakeHistory::default(); + let same = "risk-averse task-force sk- 3f2b8c1d9e7a4b5c6d0e1f2a3b4c5d6e7f8a9b0c !abc:s user@example.com ssh://git@host/r.git the bearer of bad news"; + history.room("!r:s", vec![said("$a", 1, same)]); + let mut rig = Rig::new(); + let done = rig.ok(&mut history); + assert_eq!(rig.bodies(), [same]); + assert!(done.scrubbed.is_empty()); +} + +// ---- steward purge ---- + +fn window(from_ms: u64, to_ms: u64) -> PurgeWindow { + PurgeWindow { from_ms, to_ms } +} + +fn purge_history() -> FakeHistory { + let mut history = FakeHistory::default(); + // t1 = 1000, t2 = 3000: $b at t1 and $c inside are purged, $d at t2 is kept. + history.room("!r:s", vec![said("$a", 500, "before"), said("$b", 1000, "PURGED-B"), said("$c", 2000, "PURGED-C"), said("$d", 3000, "after"), said("$e", 4000, "later")]); + history.room("!other:s", vec![said("$o1", 1000, "same time other room"), said("$o2", 2000, "same time")]); + history +} + +#[test] +fn a_purge_window_removes_exactly_its_events_and_re_chunks_the_neighbours() { + let mut history = purge_history(); + let mut rig = Rig::new(); + rig.ok(&mut history); + let other_ids: BTreeSet = rig.points().iter().filter(|point| point.payload.room_id == "!other:s").map(|point| point.id.clone()).collect(); + assert_eq!(rig.held_ids("!r:s"), ["$a", "$b", "$c", "$d", "$e"]); + assert!(request_purge(&mut rig.index, "!r:s", window(1000, 3000)).unwrap()); + let embedded_before = rig.embedded().len(); + rig.ok(&mut history); + assert_eq!(rig.held_ids("!r:s"), ["$a", "$d", "$e"], "t1 is inside, t2 is not"); + assert!(rig.embedded()[embedded_before..].iter().all(|text| !text.contains("PURGED"))); + assert!(rig.points().iter().all(|point| !point.payload.text.contains("PURGED"))); + assert_eq!(rig.points().iter().filter(|point| point.payload.room_id == "!other:s").map(|point| point.id.clone()).collect::>(), other_ids); + assert_eq!(rig.held_ids("!other:s"), ["$o1", "$o2"]); + assert_eq!(tombstone(&rig, "!r:s").as_deref(), Some("1000..3000")); +} + +#[test] +fn a_purged_first_message_changes_the_chunk_id_and_an_emptied_chunk_is_deleted() { + let mut history = FakeHistory::default(); + history.room("!r:s", vec![said("$a", 1000, "x"), said("$b", 2000, "y"), said("$c", 99_000_000, "far")]); + let mut rig = Rig::new(); + rig.ok(&mut history); + assert_eq!(rig.ids().len(), 2); + request_purge(&mut rig.index, "!r:s", window(900, 1500)).unwrap(); + rig.ok(&mut history); + assert_eq!(rig.ids(), [chunk_id("!r:s", "$b"), chunk_id("!r:s", "$c")].into()); + request_purge(&mut rig.index, "!r:s", window(99_000_000, 99_000_001)).unwrap(); + rig.ok(&mut history); + assert_eq!(rig.held_ids("!r:s"), ["$b"]); +} + +#[test] +fn a_purged_window_does_not_come_back() { + let mut history = purge_history(); + let mut rig = Rig::new(); + rig.ok(&mut history); + request_purge(&mut rig.index, "!r:s", window(1000, 3000)).unwrap(); + rig.ok(&mut history); + let want = ["$a", "$d", "$e"]; + // New events, an in-window edit of an outside message, a redaction. + history.push("!r:s", vec![said("$f", 5000, "new"), edit("$ed", "@a:s", 2500, "$a", "EDIT-IN-WINDOW"), redaction("$rd", 6000, "$e")]); + rig.ok(&mut history); + assert_eq!(rig.held_ids("!r:s"), ["$a", "$d", "$f"]); + assert!(rig.bodies().contains(&"before".to_owned()) && !rig.bodies().iter().any(|body| body.contains("EDIT-IN-WINDOW")), "the in-window edit is dropped and the original stays"); + // A lost cursor re-ingests from the start; a version bump re-ingests too. + for cursor in [None, Some("0.0:$f".to_owned())] { + let mut cursors = rig.index.cursors(); + match cursor { + None => cursors.remove(&cursor_key("!r:s")), + Some(old) => cursors.insert(cursor_key("!r:s"), old), + }; + rig.index.set_cursors(&cursors).unwrap(); + rig.ok(&mut history); + assert!(rig.bodies().iter().all(|body| !body.contains("PURGED")), "{:?}", rig.held_ids("!r:s")); + assert_eq!(rig.held_ids("!r:s"), ["$a", "$d", "$f"], "$e stays redacted by the history; {want:?} before the new events"); + } + // The reader leaves the room and joins again. + let mut left = FakeHistory::default(); + left.room("!other:s", vec![said("$o1", 1000, "x")]); + rig.ok(&mut left); + assert!(rig.held_ids("!r:s").is_empty() && tombstone(&rig, "!r:s").is_some()); + rig.ok(&mut history); + assert!(rig.bodies().iter().all(|body| !body.contains("PURGED"))); +} + +#[test] +fn a_purge_of_an_unindexed_window_writes_only_the_tombstone_and_still_holds() { + let mut rig = Rig::new(); + let writes = rig.index.state.lock().unwrap().cursor_writes.len(); + assert!(request_purge(&mut rig.index, "!never:s", window(10, 20)).unwrap()); + assert_eq!(rig.index.state.lock().unwrap().cursor_writes.len(), writes + 1); + assert!(rig.points().is_empty() && rig.embedded().is_empty()); + let mut history = FakeHistory::default(); + history.room("!never:s", vec![said("$a", 5, "ok"), said("$b", 15, "PURGED"), said("$c", 25, "ok")]); + rig.ok(&mut history); + assert_eq!(rig.held_ids("!never:s"), ["$a", "$c"]); + assert!(rig.embedded().iter().all(|text| !text.contains("PURGED"))); +} + +#[test] +fn a_repeated_or_overlapping_request_changes_nothing_or_merges_and_the_33rd_window_is_refused() { + let mut rig = Rig::new(); + assert!(request_purge(&mut rig.index, "!r:s", window(10, 20)).unwrap()); + let writes = rig.index.state.lock().unwrap().cursor_writes.len(); + assert!(!request_purge(&mut rig.index, "!r:s", window(10, 20)).unwrap()); + assert!(!request_purge(&mut rig.index, "!r:s", window(12, 18)).unwrap()); + assert_eq!(rig.index.state.lock().unwrap().cursor_writes.len(), writes, "no write for a request the tombstone already holds"); + assert!(request_purge(&mut rig.index, "!r:s", window(15, 30)).unwrap()); + assert_eq!(tombstone(&rig, "!r:s").as_deref(), Some("10..30")); + for n in 0..(PURGE_WINDOWS_MAX as u64 - 1) { + request_purge(&mut rig.index, "!r:s", window(100 + n * 10, 105 + n * 10)).unwrap(); + } + let error = format!("{:#}", request_purge(&mut rig.index, "!r:s", window(10_000, 10_001)).unwrap_err()); + assert!(error.contains("32"), "{error}"); + assert!(request_purge(&mut Rig::new().index, "!r:s", window(5, 5)).is_err()); +} + +// ---- the source stays in its lane ---- + +#[test] +fn the_ingest_code_is_pure_over_its_ports_and_names_no_credential_shape_no_second_literal_and_no_join_rule() { + let sources = [("chat.rs", include_str!("chat.rs")), ("chat_chunk.rs", include_str!("chat_chunk.rs")), ("chat_purge.rs", include_str!("chat_purge.rs")), ("room_scope.rs", include_str!("room_scope.rs"))]; + let process = ["Command", "::new"].concat(); + for (name, source) in sources { + let code = source.split("#[cfg(test)]").next().unwrap(); + assert!(!code.contains(&process) && !code.contains("std::process"), "{name}"); + assert!(!code.to_ascii_lowercase().contains("join_rule"), "{name}"); + if name != "room_scope.rs" { + assert!(!code.to_ascii_lowercase().contains("discord"), "{name}"); + } + for shape in ["sk-", "sk_live_", "ghp_", "github_pat_", "glpat-", "AKIA", "xoxb-", "PRIVATE KEY", "Bearer"] { + assert!(!code.contains(shape), "{name} names {shape}"); + } + let defined = usize::from(name == "chat.rs"); + assert_eq!((code.matches("\"chat:").count(), code.matches("\"chat-purge:").count()), (defined, defined), "{name}: each prefix is defined once"); + } + let chat = include_str!("chat.rs"); + assert_eq!(chat.matches("scrub(").count(), 1, "scrub is called in one place"); +} diff --git a/crates/huginn-daemon/src/sources/mod.rs b/crates/huginn-daemon/src/sources/mod.rs index 92bae2a..5e4133d 100644 --- a/crates/huginn-daemon/src/sources/mod.rs +++ b/crates/huginn-daemon/src/sources/mod.rs @@ -3,29 +3,37 @@ //! for semantic search. Each point's `corpus` payload (`forge` or `chat`) keeps //! the two apart; a forge search filters `forge` and a chat search `chat`. //! -//! Two truths, one writer. The forge is the truth of which repos exist and what -//! they hold; the collection is derived from it and can be deleted whenever. -//! `refresh` is the only writer of the collection, run by the `huginn-sources` -//! oneshot job: first build, incremental refresh, rebuild after a model or -//! chunker change, and retirement are all the one routine (delete by filter, -//! ingest at a commit, rewrite the cursors). The daemon only reads: `search` -//! is its whole part, and nothing in `serve` or the worker calls a write. +//! Two truths, one job. The forge is the truth of which repos exist and what +//! they hold, Synapse of what the Minds' rooms said; the collection is derived +//! from them and can be deleted whenever (but for a steward's purge tombstones, see +//! `chat_purge`). `refresh` is the forge's writer and the only creator of the +//! collection, run by the `huginn-sources` oneshot job: first build, incremental +//! refresh, rebuild after a model or chunker change, and retirement are all the one +//! routine (delete by filter, ingest at a commit, rewrite the cursors). `chat::refresh_chat` +//! joins it in the same job and the same cursors object, as the only writer of chat points; +//! it requires the collection to exist and never recreates it. The daemon only reads: `search` +//! and `search_chat` are its whole part, and nothing in `serve` or the worker calls a write. //! -//! A cursor is the commit whose chunks the collection holds for a repo. It -//! lives in the collection's own metadata, so it is lost with the collection +//! A cursor is the commit (or, for a room, the event) whose chunks the collection +//! holds. It lives in the collection's own metadata, so it is lost with the collection //! and the two cannot disagree about what is held. It is advanced only after -//! the repo's deletes and upserts are acknowledged: it may lag and never leads. -//! It is dropped before the repo's first delete, so a run that fails between -//! the delete and the upserts leaves no cursor, and the repo is ingested whole. -//! The cursors object is written whole by this job alone, once per repo. +//! the source's deletes and upserts are acknowledged: it may lag and never leads. +//! A repo's is dropped before its first delete, so a run that fails between +//! the delete and the upserts leaves no cursor, and the repo is ingested whole; a room's +//! chat chunks are written before anything they replace is deleted. The cursors +//! object is written whole by this job alone. A key is a repo name, or a chat key +//! (`chat::is_chat_key`), never both. pub mod chat; pub mod chat_chunk; +pub mod chat_purge; pub mod chunk; pub mod forge; pub mod git; pub mod room_scope; +#[cfg(test)] +mod chat_tests; #[cfg(test)] mod tests;