use std::collections::{HashMap, HashSet};
use std::fmt;
use std::ops::{Bound, Range, RangeBounds};
use std::str::FromStr;
use std::sync::Arc;
use serde::{Deserialize, Serialize, Serializer};
use crate::completion::Message;
use crate::message::{StopReason, ToolCall, UserContent};
mod aside;
mod author;
mod lineage;
mod note;
pub mod pairing;
mod turn;
pub use aside::{Aside, AsideId};
pub use author::{Author, AuthorKind, RequestId};
pub use lineage::Lineage;
pub use note::Note;
pub use turn::{TURN, TurnEnd, TurnRecord};
use pairing::{Normalized, PairingError};
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
#[serde(try_from = "State")]
pub struct Conversation {
messages: Vec<Arc<Message>>,
meta: Vec<Meta>,
#[serde(serialize_with = "saved_asides")]
asides: Vec<(AsideId, Aside)>,
#[serde(skip_serializing_if = "Option::is_none")]
halted: Option<Halt>,
next_id: u64,
next_aside: u64,
#[serde(skip)]
revision: u64,
#[serde(skip)]
lineage: Lineage,
#[serde(skip)]
unsaved: Vec<ChangeRecord>,
#[serde(skip)]
clock: Option<u64>,
}
#[derive(Clone, Debug, Serialize, Deserialize)]
struct Meta {
id: MessageId,
#[serde(default, skip_serializing_if = "Option::is_none")]
author: Option<Author>,
#[serde(default)]
made: u64,
}
#[derive(Deserialize)]
struct State {
#[serde(default)]
messages: Vec<Arc<Message>>,
#[serde(default)]
meta: Vec<Meta>,
#[serde(default)]
asides: Vec<(AsideId, Aside)>,
#[serde(default)]
halted: Option<Halt>,
#[serde(default)]
next_id: u64,
#[serde(default)]
next_aside: u64,
}
impl TryFrom<State> for Conversation {
type Error = ApplyError;
fn try_from(state: State) -> Result<Self, ApplyError> {
if state.meta.len() != state.messages.len() {
return Err(mismatch(format!(
"{} messages but {} ids",
state.messages.len(),
state.meta.len()
)));
}
let mut ids = HashSet::new();
if let Some(meta) = state
.meta
.iter()
.find(|meta| meta.id.0 >= state.next_id || !ids.insert(meta.id))
{
return Err(mismatch(format!("message id {} reused", meta.id.0)));
}
let mut asides = HashSet::new();
if let Some((id, _)) = state
.asides
.iter()
.find(|(id, _)| id.0 >= state.next_aside || !asides.insert(*id))
{
return Err(mismatch(format!("aside id {} reused", id.0)));
}
Ok(Self {
messages: state.messages,
meta: state.meta,
asides: state.asides,
halted: state.halted,
next_id: state.next_id,
next_aside: state.next_aside,
revision: 0,
lineage: Lineage::DETACHED,
unsaved: Vec::new(),
clock: None,
})
}
}
fn saved_asides<S: Serializer>(
asides: &[(AsideId, Aside)],
serializer: S,
) -> Result<S::Ok, S::Error> {
serializer.collect_seq(asides.iter().filter(|(_, aside)| aside.saved))
}
impl PartialEq for Conversation {
fn eq(&self, other: &Self) -> bool {
let saved = |(_, aside): &&(AsideId, Aside)| aside.saved;
let same = |a: &Meta, b: &Meta| a.id == b.id && a.author == b.author;
self.messages == other.messages
&& self.meta.len() == other.meta.len()
&& self.meta.iter().zip(&other.meta).all(|(a, b)| same(a, b))
&& self.halted == other.halted
&& self
.asides
.iter()
.filter(saved)
.eq(other.asides.iter().filter(saved))
}
}
#[derive(
Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize,
)]
#[serde(transparent)]
pub struct MessageId(u64);
impl MessageId {
pub const fn get(self) -> u64 {
self.0
}
}
impl fmt::Display for MessageId {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "m{}", self.0)
}
}
impl FromStr for MessageId {
type Err = ParseIdError;
fn from_str(text: &str) -> Result<Self, ParseIdError> {
parse_id(text, 'm').map(Self)
}
}
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
#[error("{0:?} is not an id")]
pub struct ParseIdError(pub String);
fn parse_id(text: &str, prefix: char) -> Result<u64, ParseIdError> {
let number = text.strip_prefix(prefix).unwrap_or(text);
number.parse().map_err(|_| ParseIdError(text.to_owned()))
}
#[derive(
Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize,
)]
pub struct MessageKey {
pub made: u64,
pub id: MessageId,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Entry {
pub message: Arc<Message>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub author: Option<Author>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub id: Option<MessageId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub made: Option<u64>,
}
impl Entry {
pub fn new(message: impl Into<Arc<Message>>) -> Self {
Self {
message: message.into(),
author: None,
id: None,
made: None,
}
}
#[must_use]
pub fn by(self, author: Author) -> Self {
Self {
author: Some(author),
..self
}
}
}
impl From<Message> for Entry {
fn from(message: Message) -> Self {
Self::new(message)
}
}
impl From<Arc<Message>> for Entry {
fn from(message: Arc<Message>) -> Self {
Self::new(message)
}
}
#[derive(Clone, Copy, Debug)]
pub struct EntryRef<'c> {
pub index: usize,
pub id: MessageId,
pub message: &'c Message,
pub author: Option<&'c Author>,
pub made: u64,
}
impl EntryRef<'_> {
pub fn key(&self) -> MessageKey {
MessageKey {
made: self.made,
id: self.id,
}
}
pub fn is_users_own(&self) -> bool {
let typed = match self.message {
Message::User { content } => content
.iter()
.any(|item| !matches!(item, UserContent::ToolResult(_))),
_ => false,
};
let users = self
.author
.is_none_or(|author| author.kind == AuthorKind::User && author.title.is_none());
typed && users
}
pub fn is_request(&self) -> bool {
let asks = match self.message {
Message::User { content } => content
.iter()
.any(|item| !matches!(item, UserContent::ToolResult(_))),
_ => false,
};
asks && self
.author
.is_none_or(|author| author.kind != AuthorKind::Rig)
}
pub fn is_stop_marker(&self) -> bool {
let by_rig = self
.author
.is_some_and(|author| author.kind == AuthorKind::Rig);
by_rig
&& matches!(self.message, Message::User { content }
if matches!(content.as_slice(), [UserContent::Text(text)] if text.text == STOPPED))
}
}
pub const STOPPED: &str = "[The request above was stopped by the user before it was answered. \
Do not carry it out unless asked again.]";
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Halt {
Stopped,
#[default]
Kept,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum Change {
Push {
entries: Vec<Entry>,
},
Splice {
from: usize,
to: usize,
removed: Vec<MessageId>,
with: Vec<Entry>,
},
Halt {
halt: Option<Halt>,
},
Aside {
id: AsideId,
aside: Option<Aside>,
},
Reset {
state: Box<Conversation>,
},
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ChangeRecord {
#[serde(flatten)]
pub change: Change,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub by: Option<Author>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub at: Option<u64>,
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Spliced {
pub removed: Vec<MessageId>,
pub changed: Vec<MessageId>,
pub restored: Vec<MessageId>,
}
#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
pub enum ApplyError {
#[error("{0}")]
Mismatch(String),
}
fn mismatch(why: String) -> ApplyError {
ApplyError::Mismatch(why)
}
pub fn short_id(id: &str) -> &str {
id.get(..8).unwrap_or(id)
}
impl Conversation {
pub fn new() -> Self {
Self::default()
}
pub fn from_messages(messages: impl IntoIterator<Item = impl Into<Arc<Message>>>) -> Self {
let mut conversation = Self::new();
conversation.extend(messages);
conversation
}
pub fn messages(&self) -> &[Arc<Message>] {
&self.messages
}
pub fn into_messages(self) -> Vec<Arc<Message>> {
self.messages
}
pub fn entries(&self) -> impl DoubleEndedIterator<Item = EntryRef<'_>> + ExactSizeIterator {
let entries = self.messages.iter().zip(&self.meta).enumerate();
entries.map(|(index, (message, meta))| EntryRef {
index,
id: meta.id,
message,
author: meta.author.as_ref(),
made: meta.made,
})
}
pub fn entry(&self, at: usize) -> Option<Entry> {
let meta = self.meta.get(at)?;
Some(Entry {
message: self.messages.get(at)?.clone(),
author: meta.author.clone(),
id: Some(meta.id),
made: Some(meta.made),
})
}
pub fn id(&self, at: usize) -> Option<MessageId> {
self.meta.get(at).map(|meta| meta.id)
}
pub fn key(&self, at: usize) -> Option<MessageKey> {
let meta = self.meta.get(at)?;
Some(MessageKey {
made: meta.made,
id: meta.id,
})
}
pub fn index_of_key(&self, key: MessageKey) -> Option<usize> {
let at = self.index_of(key.id)?;
(self.meta.get(at)?.made == key.made).then_some(at)
}
pub fn author(&self, at: usize) -> Option<&Author> {
self.meta.get(at)?.author.as_ref()
}
pub fn index_of(&self, id: MessageId) -> Option<usize> {
self.meta.iter().position(|meta| meta.id == id)
}
pub fn start_of(&self, id: MessageId) -> usize {
self.index_of(id).unwrap_or(if id.0 >= self.next_id {
self.messages.len()
} else {
0
})
}
pub fn turn_start(&self) -> MessageId {
let trailing = self.messages.iter().rev();
let trailing = trailing
.take_while(|message| matches!(&***message, Message::User { .. }))
.count();
let first = self.messages.len().saturating_sub(trailing);
let answered = (trailing > 0).then(|| self.id(first)).flatten();
answered.unwrap_or(MessageId(self.next_id))
}
pub fn asked(&self, start: MessageId) -> impl Iterator<Item = EntryRef<'_>> {
let entries = self.entries().skip(self.start_of(start));
entries.take_while(|entry| matches!(entry.message, Message::User { .. }))
}
pub fn request(&self, start: MessageId) -> Option<EntryRef<'_>> {
self.asked(start).find(EntryRef::is_request)
}
pub fn halted(&self) -> Option<Halt> {
self.halted
}
pub fn awaits_model(&self) -> bool {
let read = self.messages.iter().rev().map(Arc::as_ref).find(|message| {
!matches!(message, Message::Assistant(turn)
if turn.stop.as_ref().is_some_and(StopReason::is_failure))
});
self.halted.is_none() && matches!(read, Some(Message::User { .. }))
}
pub fn pending_calls(&self) -> Vec<ToolCall> {
pairing::pending_calls(&self.messages)
}
pub fn validate(&self) -> Result<(), PairingError> {
pairing::validate(&self.messages)
}
pub fn asides(&self) -> impl DoubleEndedIterator<Item = (AsideId, &Aside)> {
self.asides.iter().map(|(id, aside)| (*id, aside))
}
pub fn aside(&self, id: AsideId) -> Option<&Aside> {
let mut asides = self.asides.iter();
asides
.find(|(known, _)| *known == id)
.map(|(_, aside)| aside)
}
pub fn aside_position(&self, aside: &Aside) -> usize {
let Some(after) = aside.after else {
return 0;
};
match self.index_of(after) {
Some(index) => index.saturating_add(1),
None => self
.messages
.len()
.min(self.start_of(after).saturating_add(1)),
}
}
pub fn push(&mut self, entry: impl Into<Entry>) -> MessageId {
let entry = entry.into();
let by = entry.author.clone();
let mut ids = self.push_entries(vec![entry], by);
ids.pop().unwrap_or(MessageId(self.next_id))
}
pub fn extend(&mut self, messages: impl IntoIterator<Item = impl Into<Arc<Message>>>) {
let entries = messages.into_iter().map(Entry::new).collect();
self.push_entries(entries, None);
}
fn push_entries(&mut self, entries: Vec<Entry>, by: Option<Author>) -> Vec<MessageId> {
if entries.is_empty() {
return Vec::new();
}
let stop = (self.halted == Some(Halt::Stopped))
.then(|| Entry::new(Message::user(STOPPED)).by(Author::rig()));
let mut next = self.next_id;
let mut ids = Vec::with_capacity(entries.len());
let mut fill = |mut entry: Entry| {
let id = MessageId(next);
next = next.saturating_add(1);
entry.id = Some(id);
entry.made = None;
(entry, id)
};
let stop = stop.map(|entry| fill(entry).0);
let entries = entries.into_iter().map(|entry| {
let (entry, id) = fill(entry);
ids.push(id);
entry
});
let entries = stop.into_iter().chain(entries).collect();
self.record(Change::Push { entries }, by);
self.clear_halt();
ids
}
pub fn splice(
&mut self,
range: impl RangeBounds<usize>,
with: impl IntoIterator<Item = Entry>,
by: Author,
) -> Spliced {
self.replace(range, with.into_iter().collect(), by, false)
}
pub fn restore(
&mut self,
range: impl RangeBounds<usize>,
with: impl IntoIterator<Item = Entry>,
by: Author,
) -> Spliced {
self.replace(range, with.into_iter().collect(), by, true)
}
pub fn set_entries(&mut self, entries: Vec<Entry>, by: Author) -> Spliced {
let same = |entry: &Entry, at: usize| {
entry.id.is_some()
&& entry.id == self.id(at)
&& entry.author.as_ref() == self.author(at)
&& Some(&entry.message) == self.messages.get(at)
};
let (len, count) = (self.messages.len(), entries.len());
let start = entries
.iter()
.enumerate()
.take_while(|(at, entry)| same(entry, *at))
.count();
let end = entries
.iter()
.rev()
.zip((0..len).rev())
.take(len.min(count).saturating_sub(start))
.take_while(|(entry, at)| same(entry, *at))
.count();
if start == len && start == count {
return Spliced::default();
}
let with = entries
.into_iter()
.skip(start)
.take(count.saturating_sub(start).saturating_sub(end));
self.restore(start..len.saturating_sub(end), with, by)
}
fn replace(
&mut self,
range: impl RangeBounds<usize>,
mut with: Vec<Entry>,
by: Author,
restores: bool,
) -> Spliced {
let len = self.messages.len();
let to = match range.end_bound() {
Bound::Included(&end) => end.saturating_add(1),
Bound::Excluded(&end) => end,
Bound::Unbounded => len,
}
.min(len);
let from = match range.start_bound() {
Bound::Included(&start) => start,
Bound::Excluded(&start) => start.saturating_add(1),
Bound::Unbounded => 0,
}
.min(to);
if from == to && with.is_empty() {
return Spliced::default();
}
let replaced = self.meta.get(from..to).unwrap_or_default();
let removed: Vec<MessageId> = replaced.iter().map(|meta| meta.id).collect();
let in_range: HashSet<MessageId> = removed.iter().copied().collect();
let mut outside = None;
let mut taken = HashSet::new();
let mut restored = Vec::new();
let mut next = self.next_id;
for entry in &mut with {
let id = entry.id.filter(|id| !taken.contains(id));
if let Some(id) = id.filter(|id| in_range.contains(id)) {
taken.insert(id);
entry.made = None;
continue;
}
let back = id.filter(|id| {
restores
&& id.0 < self.next_id
&& !self.outside(from..to, &mut outside).contains(id)
});
if let Some(id) = back {
taken.insert(id);
restored.push(id);
continue;
}
entry.id = Some(MessageId(next));
entry.made = None;
next = next.saturating_add(1);
}
let changed = with.iter().filter_map(|entry| entry.id).collect();
let gone = removed.iter().filter(|id| !taken.contains(id)).copied();
let spliced = Spliced {
removed: gone.collect(),
changed,
restored,
};
let change = Change::Splice {
from,
to,
removed,
with,
};
self.record(change, Some(by));
if to == len {
self.clear_halt();
}
spliced
}
fn outside<'s>(
&self,
range: Range<usize>,
outside: &'s mut Option<HashSet<MessageId>>,
) -> &'s HashSet<MessageId> {
outside.get_or_insert_with(|| {
let before = self.meta.get(..range.start).unwrap_or_default();
let after = self.meta.get(range.end..).unwrap_or_default();
before.iter().chain(after).map(|meta| meta.id).collect()
})
}
pub fn edit_each(
&mut self,
by: Author,
mut edit: impl FnMut(EntryRef<'_>) -> Option<Message>,
) -> Vec<MessageId> {
let rewritten: Vec<(usize, MessageId, Message)> = self
.entries()
.filter_map(|entry| {
let (at, id) = (entry.index, entry.id);
edit(entry).map(|message| (at, id, message))
})
.collect();
let (Some(&(first, ..)), Some(&(last, ..))) = (rewritten.first(), rewritten.last()) else {
return Vec::new();
};
let ids = rewritten.iter().map(|(_, id, _)| *id).collect();
let mut rewritten = rewritten.into_iter().peekable();
let entries: Vec<Entry> = (first..=last)
.filter_map(|at| match rewritten.next_if(|(next, ..)| *next == at) {
Some((_, id, message)) => Some(Entry {
message: Arc::new(message),
author: self.author(at).cloned(),
id: Some(id),
made: None,
}),
None => self.entry(at),
})
.collect();
self.splice(first..=last, entries, by);
ids
}
pub fn normalize(&mut self, by: Author) -> Normalized {
let entries = (0..self.messages.len()).filter_map(|at| self.entry(at));
let normalized = pairing::normalize(entries.collect());
let into: HashMap<MessageId, MessageId> = normalized
.merged
.iter()
.map(|(kept, gone)| (*gone, *kept))
.collect();
let dropped: HashSet<MessageId> = normalized.dropped.iter().copied().collect();
let anchor = |id: MessageId| -> Option<MessageId> {
let before = self.meta.iter().rev().skip_while(|meta| meta.id != id);
before
.map(|meta| meta.id)
.find_map(|id| match into.get(&id) {
Some(kept) => Some(*kept),
None => (!dropped.contains(&id)).then_some(id),
})
};
let moves: Vec<(AsideId, Option<MessageId>)> = self
.asides
.iter()
.filter_map(|(at, aside)| {
let after = aside.after?;
let moves = into.contains_key(&after) || dropped.contains(&after);
moves.then(|| (*at, anchor(after)))
})
.collect();
self.set_entries(normalized.entries.clone(), by.clone());
for (id, after) in moves {
self.move_aside(id, after, &by);
}
normalized
}
pub fn reread(&mut self, from_log: Conversation) -> Result<(), ApplyError> {
if !self.unsaved.is_empty() || !from_log.unsaved.is_empty() {
return Err(mismatch(
"a value with unsaved changes is not reread".to_owned(),
));
}
let mut asides = from_log.asides;
asides.retain(|(_, aside)| aside.saved);
let unsaved = self.asides.iter().filter(|(_, aside)| !aside.saved);
if let Some((id, _)) = unsaved
.clone()
.find(|(id, _)| asides.iter().any(|(saved, _)| saved == id))
{
return Err(mismatch(format!("aside id {} is saved in the log", id.0)));
}
let next_aside = unsaved
.clone()
.map(|(id, _)| id.0.saturating_add(1))
.fold(from_log.next_aside, u64::max);
asides.extend(unsaved.cloned());
asides.sort_by_key(|(id, _)| *id);
*self = Self {
messages: from_log.messages,
meta: from_log.meta,
asides,
halted: from_log.halted,
next_id: from_log.next_id,
next_aside,
revision: from_log.revision,
lineage: from_log.lineage,
unsaved: Vec::new(),
clock: self.clock,
};
Ok(())
}
pub fn halt(&mut self, halt: Halt) -> bool {
let halts = self.awaits_model();
if halts {
self.record(Change::Halt { halt: Some(halt) }, None);
}
halts
}
pub fn resume(&mut self) -> bool {
self.clear_halt();
self.awaits_model()
}
fn clear_halt(&mut self) {
if self.halted.is_some() {
self.record(Change::Halt { halt: None }, None);
}
}
pub fn add_aside(&mut self, aside: Aside) -> AsideId {
let id = AsideId(self.next_aside);
if aside.saved {
let by = Some(aside.author.clone());
let aside = Some(aside);
self.record(Change::Aside { id, aside }, by);
} else {
self.place_aside(id, aside);
}
id
}
pub fn replace_aside(&mut self, id: AsideId, aside: Aside) -> bool {
let Some(was_saved) = self.aside(id).map(|old| old.saved) else {
return false;
};
if aside.saved {
let by = Some(aside.author.clone());
let aside = Some(aside);
self.record(Change::Aside { id, aside }, by);
} else {
self.place_aside(id, aside);
if was_saved {
self.log(Change::Aside { id, aside: None }, None);
}
}
true
}
pub fn remove_aside(&mut self, id: AsideId) -> Option<Aside> {
let at = self.asides.iter().position(|(known, _)| *known == id)?;
let (_, aside) = self.asides.remove(at);
if aside.saved {
self.log(Change::Aside { id, aside: None }, None);
}
Some(aside)
}
fn move_aside(&mut self, id: AsideId, after: Option<MessageId>, by: &Author) {
let Some(aside) = self.aside(id).cloned() else {
return;
};
let aside = aside.after(after);
if aside.saved {
let aside = Some(aside);
self.record(Change::Aside { id, aside }, Some(by.clone()));
} else {
self.place_aside(id, aside);
}
}
fn place_aside(&mut self, id: AsideId, aside: Aside) {
self.next_aside = self.next_aside.max(id.0.saturating_add(1));
match self.asides.iter_mut().find(|(known, _)| *known == id) {
Some((_, slot)) => *slot = aside,
None => self.asides.push((id, aside)),
}
}
pub fn apply(&mut self, change: &ChangeRecord) -> Result<(), ApplyError> {
let at = change.at.unwrap_or_else(|| self.now());
self.change(&change.change, at)?;
self.revision = self.revision.saturating_add(1);
Ok(())
}
pub fn dated<R>(&mut self, at: u64, edit: impl FnOnce(&mut Self) -> R) -> R {
let outer = self.clock.replace(at);
let made = edit(self);
self.clock = outer;
made
}
fn now(&self) -> u64 {
self.clock.unwrap_or_else(crate::now_ms)
}
pub fn unsaved(&self) -> &[ChangeRecord] {
&self.unsaved
}
pub fn take_unsaved(&mut self) -> Vec<ChangeRecord> {
std::mem::take(&mut self.unsaved)
}
pub fn revision(&self) -> u64 {
self.revision
}
pub fn lineage(&self) -> Lineage {
self.lineage
}
pub fn continues(&self, lineage: Lineage, changes: u64) -> bool {
let base = self.revision.saturating_sub(self.unsaved.len() as u64);
base == changes
&& (self.lineage == lineage || (changes == 0 && self.lineage == Lineage::EMPTY))
}
pub fn continue_log(&mut self, lineage: Lineage, changes: u64) {
self.lineage = lineage;
self.revision = changes.saturating_add(self.unsaved.len() as u64);
}
pub fn view(&self) -> Self {
Self {
asides: self.asides.clone(),
..self.request_view()
}
}
pub fn request_view(&self) -> Self {
Self {
messages: self.messages.clone(),
meta: self.meta.clone(),
asides: Vec::new(),
halted: self.halted,
next_id: self.next_id,
next_aside: self.next_aside,
revision: self.revision,
lineage: self.lineage,
unsaved: Vec::new(),
clock: None,
}
}
fn record(&mut self, change: Change, by: Option<Author>) {
let at = self.now();
match self.change(&change, at) {
Ok(()) => self.log_at(change, by, at),
Err(error) => tracing::error!(%error, "a conversation edit did not fit; dropped"),
}
}
fn log(&mut self, change: Change, by: Option<Author>) {
let at = self.now();
self.log_at(change, by, at);
}
fn log_at(&mut self, change: Change, by: Option<Author>, at: u64) {
self.revision = self.revision.saturating_add(1);
self.unsaved.push(ChangeRecord {
change,
by,
at: Some(at),
});
}
fn change(&mut self, change: &Change, at: u64) -> Result<(), ApplyError> {
let meta = |entry: &Entry, made: u64| Meta {
id: entry.id.unwrap_or_default(),
author: entry.author.clone(),
made,
};
match change {
Change::Push { entries } => {
let end = self.meta.len();
let next = self.new_ids(entries, &HashSet::new(), end..end)?;
for entry in entries {
self.messages.push(entry.message.clone());
self.meta.push(meta(entry, entry.made.unwrap_or(at)));
}
self.next_id = next;
}
Change::Splice {
from,
to,
removed,
with,
} => {
let (from, to) = (*from, *to);
let replaced = self.meta.get(from..to).ok_or_else(|| {
mismatch(format!(
"splice of {from}..{to} in {} entries",
self.meta.len()
))
})?;
if !replaced.iter().map(|meta| &meta.id).eq(removed) {
return Err(mismatch(format!("splice of {from}..{to} names other ids")));
}
let kept: HashMap<MessageId, u64> =
replaced.iter().map(|meta| (meta.id, meta.made)).collect();
let next = self.new_ids(with, &removed.iter().copied().collect(), from..to)?;
let metas: Vec<Meta> = with
.iter()
.map(|entry| {
let made = entry.id.and_then(|id| kept.get(&id).copied());
meta(entry, made.or(entry.made).unwrap_or(at))
})
.collect();
self.messages
.splice(from..to, with.iter().map(|entry| entry.message.clone()));
self.meta.splice(from..to, metas);
self.next_id = next;
}
Change::Halt { halt } => self.halted = *halt,
Change::Aside {
id,
aside: Some(aside),
} => self.place_aside(*id, aside.clone()),
Change::Aside { id, aside: None } => {
let at = self.asides.iter().position(|(known, _)| known == id);
let at = at.ok_or_else(|| mismatch(format!("no aside {}", id.0)))?;
self.asides.remove(at);
}
Change::Reset { state } => {
let (unsaved, revision) = (std::mem::take(&mut self.unsaved), self.revision);
let lineage = self.lineage;
*self = state.view();
self.unsaved = unsaved;
self.revision = revision;
self.lineage = lineage;
}
}
Ok(())
}
fn new_ids(
&self,
entries: &[Entry],
reused: &HashSet<MessageId>,
range: Range<usize>,
) -> Result<u64, ApplyError> {
let mut next = self.next_id;
let mut seen = HashSet::new();
let mut outside = None;
for entry in entries {
let id = entry
.id
.ok_or_else(|| mismatch("a recorded entry without an id".to_owned()))?;
if id.0 >= next {
next = id.0.saturating_add(1);
continue;
}
let fits = reused.contains(&id)
|| (id.0 < self.next_id
&& !self.outside(range.clone(), &mut outside).contains(&id));
if !fits || !seen.insert(id) {
return Err(mismatch(format!("message id {} reused", id.0)));
}
}
Ok(next)
}
}
#[cfg(test)]
mod tests;