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