use std::num::NonZeroUsize;
use haematite::db::{ObservedChange, ReadOnlyDatabase};
use serde::{Deserialize, Serialize};
use super::super::{
Captured, Catalogue, Failure, ForkProvenance, Held, document_key, edges, identifier, reads,
};
use super::{choice, compare, cursor, values};
use crate::serve::protocol::{Edge, Node, ReviewPage};
pub(in crate::serve::storage) const SCAN_ALLOWANCE: usize = 64;
#[derive(Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum Side {
Source,
Target,
}
#[derive(Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum End {
From,
To,
}
#[derive(Serialize)]
#[serde(rename_all = "snake_case")]
enum State {
Dangling,
DependsOnChoice,
}
#[derive(Serialize)]
pub(in crate::serve) struct Line {
choice_token: String,
relation: String,
from: Node,
to: Node,
missing_end: End,
removed_by: Side,
state: State,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Work {
node: Node,
removed_by: Side,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Incident {
work: Work,
end: End,
after: Option<String>,
}
#[derive(Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct Cursor {
comparison: String,
comparison_done: bool,
pending: Vec<Work>,
incident: Option<Incident>,
}
pub(in crate::serve) struct Page {
pub(in crate::serve) selection: Captured,
pub(in crate::serve) from: ForkProvenance,
pub(in crate::serve) parent: Captured,
pub(in crate::serve) lines: Vec<Line>,
pub(in crate::serve) next_after: Option<String>,
}
struct Versions {
selection: Captured,
from: ForkProvenance,
origin: Captured,
parent: Captured,
_held: Vec<Held>,
}
pub(in crate::serve) fn page(
reader: &ReadOnlyDatabase,
catalogue: &Catalogue,
request: &ReviewPage,
limit: NonZeroUsize,
) -> Result<Page, Failure> {
let scan = NonZeroUsize::new(SCAN_ALLOWANCE)
.ok_or_else(|| Failure::new("scan_allowance", "raw scan allowance must be nonzero"))?;
let (mut cursor, versions) = match request.after.as_deref() {
None => {
let compared = compare::page_both(reader, catalogue, request, scan)?;
let cursor = comparison_cursor(&compared)?;
let versions = Versions {
selection: compared.selection,
from: compared.from,
origin: compared.origin,
parent: compared.parent,
_held: compared.held,
};
return finish(&cursor, versions, Vec::new());
}
Some(text) => {
let cursor: Cursor = cursor::decode(text)?;
validate_cursor(&cursor)?;
let roots = compare::capture(reader, catalogue, &comparison_request(request, &cursor))?;
let versions = Versions {
selection: roots.selection,
from: roots.from,
origin: roots.origin,
parent: roots.parent,
_held: roots.held,
};
(cursor, versions)
}
};
if cursor.incident.is_none() {
if let Some(work) = cursor.pending.pop() {
cursor.incident = Some(Incident {
work,
end: End::From,
after: None,
});
} else if !cursor.comparison_done {
let compared = compare::page_both(
reader,
catalogue,
&comparison_request(request, &cursor),
scan,
)?;
cursor = comparison_cursor(&compared)?;
return finish(&cursor, versions, Vec::new());
}
}
let lines = match cursor.incident.take() {
Some(mut incident) => {
let (lines, next) = incident_page(reader, &versions, &incident, limit)?;
match next {
Some(next) => {
incident.after = Some(next);
cursor.incident = Some(incident);
}
None if matches!(incident.end, End::From) => {
incident.end = End::To;
incident.after = None;
cursor.incident = Some(incident);
}
None => {}
}
lines
}
None => Vec::new(),
};
finish(&cursor, versions, lines)
}
fn comparison_request(request: &ReviewPage, cursor: &Cursor) -> ReviewPage {
ReviewPage {
branch: request.branch.clone(),
after: Some(cursor.comparison.clone()),
limit: request.limit,
}
}
fn comparison_cursor(compared: &compare::Page) -> Result<Cursor, Failure> {
let deleted = first_document_removal(compared)?;
let mut pending = Vec::new();
for (side, changes) in [
(Side::Source, &compared.source),
(Side::Target, &compared.target),
] {
for change in changes {
if deleted != Some(change.key()) || !matches!(change, ObservedChange::Removed { .. }) {
continue;
}
if let Some(values::Identity::Document(collection, document_id)) =
values::identity(change.key())?
{
pending.push(Work {
node: Node {
collection,
document_id,
},
removed_by: side,
});
}
}
}
if pending.len() > 2 {
return Err(Failure::new(
"page_limit",
"deletion comparison exceeded its work bound",
));
}
pending.reverse();
Ok(Cursor {
comparison: match deleted {
Some(key) => compare::after_key(compared, key)?,
None => match &compared.next_after {
Some(next) => next.clone(),
None => compare::first_cursor(compared)?,
},
},
comparison_done: deleted.is_none() && compared.next_after.is_none(),
pending,
incident: None,
})
}
fn first_document_removal(compared: &compare::Page) -> Result<Option<&[u8]>, Failure> {
let mut first: Option<&[u8]> = None;
for change in compared.source.iter().chain(&compared.target) {
if matches!(change, ObservedChange::Removed { .. })
&& matches!(
values::identity(change.key())?,
Some(values::Identity::Document(_, _))
)
&& first.is_none_or(|key| change.key() < key)
{
first = Some(change.key());
}
}
Ok(first)
}
fn validate_cursor(cursor: &Cursor) -> Result<(), Failure> {
if cursor.pending.len() > 2 {
return Err(Failure::new(
"invalid_continuation",
"too many pending endpoint scans",
));
}
for work in cursor
.pending
.iter()
.chain(cursor.incident.iter().map(|incident| &incident.work))
{
identifier(&work.node.collection).map_err(|_| {
Failure::new(
"invalid_continuation",
"endpoint scan has an invalid collection",
)
})?;
identifier(&work.node.document_id).map_err(|_| {
Failure::new(
"invalid_continuation",
"endpoint scan has an invalid document",
)
})?;
}
Ok(())
}
fn finish(cursor: &Cursor, versions: Versions, lines: Vec<Line>) -> Result<Page, Failure> {
let done = cursor.comparison_done && cursor.pending.is_empty() && cursor.incident.is_none();
let next_after = if done {
None
} else {
Some(cursor::encode(&cursor)?)
};
Ok(Page {
selection: versions.selection,
from: versions.from,
parent: versions.parent,
lines,
next_after,
})
}
struct Triple {
ancestor: Option<Vec<u8>>,
target: Option<Vec<u8>>,
source: Option<Vec<u8>>,
}
enum Presence {
Present,
Absent,
Depends,
}
impl Triple {
fn presence(&self) -> Presence {
let value = if self.source == self.ancestor {
&self.target
} else if self.target == self.ancestor || self.source == self.target {
&self.source
} else {
return Presence::Depends;
};
if value.is_some() {
Presence::Present
} else {
Presence::Absent
}
}
}
fn triple(reader: &ReadOnlyDatabase, versions: &Versions, key: &[u8]) -> Result<Triple, Failure> {
Ok(Triple {
ancestor: reads::get_bytes(reader, &versions.origin, key)?,
target: reads::get_bytes(reader, &versions.parent, key)?,
source: reads::get_bytes(reader, &versions.selection, key)?,
})
}
fn incident_page(
reader: &ReadOnlyDatabase,
versions: &Versions,
incident: &Incident,
limit: NonZeroUsize,
) -> Result<(Vec<Line>, Option<String>), Failure> {
let pins = choice::Pins::new(&versions.origin, &versions.parent, &versions.selection)?;
let document = triple(
reader,
versions,
&document_key(
&incident.work.node.collection,
&incident.work.node.document_id,
),
)?;
let removed = match incident.work.removed_by {
Side::Source => &document.source,
Side::Target => &document.target,
};
if document.ancestor.is_none() || removed.is_some() {
return Err(Failure::new(
"invalid_continuation",
"endpoint was not deleted at the pinned versions",
));
}
let at = match incident.work.removed_by {
Side::Source => &versions.parent,
Side::Target => &versions.selection,
};
let result = match incident.end {
End::From => edges::out(
reader,
at,
&incident.work.node,
None,
incident.after.as_deref(),
limit,
None,
)?,
End::To => edges::incoming(
reader,
at,
&incident.work.node,
None,
incident.after.as_deref(),
limit,
None,
)?,
};
let mut lines = Vec::new();
for edge in result.entries {
if let Some(line) = finding(reader, versions, &pins, incident, &document, edge)? {
lines.push(line);
}
}
Ok((lines, result.next_after))
}
fn finding(
reader: &ReadOnlyDatabase,
versions: &Versions,
pins: &choice::Pins,
incident: &Incident,
document: &Triple,
edge: Edge,
) -> Result<Option<Line>, Failure> {
let (key, _) = edges::keys(&edge.from, &edge.relation, &edge.to)?;
let values = triple(reader, versions, &key)?;
let indexed = match incident.work.removed_by {
Side::Source => &values.target,
Side::Target => &values.source,
};
let properties = indexed.as_deref().ok_or_else(|| {
Failure::new(
"edge_index_inconsistent",
"incident index names an absent outgoing edge",
)
})?;
if serde_json::from_slice::<serde_json::Value>(properties).map_err(Failure::storage)?
!= edge.properties
{
return Err(Failure::new(
"edge_index_inconsistent",
"incident and outgoing edge properties differ",
));
}
let edge_presence = values.presence();
let document_presence = document.presence();
if matches!(edge_presence, Presence::Absent) || matches!(document_presence, Presence::Present) {
return Ok(None);
}
let other = match incident.end {
End::From => &edge.to,
End::To => &edge.from,
};
let other_presence = triple(
reader,
versions,
&document_key(&other.collection, &other.document_id),
)?
.presence();
let state = if matches!(edge_presence, Presence::Depends)
|| matches!(document_presence, Presence::Depends)
|| matches!(other_presence, Presence::Depends)
{
State::DependsOnChoice
} else {
State::Dangling
};
Ok(Some(Line {
choice_token: pins.token(
&key,
values.ancestor.as_deref(),
values.target.as_deref(),
values.source.as_deref(),
)?,
relation: edge.relation,
from: edge.from,
to: edge.to,
missing_end: incident.end,
removed_by: incident.work.removed_by,
state,
}))
}
#[cfg(test)]
#[path = "review_dangling_tests.rs"]
mod tests;