1use std::collections::{BTreeMap, BTreeSet};
4
5use serde::{Deserialize, Serialize};
6
7use super::{
8 CollabOpId, CollaborationAnchor, CollaborationCodecError, CollaborationOperationBodyV1,
9 CollaborationResolution, DecodedCollaborationOperation, DiscussionRecordId, DiscussionTurnV1,
10 LegacyDiscussionResolutionV1,
11};
12use crate::object::VisibilityTier;
13
14#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
15pub struct HostedCollaborationSet {
16 pub received: BTreeSet<CollabOpId>,
17 pub accepted: BTreeSet<CollabOpId>,
18 pub rejected: BTreeSet<CollabOpId>,
19}
20
21impl HostedCollaborationSet {
22 pub fn validate(
23 &self,
24 operations: &BTreeMap<CollabOpId, DecodedCollaborationOperation>,
25 ) -> Result<(), CollaborationCodecError> {
26 if !self.accepted.is_subset(&self.received) || !self.rejected.is_subset(&self.received) {
27 return Err(CollaborationCodecError::Invalid(
28 "hosted accepted and rejected sets must be subsets of received".to_string(),
29 ));
30 }
31 if !self.accepted.is_disjoint(&self.rejected) {
32 return Err(CollaborationCodecError::Invalid(
33 "hosted accepted and rejected sets must be disjoint".to_string(),
34 ));
35 }
36 for id in &self.accepted {
37 let operation = operations.get(id).ok_or_else(|| {
38 CollaborationCodecError::Invalid(format!(
39 "hosted accepted operation {id} is unavailable"
40 ))
41 })?;
42 if !operation
43 .operation
44 .parents
45 .iter()
46 .all(|parent| self.accepted.contains(parent))
47 {
48 return Err(CollaborationCodecError::Invalid(format!(
49 "hosted accepted set is not parent-closed at {id}"
50 )));
51 }
52 }
53 Ok(())
54 }
55
56 pub fn blocked_descendants(
57 &self,
58 operations: &BTreeMap<CollabOpId, DecodedCollaborationOperation>,
59 ) -> BTreeSet<CollabOpId> {
60 self.received
61 .difference(&self.accepted)
62 .filter(|id| {
63 !self.rejected.contains(id)
64 && has_unaccepted_ancestor(**id, &self.accepted, operations)
65 })
66 .copied()
67 .collect()
68 }
69}
70
71fn has_unaccepted_ancestor(
72 id: CollabOpId,
73 accepted: &BTreeSet<CollabOpId>,
74 operations: &BTreeMap<CollabOpId, DecodedCollaborationOperation>,
75) -> bool {
76 let Some(operation) = operations.get(&id) else {
77 return true;
78 };
79 let mut pending = operation.operation.parents.clone();
80 let mut seen = BTreeSet::new();
81 while let Some(parent) = pending.pop() {
82 if !accepted.contains(&parent) {
83 return true;
84 }
85 if seen.insert(parent)
86 && let Some(operation) = operations.get(&parent)
87 {
88 pending.extend(operation.operation.parents.iter().copied());
89 }
90 }
91 false
92}
93
94#[derive(Clone, Debug, PartialEq, Eq)]
95pub struct MaterializedDiscussion {
96 pub discussion_id: DiscussionRecordId,
97 pub title: String,
98 pub anchor: CollaborationAnchor,
99 pub visibility: VisibilityTier,
100 pub thread_ref: Option<String>,
101 pub turns: Vec<(CollabOpId, DiscussionTurnV1)>,
102 pub resolution: Option<CollaborationResolution>,
103 pub conflict_operations: BTreeSet<CollabOpId>,
104 pub heads: BTreeSet<CollabOpId>,
105 pub display_head: CollabOpId,
106}
107
108#[derive(Clone, Debug, Default, PartialEq, Eq)]
109pub struct MaterializedRepositoryCollaboration {
110 pub discussions: BTreeMap<DiscussionRecordId, MaterializedDiscussion>,
111 pub pending: BTreeSet<CollabOpId>,
112}
113
114pub fn materialize_repository_collaboration(
115 operations: impl IntoIterator<Item = DecodedCollaborationOperation>,
116) -> Result<MaterializedRepositoryCollaboration, CollaborationCodecError> {
117 let mut by_id = BTreeMap::new();
118 for operation in operations {
119 if by_id.insert(operation.operation_id, operation).is_some() {
120 return Err(CollaborationCodecError::Invalid(
121 "duplicate collaboration operation id".to_string(),
122 ));
123 }
124 }
125
126 let mut visible = BTreeSet::new();
127 let mut ordered = Vec::new();
128 loop {
129 let next = by_id
130 .iter()
131 .filter(|(id, operation)| {
132 !visible.contains(*id)
133 && operation
134 .operation
135 .parents
136 .iter()
137 .all(|parent| visible.contains(parent))
138 })
139 .map(|(id, operation)| (operation.operation.occurred_at_ms, *id))
140 .min();
141 let Some((_, id)) = next else { break };
142 visible.insert(id);
143 ordered.push(id);
144 }
145
146 let mut grouped: BTreeMap<DiscussionRecordId, Vec<CollabOpId>> = BTreeMap::new();
147 for id in &ordered {
148 let operation = &by_id[id].operation;
149 for parent in &operation.parents {
150 if by_id[parent].operation.discussion_id != operation.discussion_id {
151 return Err(CollaborationCodecError::Invalid(format!(
152 "operation {id} has a parent from another discussion"
153 )));
154 }
155 }
156 grouped
157 .entry(operation.discussion_id)
158 .or_default()
159 .push(*id);
160 }
161
162 let mut result = MaterializedRepositoryCollaboration {
163 discussions: BTreeMap::new(),
164 pending: by_id
165 .keys()
166 .filter(|id| !visible.contains(id))
167 .copied()
168 .collect(),
169 };
170 for (discussion_id, ids) in grouped {
171 let discussion = materialize_discussion(discussion_id, &ids, &by_id)?;
172 result.discussions.insert(discussion_id, discussion);
173 }
174 Ok(result)
175}
176
177fn materialize_discussion(
178 discussion_id: DiscussionRecordId,
179 ids: &[CollabOpId],
180 all: &BTreeMap<CollabOpId, DecodedCollaborationOperation>,
181) -> Result<MaterializedDiscussion, CollaborationCodecError> {
182 let roots = ids
183 .iter()
184 .filter(|id| all[id].operation.parents.is_empty())
185 .copied()
186 .collect::<Vec<_>>();
187 if roots.len() != 1 {
188 return Err(CollaborationCodecError::Invalid(format!(
189 "discussion {discussion_id} has {} roots",
190 roots.len()
191 )));
192 }
193 let root_id = roots[0];
194 let root = &all[&root_id].operation.body;
195 let (title, anchor, visibility, thread_ref, root_turns, base_resolution) = match root {
196 CollaborationOperationBodyV1::Open {
197 title,
198 anchor,
199 visibility,
200 turn,
201 thread_ref,
202 } => (
203 title.clone(),
204 anchor.clone(),
205 visibility.clone(),
206 thread_ref.clone(),
207 vec![turn.clone()],
208 None,
209 ),
210 CollaborationOperationBodyV1::LegacyImported {
211 title,
212 anchor,
213 visibility,
214 turns,
215 resolution,
216 ..
217 } => (
218 title.clone(),
219 anchor.clone(),
220 visibility.clone(),
221 None,
222 turns.clone(),
223 legacy_resolution(resolution),
224 ),
225 _ => {
226 return Err(CollaborationCodecError::Invalid(format!(
227 "discussion {discussion_id} root is not an open or legacy import"
228 )));
229 }
230 };
231
232 let mut turns = root_turns
233 .into_iter()
234 .map(|turn| (root_id, turn))
235 .collect::<Vec<_>>();
236 let mut state_operations = BTreeSet::new();
237 for id in ids.iter().copied().filter(|id| *id != root_id) {
238 match &all[&id].operation.body {
239 CollaborationOperationBodyV1::AppendTurn { turn } => turns.push((id, turn.clone())),
240 CollaborationOperationBodyV1::Resolve { .. }
241 | CollaborationOperationBodyV1::Reopen { .. }
242 | CollaborationOperationBodyV1::ResolveConflict { .. } => {
243 state_operations.insert(id);
244 }
245 CollaborationOperationBodyV1::Open { .. }
246 | CollaborationOperationBodyV1::LegacyImported { .. } => {
247 return Err(CollaborationCodecError::Invalid(format!(
248 "discussion {discussion_id} has multiple root operations"
249 )));
250 }
251 }
252 }
253
254 let maximal_state = state_operations
255 .iter()
256 .filter(|candidate| {
257 !state_operations
258 .iter()
259 .any(|other| candidate != &other && precedes(**candidate, *other, all))
260 })
261 .copied()
262 .collect::<BTreeSet<_>>();
263 let mut outcomes = BTreeMap::new();
264 for id in &maximal_state {
265 outcomes.insert(*id, resolution_outcome(*id, all, &mut BTreeSet::new())?);
266 }
267 let first_outcome = outcomes.values().next().cloned();
268 let conflicts = if outcomes
269 .values()
270 .all(|outcome| Some(outcome) == first_outcome.as_ref())
271 {
272 BTreeSet::new()
273 } else {
274 outcomes.keys().copied().collect()
275 };
276 let resolution = if conflicts.is_empty() {
277 first_outcome.unwrap_or(base_resolution)
278 } else {
279 None
280 };
281
282 let ids_set = ids.iter().copied().collect::<BTreeSet<_>>();
283 let heads = ids_set
284 .iter()
285 .filter(|candidate| {
286 !ids_set
287 .iter()
288 .any(|other| candidate != &other && precedes(**candidate, *other, all))
289 })
290 .copied()
291 .collect::<BTreeSet<_>>();
292 let display_head = *heads.iter().next().expect("root guarantees a head");
293 Ok(MaterializedDiscussion {
294 discussion_id,
295 title,
296 anchor,
297 visibility,
298 thread_ref,
299 turns,
300 resolution,
301 conflict_operations: conflicts,
302 heads,
303 display_head,
304 })
305}
306
307fn resolution_outcome(
308 id: CollabOpId,
309 operations: &BTreeMap<CollabOpId, DecodedCollaborationOperation>,
310 visiting: &mut BTreeSet<CollabOpId>,
311) -> Result<Option<CollaborationResolution>, CollaborationCodecError> {
312 if !visiting.insert(id) {
313 return Err(CollaborationCodecError::Invalid(format!(
314 "collaboration conflict resolution cycle at {id}"
315 )));
316 }
317 let body = &operations
318 .get(&id)
319 .ok_or_else(|| CollaborationCodecError::Invalid(format!("missing operation {id}")))?
320 .operation
321 .body;
322 let result = match body {
323 CollaborationOperationBodyV1::Resolve { resolution } => Some(resolution.clone()),
324 CollaborationOperationBodyV1::Reopen { .. } => None,
325 CollaborationOperationBodyV1::ResolveConflict { selected, .. } => {
326 resolution_outcome(*selected, operations, visiting)?
327 }
328 _ => {
329 return Err(CollaborationCodecError::Invalid(format!(
330 "operation {id} does not select a resolution outcome"
331 )));
332 }
333 };
334 visiting.remove(&id);
335 Ok(result)
336}
337
338fn precedes(
339 ancestor: CollabOpId,
340 descendant: CollabOpId,
341 operations: &BTreeMap<CollabOpId, DecodedCollaborationOperation>,
342) -> bool {
343 let mut pending = operations[&descendant].operation.parents.clone();
344 let mut seen = BTreeSet::new();
345 while let Some(id) = pending.pop() {
346 if id == ancestor {
347 return true;
348 }
349 if seen.insert(id)
350 && let Some(operation) = operations.get(&id)
351 {
352 pending.extend(operation.operation.parents.iter().copied());
353 }
354 }
355 false
356}
357
358fn legacy_resolution(value: &LegacyDiscussionResolutionV1) -> Option<CollaborationResolution> {
359 match value {
360 LegacyDiscussionResolutionV1::Open => None,
361 LegacyDiscussionResolutionV1::AddressedByState { state_id } => {
362 Some(CollaborationResolution::AddressedByState {
363 state_id: *state_id,
364 })
365 }
366 LegacyDiscussionResolutionV1::Dismissed { reason } => {
367 Some(CollaborationResolution::Dismissed {
368 reason: reason.clone(),
369 })
370 }
371 LegacyDiscussionResolutionV1::Annotation { annotation_id } => {
372 Some(CollaborationResolution::Annotation {
373 annotation_id: annotation_id.clone(),
374 })
375 }
376 }
377}
378
379#[cfg(test)]
380mod tests {
381 use super::*;
382 use crate::object::{
383 Attribution, CollaborationIdempotencyKey, CollaborationOperationEnvelope, Principal,
384 };
385
386 fn discussion_id() -> DiscussionRecordId {
387 "disc-018f47ea-4a54-7c89-b012-3456789abcde".parse().unwrap()
388 }
389
390 fn author() -> Attribution {
391 Attribution::human(Principal::new("Ada", "ada@example.com"))
392 }
393
394 fn decoded(
395 parents: Vec<CollabOpId>,
396 key: &str,
397 at: i64,
398 body: CollaborationOperationBodyV1,
399 ) -> DecodedCollaborationOperation {
400 let operation = CollaborationOperationEnvelope::new(
401 discussion_id(),
402 parents,
403 CollaborationIdempotencyKey::new(key).unwrap(),
404 author(),
405 at,
406 body,
407 )
408 .unwrap();
409 let bytes = operation.encode().unwrap();
410 CollaborationOperationEnvelope::decode(&bytes).unwrap()
411 }
412
413 fn root() -> DecodedCollaborationOperation {
414 decoded(
415 vec![],
416 "root",
417 1,
418 CollaborationOperationBodyV1::Open {
419 title: "Review".to_string(),
420 anchor: CollaborationAnchor::Repository,
421 visibility: VisibilityTier::default(),
422 turn: DiscussionTurnV1::new("first").unwrap(),
423 thread_ref: None,
424 },
425 )
426 }
427
428 #[test]
429 fn op_set_union_converges_independent_of_arrival_order() {
430 let root = root();
431 let a = decoded(
432 vec![root.operation_id],
433 "a",
434 2,
435 CollaborationOperationBodyV1::AppendTurn {
436 turn: DiscussionTurnV1::new("a").unwrap(),
437 },
438 );
439 let b = decoded(
440 vec![root.operation_id],
441 "b",
442 3,
443 CollaborationOperationBodyV1::AppendTurn {
444 turn: DiscussionTurnV1::new("b").unwrap(),
445 },
446 );
447 let forward =
448 materialize_repository_collaboration(vec![root.clone(), a.clone(), b.clone()]).unwrap();
449 let reverse = materialize_repository_collaboration(vec![b, a, root]).unwrap();
450 assert_eq!(forward, reverse);
451 let discussion = &forward.discussions[&discussion_id()];
452 assert_eq!(discussion.turns.len(), 3);
453 assert_eq!(discussion.heads.len(), 2);
454 assert_eq!(
455 discussion.display_head,
456 *discussion.heads.iter().next().unwrap()
457 );
458 }
459
460 #[test]
461 fn missing_parent_blocks_descendant_until_causal_closure_arrives() {
462 let root = root();
463 let missing = CollabOpId::from_bytes([9; 32]);
464 let child = decoded(
465 vec![missing],
466 "child",
467 2,
468 CollaborationOperationBodyV1::AppendTurn {
469 turn: DiscussionTurnV1::new("waiting").unwrap(),
470 },
471 );
472 let materialized = materialize_repository_collaboration(vec![root, child.clone()]).unwrap();
473 assert_eq!(materialized.pending, BTreeSet::from([child.operation_id]));
474 }
475
476 #[test]
477 fn competing_resolutions_conflict_and_causal_reopen_clears_resolution() {
478 let root = root();
479 let left = decoded(
480 vec![root.operation_id],
481 "left",
482 2,
483 CollaborationOperationBodyV1::Resolve {
484 resolution: CollaborationResolution::Dismissed {
485 reason: "obsolete".to_string(),
486 },
487 },
488 );
489 let right = decoded(
490 vec![root.operation_id],
491 "right",
492 3,
493 CollaborationOperationBodyV1::Resolve {
494 resolution: CollaborationResolution::Annotation {
495 annotation_id: "ann-1".to_string(),
496 },
497 },
498 );
499 let conflicted =
500 materialize_repository_collaboration(vec![root.clone(), left.clone(), right.clone()])
501 .unwrap();
502 assert_eq!(
503 conflicted.discussions[&discussion_id()]
504 .conflict_operations
505 .len(),
506 2
507 );
508 assert_eq!(conflicted.discussions[&discussion_id()].resolution, None);
509
510 let mut competing = vec![left.operation_id, right.operation_id];
511 competing.sort();
512 let selected = competing[0];
513 let resolved = decoded(
514 competing.clone(),
515 "resolve-conflict",
516 4,
517 CollaborationOperationBodyV1::ResolveConflict {
518 competing,
519 selected,
520 },
521 );
522 let reopened = decoded(
523 vec![resolved.operation_id],
524 "reopen",
525 5,
526 CollaborationOperationBodyV1::Reopen {
527 reason: "new evidence".to_string(),
528 },
529 );
530 let view =
531 materialize_repository_collaboration(vec![root, left, right, resolved, reopened])
532 .unwrap();
533 assert_eq!(view.discussions[&discussion_id()].resolution, None);
534 }
535
536 #[test]
537 fn concurrent_reopen_and_resolve_surface_conflict() {
538 let root = root();
539 let resolved = decoded(
540 vec![root.operation_id],
541 "resolve",
542 2,
543 CollaborationOperationBodyV1::Resolve {
544 resolution: CollaborationResolution::Dismissed {
545 reason: "done".to_string(),
546 },
547 },
548 );
549 let reopened = decoded(
550 vec![root.operation_id],
551 "reopen",
552 3,
553 CollaborationOperationBodyV1::Reopen {
554 reason: "new evidence".to_string(),
555 },
556 );
557 let view =
558 materialize_repository_collaboration(vec![root, resolved.clone(), reopened.clone()])
559 .unwrap();
560 assert_eq!(
561 view.discussions[&discussion_id()].conflict_operations,
562 BTreeSet::from([resolved.operation_id, reopened.operation_id])
563 );
564 }
565
566 #[test]
567 fn competing_conflict_resolutions_form_a_recursive_conflict() {
568 let root = root();
569 let left = decoded(
570 vec![root.operation_id],
571 "left",
572 2,
573 CollaborationOperationBodyV1::Resolve {
574 resolution: CollaborationResolution::Dismissed {
575 reason: "left".to_string(),
576 },
577 },
578 );
579 let right = decoded(
580 vec![root.operation_id],
581 "right",
582 3,
583 CollaborationOperationBodyV1::Resolve {
584 resolution: CollaborationResolution::Dismissed {
585 reason: "right".to_string(),
586 },
587 },
588 );
589 let mut competing = vec![left.operation_id, right.operation_id];
590 competing.sort();
591 let choose_left = decoded(
592 competing.clone(),
593 "choose-left",
594 4,
595 CollaborationOperationBodyV1::ResolveConflict {
596 competing: competing.clone(),
597 selected: left.operation_id,
598 },
599 );
600 let choose_right = decoded(
601 competing.clone(),
602 "choose-right",
603 5,
604 CollaborationOperationBodyV1::ResolveConflict {
605 competing,
606 selected: right.operation_id,
607 },
608 );
609 let view = materialize_repository_collaboration(vec![
610 root,
611 left,
612 right,
613 choose_left.clone(),
614 choose_right.clone(),
615 ])
616 .unwrap();
617 assert_eq!(
618 view.discussions[&discussion_id()].conflict_operations,
619 BTreeSet::from([choose_left.operation_id, choose_right.operation_id])
620 );
621 }
622
623 #[test]
624 fn hosted_sets_separate_rejected_and_blocked_descendants() {
625 let root = root();
626 let rejected = decoded(
627 vec![root.operation_id],
628 "rejected",
629 2,
630 CollaborationOperationBodyV1::AppendTurn {
631 turn: DiscussionTurnV1::new("rejected").unwrap(),
632 },
633 );
634 let child = decoded(
635 vec![rejected.operation_id],
636 "child",
637 3,
638 CollaborationOperationBodyV1::AppendTurn {
639 turn: DiscussionTurnV1::new("blocked").unwrap(),
640 },
641 );
642 let grandchild = decoded(
643 vec![child.operation_id],
644 "grandchild",
645 4,
646 CollaborationOperationBodyV1::AppendTurn {
647 turn: DiscussionTurnV1::new("also blocked").unwrap(),
648 },
649 );
650 let operations = [
651 root.clone(),
652 rejected.clone(),
653 child.clone(),
654 grandchild.clone(),
655 ]
656 .into_iter()
657 .map(|operation| (operation.operation_id, operation))
658 .collect();
659 let hosted = HostedCollaborationSet {
660 received: BTreeSet::from([
661 root.operation_id,
662 rejected.operation_id,
663 child.operation_id,
664 grandchild.operation_id,
665 ]),
666 accepted: BTreeSet::from([root.operation_id]),
667 rejected: BTreeSet::from([rejected.operation_id]),
668 };
669 hosted.validate(&operations).unwrap();
670 assert_eq!(
671 hosted.blocked_descendants(&operations),
672 BTreeSet::from([child.operation_id, grandchild.operation_id])
673 );
674 let invalid = HostedCollaborationSet {
675 received: hosted.received.clone(),
676 accepted: BTreeSet::from([root.operation_id, child.operation_id]),
677 rejected: BTreeSet::new(),
678 };
679 assert!(invalid.validate(&operations).is_err());
680 }
681}