use std::{
borrow::Borrow,
collections::HashMap,
convert::Infallible,
error::Error,
fmt::{self, Debug},
future::{self, Future},
hash::Hash,
sync::{Arc, PoisonError, RwLock, RwLockReadGuard, RwLockWriteGuard},
};
use futures_util::{Stream, stream};
use tokio::sync::watch;
use crate::{Categorized, Category, Envelope, Failure, Permanent, Revision, Revisioned};
mod index;
#[cfg(test)]
mod tests;
pub use index::InMemoryIndex;
pub trait View<T>: Send + Sync {
type Error: Error + Categorized + Send + Sync + 'static;
fn all(&self) -> impl Stream<Item = Result<T, Self::Error>> + Send;
}
pub trait KeyedView<K: ?Sized, T>: View<T> {
fn get(&self, key: &K) -> impl Future<Output = Result<Option<T>, Self::Error>> + Send;
}
pub struct InMemoryView<K, T> {
shared: Arc<Shared<K, T>>,
}
type KeyOf<K, T> = Box<dyn Fn(&T) -> K + Send + Sync>;
type RevisionIn<T> = Box<dyn Fn(&mut T) -> &mut Option<Revision> + Send + Sync>;
struct Shared<K, T> {
key_of: KeyOf<K, T>,
revision_in: Option<RevisionIn<T>>,
entries: RwLock<Entries<K, T>>,
revision: watch::Sender<Option<Revision>>,
}
struct Entries<K, T> {
items: HashMap<K, Entry<T>>,
order: Vec<K>,
indexes: Vec<Arc<dyn Indexer<K, T>>>,
}
struct Entry<T> {
item: T,
revision: Revision,
part: usize,
position: usize,
}
trait Indexer<K, T>: Send + Sync {
fn add(&self, key: &K, item: &T);
fn remove(&self, key: &K, item: &T);
}
struct AppliedEvent {
id: Revision,
part: usize,
}
impl AppliedEvent {
fn of<E>(event: &Envelope<E>) -> Self {
Self {
id: event.id,
part: event.part,
}
}
fn is_newer_than<T>(&self, entry: &Entry<T>) -> bool {
if self.id == entry.revision {
return self.part > entry.part;
}
self.id > entry.revision
}
}
impl<K, T> InMemoryView<K, T>
where
K: Clone + Debug + Eq + Hash + Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
{
#[must_use]
pub fn new(key_of: impl Fn(&T) -> K + Send + Sync + 'static) -> Self {
Self::build(Box::new(key_of), None)
}
#[must_use]
pub fn with_revision_in(
key_of: impl Fn(&T) -> K + Send + Sync + 'static,
field: impl Fn(&mut T) -> &mut Option<Revision> + Send + Sync + 'static,
) -> Self {
Self::build(Box::new(key_of), Some(Box::new(field)))
}
fn build(key_of: KeyOf<K, T>, revision_in: Option<RevisionIn<T>>) -> Self {
Self {
shared: Arc::new(Shared {
key_of,
revision_in,
entries: RwLock::new(Entries {
items: HashMap::new(),
order: Vec::new(),
indexes: Vec::new(),
}),
revision: watch::Sender::new(None),
}),
}
}
pub fn seen(&self, event_id: Revision) {
let _ = self.shared.revision.send_if_modified(|seen| {
if seen.is_some_and(|seen| seen >= event_id) {
return false;
}
*seen = Some(event_id);
true
});
}
pub fn insert<E>(&self, event: &Envelope<E>, item: T) -> Result<Outcome, ViewError> {
let event = AppliedEvent::of(event);
let mut entries = self.shared.write();
let key = (self.shared.key_of)(&item);
if let Some(entry) = entries.items.get(&key) {
if !event.is_newer_than(entry) {
return Ok(Outcome::AlreadyApplied);
}
return Err(ViewError::permanent(format!(
"event {} inserts an item with the key {key:?}, which is already taken",
event.id,
)));
}
self.shared.add(&mut entries, key, item, &event);
Ok(Outcome::Added)
}
pub fn upsert<E, Q>(
&self,
key: &Q,
event: &Envelope<E>,
change: impl FnOnce(&mut T),
) -> Result<Outcome, ViewError>
where
K: Borrow<Q>,
Q: Debug + Eq + Hash + ?Sized,
T: Default,
{
let event = AppliedEvent::of(event);
let mut entries = self.shared.write();
if entries.items.contains_key(key) {
return self.shared.change(&mut entries, key, &event, change);
}
let mut item = T::default();
change(&mut item);
let item_key = (self.shared.key_of)(&item);
if item_key.borrow() != key {
return Err(ViewError::permanent(format!(
"event {} upserts an item with the key {item_key:?} under the key {key:?}",
event.id,
)));
}
self.shared.add(&mut entries, item_key, item, &event);
Ok(Outcome::Added)
}
pub fn update<E, Q>(
&self,
key: &Q,
event: &Envelope<E>,
change: impl FnOnce(&mut T),
) -> Result<Outcome, ViewError>
where
K: Borrow<Q>,
Q: Eq + Hash + ?Sized,
{
let event = AppliedEvent::of(event);
self.shared
.change(&mut self.shared.write(), key, &event, change)
}
pub fn delete<E, Q>(&self, key: &Q, event: &Envelope<E>) -> Outcome
where
K: Borrow<Q>,
Q: Eq + Hash + ?Sized,
{
let event = AppliedEvent::of(event);
self.shared.write().delete(key, &event)
}
pub fn update_where<E>(
&self,
matches: impl FnMut(&T) -> bool,
event: &Envelope<E>,
change: impl FnMut(&mut T),
) -> Result<usize, ViewError> {
let event = AppliedEvent::of(event);
let mut entries = self.shared.write();
let keys = entries.keys_where(matches);
self.shared.change_all(&mut entries, keys, &event, change)
}
pub fn delete_where<E>(&self, matches: impl FnMut(&T) -> bool, event: &Envelope<E>) -> usize {
let event = AppliedEvent::of(event);
let mut entries = self.shared.write();
let keys = entries.keys_where(matches);
entries.delete_all(&keys, &event)
}
}
impl<K, T> Shared<K, T>
where
K: Clone + Debug + Eq + Hash + Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
{
fn read(&self) -> RwLockReadGuard<'_, Entries<K, T>> {
self.entries.read().unwrap_or_else(PoisonError::into_inner)
}
fn write(&self) -> RwLockWriteGuard<'_, Entries<K, T>> {
self.entries.write().unwrap_or_else(PoisonError::into_inner)
}
fn set_revision(&self, item: &mut T, event: &AppliedEvent) {
if let Some(field) = &self.revision_in {
*field(item) = Some(event.id);
}
}
fn add(&self, entries: &mut Entries<K, T>, key: K, mut item: T, event: &AppliedEvent) {
self.set_revision(&mut item, event);
for index in &entries.indexes {
index.add(&key, &item);
}
let position = entries.order.len();
entries.order.push(key.clone());
let _ = entries.items.insert(
key,
Entry {
item,
revision: event.id,
part: event.part,
position,
},
);
}
fn change<Q>(
&self,
entries: &mut Entries<K, T>,
key: &Q,
event: &AppliedEvent,
change: impl FnOnce(&mut T),
) -> Result<Outcome, ViewError>
where
K: Borrow<Q>,
Q: Eq + Hash + ?Sized,
{
let Some((key, entry)) = entries.items.get_key_value(key) else {
return Ok(Outcome::Missing);
};
if !event.is_newer_than(entry) {
return Ok(Outcome::AlreadyApplied);
}
let mut changed = entry.item.clone();
change(&mut changed);
let changed_key = (self.key_of)(&changed);
if changed_key != *key {
return Err(ViewError::permanent(format!(
"event {} changes the key of an item from {key:?} to {changed_key:?}",
event.id,
)));
}
self.set_revision(&mut changed, event);
for index in &entries.indexes {
index.remove(&changed_key, &entry.item);
index.add(&changed_key, &changed);
}
let entry = entries
.items
.get_mut::<K>(&changed_key)
.expect("the item is still there, since the view is locked");
entry.item = changed;
entry.revision = event.id;
entry.part = event.part;
Ok(Outcome::Applied)
}
fn change_all(
&self,
entries: &mut Entries<K, T>,
keys: Vec<K>,
event: &AppliedEvent,
mut change: impl FnMut(&mut T),
) -> Result<usize, ViewError> {
let mut changed = 0;
for key in keys {
if self.change(entries, &key, event, &mut change)? == Outcome::Applied {
changed += 1;
}
}
Ok(changed)
}
}
impl<K: Eq + Hash, T> Entries<K, T> {
fn delete<Q>(&mut self, key: &Q, event: &AppliedEvent) -> Outcome
where
K: Borrow<Q> + Clone,
Q: Eq + Hash + ?Sized,
{
let Some((key, entry)) = self.items.get_key_value(key) else {
return Outcome::Missing;
};
if !event.is_newer_than(entry) {
return Outcome::AlreadyApplied;
}
for index in &self.indexes {
index.remove(key, &entry.item);
}
let key = key.clone();
let _ = self.items.remove::<K>(&key);
self.compact();
Outcome::Applied
}
fn delete_all(&mut self, keys: &[K], event: &AppliedEvent) -> usize
where
K: Clone,
{
keys.iter()
.filter(|key| self.delete(*key, event) == Outcome::Applied)
.count()
}
fn listed(&self) -> impl Iterator<Item = (&K, &Entry<T>)> {
self.order.iter().enumerate().filter_map(|(position, key)| {
self.items
.get_key_value(key)
.filter(|(_, entry)| entry.position == position)
})
}
fn keys_where(&self, mut matches: impl FnMut(&T) -> bool) -> Vec<K>
where
K: Clone,
{
self.listed()
.filter(|(_, entry)| matches(&entry.item))
.map(|(key, _)| key.clone())
.collect()
}
fn compact(&mut self) {
if self.order.len() < 2 * self.items.len() + 16 {
return;
}
let mut order = Vec::with_capacity(self.items.len());
for (position, key) in self.order.drain(..).enumerate() {
if let Some(entry) = self.items.get_mut(&key)
&& entry.position == position
{
entry.position = order.len();
order.push(key);
}
}
self.order = order;
}
}
impl<K, T> View<T> for InMemoryView<K, T>
where
K: Clone + Debug + Eq + Hash + Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
{
type Error = Infallible;
fn all(&self) -> impl Stream<Item = Result<T, Infallible>> + Send {
let items: Vec<T> = self
.shared
.read()
.listed()
.map(|(_, entry)| entry.item.clone())
.collect();
stream::iter(items.into_iter().map(Ok))
}
}
impl<K, Q, T> KeyedView<Q, T> for InMemoryView<K, T>
where
K: Borrow<Q> + Clone + Debug + Eq + Hash + Send + Sync + 'static,
Q: Eq + Hash + ?Sized,
T: Clone + Send + Sync + 'static,
{
fn get(&self, key: &Q) -> impl Future<Output = Result<Option<T>, Infallible>> + Send {
future::ready(Ok(self
.shared
.read()
.items
.get(key)
.map(|entry| entry.item.clone())))
}
}
impl<K, T> Revisioned for InMemoryView<K, T>
where
K: Clone + Debug + Eq + Hash + Send + Sync + 'static,
T: Clone + Send + Sync + 'static,
{
type Error = Infallible;
fn revision(&self) -> impl Future<Output = Result<Option<Revision>, Infallible>> + Send {
future::ready(Ok(*self.shared.revision.borrow()))
}
async fn wait_for(&self, revision: Revision) -> Result<(), Infallible> {
let mut seen = self.shared.revision.subscribe();
drop(
seen.wait_for(|seen| seen.is_some_and(|seen| seen >= revision))
.await
.expect("the view keeps its revision while it is borrowed"),
);
Ok(())
}
}
impl<K, T> Debug for InMemoryView<K, T> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
let entries = self
.shared
.entries
.read()
.unwrap_or_else(PoisonError::into_inner);
formatter
.debug_struct("InMemoryView")
.field("items", &entries.items.len())
.field("indexes", &entries.indexes.len())
.field(
"revision",
&self.shared.revision.borrow().map(|seen| seen.to_string()),
)
.finish_non_exhaustive()
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Outcome {
Applied,
Missing,
AlreadyApplied,
Added,
}
impl fmt::Display for Outcome {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Applied => "applied",
Self::Missing => "missing",
Self::AlreadyApplied => "already applied",
Self::Added => "added",
})
}
}
#[derive(Debug)]
pub enum ViewError {
Permanent(Permanent),
}
impl ViewError {
fn permanent(message: String) -> Self {
Self::Permanent(Permanent::Other(Failure::new(message, None)))
}
}
impl fmt::Display for ViewError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Permanent(failure) => fmt::Display::fmt(failure, formatter),
}
}
}
impl Error for ViewError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
match self {
Self::Permanent(failure) => failure.source(),
}
}
}
impl Categorized for ViewError {
fn category(&self) -> Option<Category> {
Some(Category::Permanent)
}
}