onetaskgraph_core/engine/copy.rs
1//! The copy verb: one item out of one source and into another, by the rules that make a
2//! second copy an update rather than a duplicate.
3//!
4//! Correspondence lives on the item and never in a table. A copied item carries
5//! [`GlobalId::ORIGIN_KEY`], whose value is the qualified id it was copied from, and the
6//! two match rules below read exactly that — so nothing here is written down outside the
7//! plugin that owns the item, and the invariant this engine is built around is untouched.
8//!
9//! 1. **Follow the origin.** An item already carrying an origin whose source half is the
10//! destination names the destination item *directly*, and the copy updates it. This is
11//! the half that makes an edit's copy-back an update: the local file came from the
12//! remote item and knows which one.
13//! 2. **Search by origin.** Otherwise the destination is scanned, one page at a time, for
14//! an item whose origin is the id being copied. Found, the copy updates it; not found,
15//! the copy creates one carrying that origin.
16//!
17//! Which rule found the item decides what the copy records there. A copy that got its
18//! target from rule 1 is a copy-back — the destination is the *original*, and the item
19//! being copied is the one that came from it — so the destination keeps the origin it
20//! already holds, holding none included. Every other copy records the id it was copied
21//! from. See [`recorded`] for what stamping a copy-back's own id there costs.
22//!
23//! A destination write is at the user's explicit request, names its destination, goes
24//! through that source's own write interface into that source's own store, and is never
25//! read back to answer a query. That is what makes it a write and not a cache.
26
27use std::collections::BTreeMap;
28
29use onetaskgraph_plugin_api::{
30 Cursor, DependencyEdge, DependencyEndpoint, DependencyKind, Direction, Document, DocumentQuery,
31 ItemKind, ItemWrite, NativeId, Page, PageRequest, Project, ProjectQuery, Repository,
32 SourceError, SourceName, Task, TaskQuery,
33};
34use schemars::JsonSchema;
35use serde::{Deserialize, Serialize};
36use serde_json::Value;
37
38use crate::GlobalId;
39use crate::resolve::ResolvedSource;
40
41use super::fetch::{fits, unrepeated};
42use super::local::ProjectSelector;
43use super::{Engine, EngineError, Filters, LeftBehind, Paging, TaskRequest};
44
45/// A request to copy work into one configured destination.
46#[derive(Debug, Clone)]
47pub struct CopyRequest {
48 /// The qualified items to copy, in the order they were named.
49 pub items: CopyItems,
50 /// What those ids name, and what comes with them.
51 pub scope: CopyScope,
52 /// The configured source to copy into — a source name, never a qualified id.
53 pub destination: SourceName,
54 /// How to re-establish a correspondence the two origin rules cannot find.
55 pub match_by: Option<MatchBy>,
56 /// Whether an origin naming nothing at the destination falls through to the search
57 /// rule instead of refusing.
58 pub recreate: bool,
59 /// Whether to perform every read and no write.
60 pub dry_run: bool,
61}
62
63/// The items one copy names: at least one, because a copy naming none is not a copy.
64///
65/// A newtype rather than a bare `Vec`, for the reason [`Repository`] is one: the empty
66/// list is not a copy of nothing, it is a caller mistake, and a type that can hold it
67/// leaves every reader to decide what it means — a report with no entries, an error, a
68/// silent success. None of those is better than not being able to say it.
69#[derive(Debug, Clone, PartialEq, Eq)]
70pub struct CopyItems(Vec<GlobalId>);
71
72impl CopyItems {
73 /// The items a caller named, or `None` when they named none.
74 #[must_use]
75 pub fn new(items: Vec<GlobalId>) -> Option<Self> {
76 (!items.is_empty()).then_some(Self(items))
77 }
78
79 /// The items, in the order they were named.
80 #[must_use]
81 pub fn as_slice(&self) -> &[GlobalId] {
82 &self.0
83 }
84}
85
86/// What the ids a copy names are, and what travels with them.
87///
88/// One value rather than a kind beside a flag, because three of the four combinations
89/// those two would make are real and the fourth — tasks, with the tasks of each also
90/// copied — means nothing.
91#[derive(Debug, Clone, Copy, PartialEq, Eq)]
92pub enum CopyScope {
93 /// The ids name tasks, and only those tasks are copied.
94 Tasks,
95 /// The ids name projects.
96 Projects {
97 /// Whether the tasks in each project are copied too.
98 tasks: bool,
99 },
100 /// The ids name documents, and only those documents are copied.
101 ///
102 /// Nothing travels with a document: it takes part in no dependency graph, and it holds
103 /// nothing of its own the way a project holds tasks.
104 Documents,
105}
106
107/// The caller-named escape for a correspondence neither origin rule can find.
108///
109/// A person editing Markdown who deletes or corrupts the origin key leaves an item rule 1
110/// cannot use and rule 2 cannot find, and the next copy would create a second item. This
111/// is how that is re-established without hand-editing ids.
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub enum MatchBy {
114 /// Match the item whose title is the same.
115 Title,
116 /// Match the item whose value at this metadata key is the same.
117 Metadata(String),
118}
119
120impl MatchBy {
121 /// The spelling a caller types, `title` or any metadata key.
122 #[must_use]
123 pub fn parse(key: &str) -> Self {
124 if key == "title" {
125 Self::Title
126 } else {
127 Self::Metadata(key.to_owned())
128 }
129 }
130}
131
132/// What a copy did, one entry per item.
133///
134/// The same per-item outcomes reach every consumer: the machine-readable output renders
135/// this, the rendered output renders this, and a Rust caller is handed it.
136#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
137pub struct CopyReport {
138 /// One entry per item the copy considered, in the order it considered them.
139 pub items: Vec<CopyOutcome>,
140}
141
142/// What happened to one item.
143///
144/// `action` and `destination` are one value rather than two fields side by side: an
145/// updated item without a destination id, or an orphan without one, are states this type
146/// must not be able to say — the id *is* what those outcomes are about. The one outcome
147/// that legitimately has none is a dry run that would create, because nothing was
148/// created and there is no id to report.
149#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
150pub struct CopyOutcome {
151 /// The qualified id the item was read from.
152 pub source: GlobalId,
153 /// What happened to it, and where.
154 #[serde(flatten)]
155 pub action: CopyAction,
156}
157
158impl CopyOutcome {
159 /// The qualified id this outcome landed on, when it landed on one.
160 #[must_use]
161 pub fn destination(&self) -> Option<&GlobalId> {
162 self.action.destination()
163 }
164}
165
166/// The four things a copy can do to one item, and the id each of them is about.
167#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
168#[serde(tag = "action", rename_all = "kebab-case")]
169pub enum CopyAction {
170 // llmlint: ignore[names_match_behavior] `created` is Contract D's serialized action
171 // for both a completed create and a dry run that would create; the optional destination
172 // distinguishes those cases, and renaming this public variant would break Rust callers.
173 /// The destination held no counterpart, so one was created.
174 Created {
175 /// The id it was created under, or `null` for a dry run that would have created
176 /// one — there is no id, because nothing was.
177 destination: Option<GlobalId>,
178 },
179 /// The destination held a counterpart and it now reads as the source does.
180 Updated {
181 /// The item that was updated.
182 destination: GlobalId,
183 },
184 /// The destination held a counterpart that already read that way; nothing was written.
185 Unchanged {
186 /// The item that already said it.
187 destination: GlobalId,
188 },
189 /// The destination holds a counterpart the source no longer does. A copy never
190 /// deletes, so it was left exactly as it is.
191 Orphaned {
192 /// The item that was left alone.
193 destination: GlobalId,
194 },
195}
196
197impl CopyAction {
198 /// The qualified id this action is about, when there is one.
199 #[must_use]
200 pub fn destination(&self) -> Option<&GlobalId> {
201 match self {
202 Self::Created { destination } => destination.as_ref(),
203 Self::Updated { destination }
204 | Self::Unchanged { destination }
205 | Self::Orphaned { destination } => Some(destination),
206 }
207 }
208
209 /// The word this action serializes as, taken from its own `Serialize`.
210 ///
211 /// Read back off the wire form rather than written out again in a `match`, for the
212 /// reason `render::wire` gives: a second spelling of `unchanged` would be a second
213 /// place for it to drift from the one a caller reads.
214 #[must_use]
215 pub fn name(&self) -> String {
216 serde_json::to_value(self).expect("a contract enum serialises")["action"]
217 .as_str()
218 .expect("an internally tagged enum carries its tag")
219 .to_owned()
220 }
221}
222
223/// Where one item is going at the destination.
224enum Target {
225 /// Update the destination item with this id, reached by the rule named.
226 Update {
227 /// The destination item this copy updates.
228 id: NativeId,
229 /// Which rule found it.
230 found: Found,
231 },
232 /// Create one.
233 Create,
234}
235
236/// Which of the rules above found the destination item a copy is updating.
237///
238/// The two are the same instruction — update that item — and a different answer about the
239/// origin, which is why the distinction is carried this far rather than dropped where it
240/// is made. See [`recorded`].
241#[derive(Clone, Copy, PartialEq, Eq)]
242enum Found {
243 /// Rule 1: the item being copied already named it, so this copy is a copy-back.
244 Origin,
245 /// Rule 2 or the caller's matching escape: the destination was searched for it.
246 Search,
247}
248
249/// What a scan of the destination is looking for.
250enum Wanted {
251 /// An item recording this qualified id as its origin.
252 Origin(String),
253 /// An item whose title is this.
254 Title(String),
255 /// An item holding this value at this metadata key.
256 Metadata(String, Value),
257}
258
259impl Wanted {
260 /// Whether one destination item is the one being looked for.
261 fn found(&self, title: &str, metadata: &BTreeMap<String, Value>) -> bool {
262 match self {
263 Self::Origin(id) => {
264 metadata.get(GlobalId::ORIGIN_KEY) == Some(&Value::String(id.clone()))
265 }
266 Self::Title(wanted) => title == wanted,
267 Self::Metadata(key, value) => metadata.get(key) == Some(value),
268 }
269 }
270}
271
272/// What the destination held before this copy touched one item.
273///
274/// Read once, in [`Engine::land`], and used three times over: to decide whether the write
275/// would change anything, to repair the item's edges once the rest of the copy has landed,
276/// and — if the copy cannot finish — to put the item back exactly as it was.
277#[derive(Clone)]
278struct Prior {
279 /// The item as the destination held it.
280 item: Item,
281 /// Its forward edges there.
282 edges: Vec<DependencyEdge>,
283}
284
285/// One item that landed with an edge whose far end was not written yet.
286///
287/// Held until every item of the whole copy has landed, because the far end may be in
288/// another project of the same command: a copy of two projects at once is one copied set,
289/// not two, and an edge across them is remapped rather than written as a foreign id.
290struct Deferred {
291 /// The item, as it was read and resolved.
292 item: Planned,
293 /// The destination project it was filed under.
294 filed: Option<NativeId>,
295 /// Where it landed.
296 destination: NativeId,
297 /// What the destination held there before, when it held anything.
298 prior: Option<Prior>,
299}
300
301/// What one item's undo has to do to put the destination back.
302enum Undo {
303 /// The copy created it, so undoing means removing it.
304 Created {
305 /// Which write interface removes it.
306 kind: Level,
307 /// The destination id it was created under.
308 id: NativeId,
309 },
310 /// The copy overwrote something, so undoing means writing that something back.
311 ///
312 /// No `kind` beside the id, unlike the variant above: what was there says which of the
313 /// two write interfaces takes it back, and a second spelling of that could disagree
314 /// with it.
315 Updated {
316 /// The destination id that was overwritten.
317 id: NativeId,
318 /// What was there before.
319 prior: Prior,
320 },
321}
322
323impl Undo {
324 /// The destination id this entry is about.
325 fn id(&self) -> &NativeId {
326 match self {
327 Self::Created { id, .. } | Self::Updated { id, .. } => id,
328 }
329 }
330
331 /// Which of the destination's three write interfaces this entry belongs to.
332 ///
333 /// An id alone does not identify a destination item: nothing stops a destination
334 /// numbering its tasks and its projects in one namespace, and a local-Markdown store
335 /// filing `alpha.md` under both is the ordinary case rather than the contrived one.
336 /// So this pairs with `id` wherever one entry has to be told from another.
337 fn kind(&self) -> Level {
338 match self {
339 Self::Created { kind, .. } => *kind,
340 // Read off what was there, for the reason the variant carries no `kind` of
341 // its own: two spellings of one fact can disagree, and this one cannot.
342 Self::Updated { prior, .. } => prior.item.level(),
343 }
344 }
345}
346
347/// Everything one copy has written, in the order it wrote it, so a copy that cannot finish
348/// can undo its own writes.
349///
350/// This is not state the engine keeps: it lives for the length of one `copy` call and is
351/// dropped with it, so the invariant that nothing of a user's work is written down outside
352/// the plugin that owns it is untouched.
353#[derive(Default)]
354struct Journal {
355 /// One entry per destination item this copy first touched, in that order.
356 entries: Vec<Undo>,
357}
358
359impl Journal {
360 /// Record what has to happen to put one destination item back.
361 ///
362 /// The *first* entry for an id is the one that matters and later ones are dropped: an
363 /// item written twice — once as it lands, once when its edges are repaired — was only
364 /// ever one thing before this copy started, and that is what undoing it restores.
365 fn record(&mut self, entry: Undo) {
366 if self
367 .entries
368 .iter()
369 .any(|held| held.kind() == entry.kind() && held.id() == entry.id())
370 {
371 return;
372 }
373 self.entries.push(entry);
374 }
375}
376
377/// One item, read and resolved, on its way into the destination.
378struct Planned {
379 /// Where it came from.
380 source: GlobalId,
381 /// The item as its source reported it.
382 item: Item,
383 /// Its forward edges, as its source reported them.
384 edges: Vec<DependencyEdge>,
385 /// Where it is going.
386 target: Target,
387}
388
389/// A task, a project or a document, so the copy path is written once.
390#[derive(Clone)]
391enum Item {
392 /// A task.
393 Task(Box<Task>),
394 /// A project.
395 Project(Box<Project>),
396 /// A document.
397 Document(Box<Document>),
398}
399
400impl Item {
401 fn id(&self) -> &NativeId {
402 match self {
403 Self::Task(task) => &task.id,
404 Self::Project(project) => &project.id,
405 Self::Document(document) => &document.id,
406 }
407 }
408
409 fn level(&self) -> Level {
410 match self {
411 Self::Task(_) => Level::Task,
412 Self::Project(_) => Level::Project,
413 Self::Document(_) => Level::Document,
414 }
415 }
416}
417
418/// Which of a destination's three read-and-write interfaces one item belongs to.
419///
420/// Deliberately not [`ItemKind`]: that enum names what a *dependency endpoint* points at,
421/// and the contract gives it no document variant because nothing may point at a document.
422/// This one names which pair of methods reads and writes an item, which is a different
423/// question with a third answer.
424#[derive(Debug, Clone, Copy, PartialEq, Eq)]
425enum Level {
426 /// `get_task`, `write_task`, `delete_task`.
427 Task,
428 /// `get_project`, `write_project`, `delete_project`.
429 Project,
430 /// `get_document`, `write_document`, `delete_document`.
431 Document,
432}
433
434impl Engine {
435 /// Copy every item a request names into one configured destination.
436 ///
437 /// This is the whole of the verb, and the command line drives exactly this: a copy a
438 /// Rust caller makes and a copy typed at a shell are the same call, so the two cannot
439 /// answer the same copy differently.
440 ///
441 /// # Errors
442 ///
443 /// Returns [`EngineError`] when the destination is not configured, cannot be built,
444 /// cannot be written, or — for a document copy — declares it has no documents; when an
445 /// id names nothing; when an origin names an item the
446 /// destination no longer holds and `--recreate` was not given; and when the
447 /// destination refuses the write — including a field or a metadata key it cannot
448 /// carry, which it names rather than dropping.
449 pub async fn copy(&self, request: &CopyRequest) -> Result<CopyReport, EngineError> {
450 let destination = self.writable(&request.destination)?;
451 // Before anything is read, and from the declaration rather than from a failed
452 // write: a destination that says it has no documents has nowhere to put one.
453 if request.scope == CopyScope::Documents {
454 documentary(destination)?;
455 }
456 let mut journal = Journal::default();
457 match self.copy_all(destination, request, &mut journal).await {
458 Ok(report) => Ok(report),
459 Err(error) => Err(self.undo(destination, journal, error).await),
460 }
461 }
462
463 /// The copy itself, with everything it writes recorded so a failure can be undone.
464 ///
465 /// The ids named together are **one** copied set, and that is what makes an edge
466 /// between any two of them a real edge at the destination: a copy of two projects at
467 /// once knows that a task in the first depends on a task in the second, and a task
468 /// knows that the project it belongs to is being created beside it. Copying them one
469 /// at a time could not, and wrote the far end as the id it had at its *source* — a
470 /// dangling reference to somewhere the destination has never heard of.
471 async fn copy_all(
472 &self,
473 destination: &ResolvedSource,
474 request: &CopyRequest,
475 journal: &mut Journal,
476 ) -> Result<CopyReport, EngineError> {
477 // Keyed by the qualified id's own rendering, which is what a recorded origin holds
478 // anyway — making `GlobalId` orderable for one local map would put an ordering on
479 // a contract type for a reason no caller of it has.
480 let mut written: BTreeMap<String, NativeId> = BTreeMap::new();
481 let mut deferred: Vec<Deferred> = Vec::new();
482 // The whole copied set, established before anything is written. For a project
483 // copy that means reading every named project's membership first: the set is the
484 // whole request rather than one project of it.
485 let mut membership = Vec::new();
486 let mut copied = Vec::new();
487 match request.scope {
488 CopyScope::Tasks | CopyScope::Documents => {
489 copied.extend(request.items.as_slice().iter().cloned());
490 }
491 CopyScope::Projects { tasks } => {
492 for id in request.items.as_slice() {
493 let members = if tasks {
494 self.project_members(id).await?
495 } else {
496 Vec::new()
497 };
498 copied.push(id.clone());
499 copied.extend(members.iter().cloned());
500 membership.push((id.clone(), members));
501 }
502 }
503 }
504 let items = match request.scope {
505 CopyScope::Tasks | CopyScope::Documents => {
506 self.copy_items(
507 destination,
508 request,
509 match request.scope {
510 CopyScope::Documents => Level::Document,
511 _ => Level::Task,
512 },
513 request.items.as_slice(),
514 None,
515 &copied,
516 &mut written,
517 &mut deferred,
518 journal,
519 )
520 .await?
521 }
522 CopyScope::Projects { tasks } => {
523 let mut items = Vec::new();
524 for (id, members) in &membership {
525 items.extend(
526 self.copy_project(
527 destination,
528 request,
529 id,
530 members,
531 tasks,
532 &copied,
533 &mut written,
534 &mut deferred,
535 journal,
536 )
537 .await?,
538 );
539 }
540 items
541 }
542 };
543 self.repair(destination, request, &copied, &written, deferred, journal)
544 .await?;
545 Ok(CopyReport { items })
546 }
547
548 /// Write every deferred item again, now that every destination id is known.
549 ///
550 /// This is the second half of the two passes an edge between two items of one copy
551 /// needs: the far end's destination id does not exist until it has been created, so
552 /// the item that points at it lands first without that edge and is completed here.
553 /// It runs once for the whole request rather than once per project, because a far end
554 /// may be in a project this copy has not reached yet.
555 async fn repair(
556 &self,
557 destination: &ResolvedSource,
558 request: &CopyRequest,
559 copied: &[GlobalId],
560 written: &BTreeMap<String, NativeId>,
561 deferred: Vec<Deferred>,
562 journal: &mut Journal,
563 ) -> Result<(), EngineError> {
564 if request.dry_run {
565 return Ok(());
566 }
567 for entry in deferred {
568 let edges = mapped_edges(
569 &entry.item.edges,
570 &entry.item.source.source,
571 destination,
572 copied,
573 written,
574 );
575 self.write(
576 destination,
577 &entry.item,
578 Some(entry.destination),
579 entry.filed,
580 &resolved(&edges),
581 entry.prior,
582 journal,
583 )
584 .await?;
585 }
586 Ok(())
587 }
588
589 /// Put the destination back the way this copy found it, then report why it failed.
590 ///
591 /// Undone in reverse, and an item this copy created is removed rather than restored —
592 /// the entry recording what it looked like a moment after creation is not a state
593 /// anybody asked for. When the destination cannot take one of them back, the refusal
594 /// says so and names what is still there, because a user told "the copy failed" about
595 /// a destination that is not as they left it will copy again over a tree nobody
596 /// described.
597 async fn undo(
598 &self,
599 destination: &ResolvedSource,
600 journal: Journal,
601 error: EngineError,
602 ) -> EngineError {
603 let created: Vec<(Level, &NativeId)> = journal
604 .entries
605 .iter()
606 .filter_map(|entry| match entry {
607 Undo::Created { kind, id } => Some((*kind, id)),
608 Undo::Updated { .. } => None,
609 })
610 .collect();
611 // The ids and the refusal are one value rather than two, because they are one
612 // fact: an item is only left behind because the destination refused to take it
613 // back, so the first refusal carries the first id and neither half can be
614 // recorded without the other.
615 let mut unrestored: Option<(LeftBehind, SourceError)> = None;
616 for entry in journal.entries.iter().rev() {
617 let outcome = match entry {
618 Undo::Created { kind, id } => remove(destination, *kind, id).await,
619 Undo::Updated { id, prior, .. } if !created.contains(&(prior.item.level(), id)) => {
620 restore(destination, id, prior).await
621 }
622 Undo::Updated { .. } => Ok(()),
623 };
624 if let Err(problem) = outcome {
625 let id = GlobalId::new(destination.name().clone(), entry.id().clone());
626 match &mut unrestored {
627 Some((left_behind, _)) => left_behind.push(id),
628 None => unrestored = Some((LeftBehind::new(id), problem)),
629 }
630 }
631 }
632 match unrestored {
633 None => error,
634 Some((left_behind, refusal)) => EngineError::CopyNotUndone {
635 error: Box::new(error),
636 left_behind,
637 refusal,
638 },
639 }
640 }
641
642 /// The destination source, once it is established it exists and can be written.
643 fn writable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
644 let name = self.known(name)?;
645 if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
646 return Err(EngineError::DestinationUnavailable {
647 name: name.to_string(),
648 error: unavailable.error().clone(),
649 });
650 }
651 let source = self
652 .ready()
653 .find(|source| source.name() == &name)
654 .ok_or(EngineError::NoSources)?;
655 if !source.source().writes().is_supported() {
656 return Err(EngineError::NotWritable {
657 name: name.to_string(),
658 kind: source.kind().to_owned(),
659 });
660 }
661 Ok(source)
662 }
663
664 /// Copy one project and, unless they are excluded, every task in it.
665 // llmlint: ignore[suppressions_justified] Five of these are the copy's own running
666 // state — the copied set, the ids written so far, the items held back for repair and
667 // the undo journal — and every one of them is shared across the whole request rather
668 // than per project, which is the defect this signature exists to close. Bundling them
669 // into a context struct would put a lifetime and a borrow split around state that is
670 // threaded through three call sites and read nowhere else.
671 #[allow(clippy::too_many_arguments)]
672 async fn copy_project(
673 &self,
674 destination: &ResolvedSource,
675 request: &CopyRequest,
676 id: &GlobalId,
677 members: &[GlobalId],
678 tasks: bool,
679 copied: &[GlobalId],
680 written: &mut BTreeMap<String, NativeId>,
681 deferred: &mut Vec<Deferred>,
682 journal: &mut Journal,
683 ) -> Result<Vec<CopyOutcome>, EngineError> {
684 // On a repeat copy, compare the project with its final remapped edges before the
685 // first pass temporarily rewrites it. This preserves an `unchanged` outcome when
686 // the project and every copied member already have counterparts.
687 let project_plan = self.plan(destination, request, Level::Project, id).await?;
688 let mut known = BTreeMap::new();
689 if let Target::Update { id: target, .. } = &project_plan.target {
690 known.insert(id.to_string(), target.clone());
691 }
692 for member in members {
693 let member_plan = self.plan(destination, request, Level::Task, member).await?;
694 if let Target::Update { id: target, .. } = member_plan.target {
695 known.insert(member.to_string(), target);
696 }
697 }
698 let project_was_unchanged = if let Target::Update { id: target, .. } = &project_plan.target
699 {
700 let edges = mapped_edges(&project_plan.edges, &id.source, destination, copied, &known);
701 let held = self.prior(destination, Level::Project, target).await?;
702 !edges.iter().any(Option::is_none)
703 && !changes(
704 held.as_ref(),
705 &project_plan,
706 target,
707 &None,
708 &resolved(&edges),
709 )
710 } else {
711 false
712 };
713 let mut outcomes = self
714 .copy_items(
715 destination,
716 request,
717 Level::Project,
718 std::slice::from_ref(id),
719 None,
720 copied,
721 written,
722 deferred,
723 journal,
724 )
725 .await?;
726 if !tasks {
727 return Ok(outcomes);
728 }
729 // `None` when a dry run would have created the project: nothing was written, so
730 // there is no destination project id to file the tasks under. Every task is still
731 // read and still reported, because that is what a dry run is for.
732 let project = outcomes.first().and_then(CopyOutcome::destination).cloned();
733 let task_outcomes = self
734 .copy_items(
735 destination,
736 request,
737 Level::Task,
738 members,
739 project.as_ref().map(|project| project.native.clone()),
740 copied,
741 written,
742 deferred,
743 journal,
744 )
745 .await?;
746 outcomes.extend(task_outcomes);
747 if let Some(project) = project {
748 if project_was_unchanged {
749 outcomes[0].action = CopyAction::Unchanged {
750 destination: project.clone(),
751 };
752 }
753 outcomes.extend(
754 self.orphans(destination, id, &project.native, members)
755 .await?,
756 );
757 }
758 Ok(outcomes)
759 }
760
761 /// Every task the source holds in `project`, by qualified id.
762 async fn project_members(&self, project: &GlobalId) -> Result<Vec<GlobalId>, EngineError> {
763 let mut request = TaskRequest {
764 sources: vec![project.source.clone()],
765 filters: Filters::default(),
766 project: ProjectSelector::Qualified(project.clone()),
767 paging: Paging {
768 limit: PROJECT_PAGE,
769 token: None,
770 },
771 };
772 let mut members = Vec::new();
773 // Pages by this engine's own token rather than by a source cursor, and the
774 // asymmetry with the three walks below is deliberate. A source answering two
775 // cursors with each other advances on every page, so `unrepeated` under the list
776 // verb never fires; the cycle shows only as a token handed back unchanged, which
777 // is what this loop pages by. Point it at the source and a project copy spins.
778 //
779 // No page bound here for the same reason: `Engine::tasks` merges under the budget
780 // it was asked, so nothing longer than `PROJECT_PAGE` can arrive. `fits` in
781 // `fetch::walk` refuses the page a source can really overrun, its own, while
782 // these very members are read.
783 let misbehaved = |error| EngineError::SourceRefused {
784 name: project.source.to_string(),
785 error,
786 };
787 loop {
788 let asked = request.paging.token.clone();
789 let response = self.tasks(&request).await?;
790 if let Some(failure) = response.errors.first() {
791 return Err(EngineError::SourceRefused {
792 name: failure.source.to_string(),
793 error: failure.error.clone(),
794 });
795 }
796 unrepeated(
797 response.next.as_ref(),
798 asked.as_ref(),
799 "the tasks of a project were being read for a copy",
800 )
801 .map_err(misbehaved)?;
802 members.extend(response.items.into_iter().map(|task| task.id));
803 match response.next {
804 Some(token) => request.paging.token = Some(token),
805 None => return Ok(members),
806 }
807 }
808 }
809
810 /// Destination tasks filed under the copied project whose origin the source no longer
811 /// holds.
812 ///
813 /// A copy never deletes, so each is left exactly as it is and reported.
814 async fn orphans(
815 &self,
816 destination: &ResolvedSource,
817 project: &GlobalId,
818 at_destination: &NativeId,
819 copied: &[GlobalId],
820 ) -> Result<Vec<CopyOutcome>, EngineError> {
821 let mut orphans = Vec::new();
822 let mut cursor: Option<Cursor> = None;
823 loop {
824 let asked = cursor.clone();
825 let request = request_for(destination, cursor);
826 let page: Page<Task> = destination
827 .source()
828 .query_tasks(&TaskQuery::default(), &request)
829 .await
830 .map_err(|error| refused(destination, error))?;
831 fits(page.items.len(), request.limit).map_err(|error| refused(destination, error))?;
832 for task in &page.items {
833 if task.project.as_ref() != Some(at_destination) {
834 continue;
835 }
836 let Some(origin) = origin_of(&task.metadata) else {
837 continue;
838 };
839 if origin.source != project.source || copied.contains(&origin) {
840 continue;
841 }
842 orphans.push(CopyOutcome {
843 source: origin,
844 action: CopyAction::Orphaned {
845 destination: GlobalId::new(destination.name().clone(), task.id.clone()),
846 },
847 });
848 }
849 unrepeated(
850 page.next.as_ref(),
851 asked.as_ref(),
852 "the destination was being read for items the copy left behind",
853 )
854 .map_err(|error| refused(destination, error))?;
855 match page.next {
856 Some(next) => cursor = Some(next),
857 None => return Ok(orphans),
858 }
859 }
860 }
861
862 /// Read, resolve and write every item named, holding back the ones whose edges are
863 /// not resolvable yet.
864 ///
865 /// An edge between two items of one copy can point at a member whose destination id
866 /// does not exist until it has been created, so the item that points at it lands
867 /// without that edge and is handed to `deferred`. [`Engine::repair`] finishes it once
868 /// the *whole* request has landed — not once this call has, because the far end may
869 /// be in another project of the same command.
870 // llmlint: ignore[suppressions_justified] The same running state `copy_project` threads,
871 // for the same reason: it belongs to one `copy` call and is shared across every item of
872 // it, and a struct around it would add a borrow split for no reader's benefit.
873 #[allow(clippy::too_many_arguments)]
874 async fn copy_items(
875 &self,
876 destination: &ResolvedSource,
877 request: &CopyRequest,
878 kind: Level,
879 items: &[GlobalId],
880 project: Option<NativeId>,
881 copied: &[GlobalId],
882 written: &mut BTreeMap<String, NativeId>,
883 deferred: &mut Vec<Deferred>,
884 journal: &mut Journal,
885 ) -> Result<Vec<CopyOutcome>, EngineError> {
886 let mut planned = Vec::new();
887 for id in items {
888 planned.push(self.plan(destination, request, kind, id).await?);
889 }
890
891 for item in &planned {
892 if let Target::Update { id, .. } = &item.target {
893 written.insert(item.source.to_string(), id.clone());
894 }
895 }
896
897 // Resolved once per item, and used by both passes: the repair pass writes the
898 // same item again, and re-deriving this there could file it somewhere else.
899 let mut filed = Vec::new();
900 for item in &planned {
901 filed.push(self.filed(destination, item, project.clone()).await?);
902 }
903
904 let mut outcomes = Vec::new();
905 let mut unresolved = Vec::new();
906 let mut priors = Vec::new();
907 for (index, item) in planned.iter().enumerate() {
908 let edges = mapped_edges(
909 &item.edges,
910 &item.source.source,
911 destination,
912 copied,
913 written,
914 );
915 if edges.iter().any(Option::is_none) {
916 unresolved.push(index);
917 }
918 let (outcome, prior) = self
919 .land(
920 destination,
921 request,
922 item,
923 filed[index].clone(),
924 &edges,
925 journal,
926 )
927 .await?;
928 if let Some(id) = outcome.destination() {
929 written.insert(item.source.to_string(), id.native.clone());
930 }
931 outcomes.push(outcome);
932 priors.push(prior);
933 }
934
935 if !request.dry_run {
936 for (index, item) in planned.into_iter().enumerate() {
937 if !unresolved.contains(&index) {
938 continue;
939 }
940 // Every item a copy that is not a dry run lands has a destination id: the
941 // one outcome without one is a dry run that would have created, and this
942 // block does not run for a dry run.
943 let id = outcomes[index]
944 .destination()
945 .expect("a copy that writes lands every item it planned")
946 .clone();
947 deferred.push(Deferred {
948 item,
949 filed: filed[index].clone(),
950 destination: id.native,
951 prior: priors[index].clone(),
952 });
953 }
954 }
955 Ok(outcomes)
956 }
957
958 /// Read one item and its forward edges, and decide where it is going.
959 async fn plan(
960 &self,
961 destination: &ResolvedSource,
962 request: &CopyRequest,
963 kind: Level,
964 id: &GlobalId,
965 ) -> Result<Planned, EngineError> {
966 let source = self.readable(&id.source)?;
967 if kind == Level::Document {
968 documentary(source)?;
969 }
970 let item = match kind {
971 Level::Task => source
972 .source()
973 .get_task(&id.native)
974 .await
975 .map_err(|error| refused(source, error))?
976 .map(|task| Item::Task(Box::new(task))),
977 Level::Project => source
978 .source()
979 .get_project(&id.native)
980 .await
981 .map_err(|error| refused(source, error))?
982 .map(|project| Item::Project(Box::new(project))),
983 Level::Document => source
984 .source()
985 .get_document(&id.native)
986 .await
987 .map_err(|error| refused(source, error))?
988 .map(|document| Item::Document(Box::new(document))),
989 }
990 .ok_or_else(|| EngineError::NoSuchItem { id: id.to_string() })?;
991 let edges = forward_edges(source, &id.native, item.level()).await?;
992 let target = self.target(destination, request, id, &item).await?;
993 Ok(Planned {
994 source: id.clone(),
995 item,
996 edges,
997 target,
998 })
999 }
1000
1001 /// Which destination item this one corresponds to, by the two origin rules and the
1002 /// caller's escape.
1003 async fn target(
1004 &self,
1005 destination: &ResolvedSource,
1006 request: &CopyRequest,
1007 id: &GlobalId,
1008 item: &Item,
1009 ) -> Result<Target, EngineError> {
1010 let (title, metadata) = described(item);
1011 if let Some(origin) = origin_of(metadata)
1012 && &origin.source == destination.name()
1013 {
1014 if exists(destination, &origin.native, item.level()).await? {
1015 return Ok(Target::Update {
1016 id: origin.native,
1017 found: Found::Origin,
1018 });
1019 }
1020 if !request.recreate {
1021 return Err(EngineError::StaleOrigin {
1022 item: id.to_string(),
1023 origin: origin.to_string(),
1024 });
1025 }
1026 }
1027 if let Some(found) = self
1028 .scan(destination, item.level(), &Wanted::Origin(id.to_string()))
1029 .await?
1030 {
1031 return Ok(Target::Update {
1032 id: found,
1033 found: Found::Search,
1034 });
1035 }
1036 let wanted = match &request.match_by {
1037 Some(MatchBy::Title) => Some(Wanted::Title(title.to_owned())),
1038 Some(MatchBy::Metadata(key)) => metadata
1039 .get(key)
1040 .map(|value| Wanted::Metadata(key.clone(), value.clone())),
1041 None => None,
1042 };
1043 if let Some(wanted) = wanted
1044 && let Some(found) = self.scan(destination, item.level(), &wanted).await?
1045 {
1046 return Ok(Target::Update {
1047 id: found,
1048 found: Found::Search,
1049 });
1050 }
1051 Ok(Target::Create)
1052 }
1053
1054 /// Walk the destination one page at a time, looking for `wanted`.
1055 ///
1056 /// One page is held at a time and nothing is written down, which is the same bound
1057 /// every other compensation in this engine works under.
1058 async fn scan(
1059 &self,
1060 destination: &ResolvedSource,
1061 kind: Level,
1062 wanted: &Wanted,
1063 ) -> Result<Option<NativeId>, EngineError> {
1064 let mut cursor: Option<Cursor> = None;
1065 loop {
1066 let asked = cursor.clone();
1067 let request = request_for(destination, cursor);
1068 let next = match kind {
1069 Level::Task => {
1070 let page = destination
1071 .source()
1072 .query_tasks(&TaskQuery::default(), &request)
1073 .await
1074 .map_err(|error| refused(destination, error))?;
1075 fits(page.items.len(), request.limit)
1076 .map_err(|error| refused(destination, error))?;
1077 for task in &page.items {
1078 if wanted.found(&task.title, &task.metadata) {
1079 return Ok(Some(task.id.clone()));
1080 }
1081 }
1082 page.next
1083 }
1084 Level::Project => {
1085 let page = destination
1086 .source()
1087 .query_projects(&ProjectQuery::default(), &request)
1088 .await
1089 .map_err(|error| refused(destination, error))?;
1090 fits(page.items.len(), request.limit)
1091 .map_err(|error| refused(destination, error))?;
1092 for project in &page.items {
1093 if wanted.found(&project.title, &project.metadata) {
1094 return Ok(Some(project.id.clone()));
1095 }
1096 }
1097 page.next
1098 }
1099 Level::Document => {
1100 let page = destination
1101 .source()
1102 .query_documents(&DocumentQuery::default(), &request)
1103 .await
1104 .map_err(|error| refused(destination, error))?;
1105 fits(page.items.len(), request.limit)
1106 .map_err(|error| refused(destination, error))?;
1107 for document in &page.items {
1108 if wanted.found(&document.title, &document.metadata) {
1109 return Ok(Some(document.id.clone()));
1110 }
1111 }
1112 page.next
1113 }
1114 };
1115 unrepeated(
1116 next.as_ref(),
1117 asked.as_ref(),
1118 "the destination was being scanned for the item to update",
1119 )
1120 .map_err(|error| refused(destination, error))?;
1121 match next {
1122 Some(next) => cursor = Some(next),
1123 None => return Ok(None),
1124 }
1125 }
1126 }
1127
1128 /// Write one planned item, or say what a dry run would have done.
1129 ///
1130 /// Answers with what the destination held there beforehand as well, which is what
1131 /// makes an item written twice restorable to what it was rather than to what this
1132 /// copy's first pass left.
1133 async fn land(
1134 &self,
1135 destination: &ResolvedSource,
1136 request: &CopyRequest,
1137 item: &Planned,
1138 project: Option<NativeId>,
1139 edges: &[Option<DependencyEdge>],
1140 journal: &mut Journal,
1141 ) -> Result<(CopyOutcome, Option<Prior>), EngineError> {
1142 let target = match &item.target {
1143 Target::Update { id, .. } => Some(id.clone()),
1144 Target::Create => None,
1145 };
1146 // One read of the destination item, used to decide whether the write changes
1147 // anything and — if the copy cannot finish — to put that item back.
1148 let prior = match &target {
1149 Some(id) => self.prior(destination, item.item.level(), id).await?,
1150 None => None,
1151 };
1152 let edges = resolved(edges);
1153 let qualified = |native: NativeId| GlobalId::new(destination.name().clone(), native);
1154 if let Some(id) = &target
1155 && !changes(prior.as_ref(), item, id, &project, &edges)
1156 {
1157 return Ok((
1158 CopyOutcome {
1159 source: item.source.clone(),
1160 action: CopyAction::Unchanged {
1161 destination: qualified(id.clone()),
1162 },
1163 },
1164 prior,
1165 ));
1166 }
1167 if request.dry_run {
1168 return Ok((
1169 CopyOutcome {
1170 source: item.source.clone(),
1171 action: match target {
1172 Some(id) => CopyAction::Updated {
1173 destination: qualified(id),
1174 },
1175 // Null only here: nothing was created, so there is no id to report.
1176 None => CopyAction::Created { destination: None },
1177 },
1178 },
1179 prior,
1180 ));
1181 }
1182 let updating = target.is_some();
1183 let written = qualified(
1184 self.write(
1185 destination,
1186 item,
1187 target,
1188 project,
1189 &edges,
1190 prior.clone(),
1191 journal,
1192 )
1193 .await?,
1194 );
1195 Ok((
1196 CopyOutcome {
1197 source: item.source.clone(),
1198 action: if updating {
1199 CopyAction::Updated {
1200 destination: written,
1201 }
1202 } else {
1203 CopyAction::Created {
1204 destination: Some(written),
1205 }
1206 },
1207 },
1208 prior,
1209 ))
1210 }
1211
1212 /// Which destination project this item is filed under, when it is filed at all.
1213 ///
1214 /// A task copied as part of a project copy is filed under that project's counterpart,
1215 /// which the copy has just established. A task copied on its own has to find it.
1216 async fn filed(
1217 &self,
1218 destination: &ResolvedSource,
1219 item: &Planned,
1220 project: Option<NativeId>,
1221 ) -> Result<Option<NativeId>, EngineError> {
1222 match (&item.item, project) {
1223 (Item::Task(task), None) => {
1224 self.counterpart(destination, item, task.project.as_ref())
1225 .await
1226 }
1227 (Item::Document(document), None) => {
1228 self.counterpart(destination, item, document.project.as_ref())
1229 .await
1230 }
1231 (Item::Task(_) | Item::Document(_), filed) => Ok(filed),
1232 (Item::Project(_), _) => Ok(None),
1233 }
1234 }
1235
1236 /// The destination project this task's own project corresponds to, when there is one.
1237 ///
1238 /// A task copied on its own keeps its source's project id when the destination holds
1239 /// no counterpart: the field is opaque to this engine, and dropping it would lose
1240 /// what the source said.
1241 async fn counterpart(
1242 &self,
1243 destination: &ResolvedSource,
1244 item: &Planned,
1245 project: Option<&NativeId>,
1246 ) -> Result<Option<NativeId>, EngineError> {
1247 let Some(project) = project else {
1248 return Ok(None);
1249 };
1250 let qualified = GlobalId::new(item.source.source.clone(), project.clone());
1251 let found = self
1252 .scan(
1253 destination,
1254 Level::Project,
1255 &Wanted::Origin(qualified.to_string()),
1256 )
1257 .await?;
1258 Ok(Some(found.unwrap_or_else(|| project.clone())))
1259 }
1260
1261 /// What the destination holds at one id, item and forward edges together.
1262 ///
1263 /// One read for both purposes it serves — deciding whether a write changes anything,
1264 /// and putting the item back if the copy cannot finish — because a second read of the
1265 /// same item is a second round trip against a hosted destination for nothing.
1266 async fn prior(
1267 &self,
1268 destination: &ResolvedSource,
1269 kind: Level,
1270 id: &NativeId,
1271 ) -> Result<Option<Prior>, EngineError> {
1272 let held = match kind {
1273 Level::Task => destination
1274 .source()
1275 .get_task(id)
1276 .await
1277 .map_err(|error| refused(destination, error))?
1278 .map(|task| Item::Task(Box::new(task))),
1279 Level::Project => destination
1280 .source()
1281 .get_project(id)
1282 .await
1283 .map_err(|error| refused(destination, error))?
1284 .map(|project| Item::Project(Box::new(project))),
1285 Level::Document => destination
1286 .source()
1287 .get_document(id)
1288 .await
1289 .map_err(|error| refused(destination, error))?
1290 .map(|document| Item::Document(Box::new(document))),
1291 };
1292 let Some(item) = held else {
1293 return Ok(None);
1294 };
1295 let edges = forward_edges(destination, id, kind).await?;
1296 Ok(Some(Prior { item, edges }))
1297 }
1298
1299 /// Hand one item to the destination's own write interface, recording how to take it
1300 /// back.
1301 // llmlint: ignore[suppressions_justified] A write is the item, where it is going, what
1302 // it is filed under, its edges, what was there before and the journal that records how
1303 // to put it back. Each is a distinct decision made by a different part of the copy, and
1304 // grouping them would only move the argument list to a constructor.
1305 #[allow(clippy::too_many_arguments)]
1306 async fn write(
1307 &self,
1308 destination: &ResolvedSource,
1309 item: &Planned,
1310 target: Option<NativeId>,
1311 project: Option<NativeId>,
1312 edges: &[DependencyEdge],
1313 prior: Option<Prior>,
1314 journal: &mut Journal,
1315 ) -> Result<NativeId, EngineError> {
1316 let created_kind = item.item.level();
1317 let suggested = target.clone().unwrap_or_else(|| item.item.id().clone());
1318 // Settled before the journal takes `prior`, and from that same read: what the
1319 // destination holds at the origin key is what a copy-back leaves there.
1320 let origin = recorded(item, prior.as_ref());
1321 // Recorded *before* the write rather than after it. A destination's own write is
1322 // several calls — `docs/plugin-protocol.md` §4.9 — and one of them failing leaves
1323 // the ones before it applied. No source can put those back, because only this
1324 // journal holds what was there; recorded after a successful write, an update that
1325 // stopped part way was the one way a copy could end and leave the destination
1326 // altered. A restore of an item the write never reached rewrites what is already
1327 // there, which costs one mutation and is what "either complete or it never
1328 // happened" is worth.
1329 if let (Some(id), Some(prior)) = (target.clone(), prior) {
1330 journal.record(Undo::Updated { id, prior });
1331 }
1332 let landed = match outgoing(item, suggested, project, &origin) {
1333 Item::Task(task) => destination
1334 .source()
1335 .write_task(&ItemWrite {
1336 target: target.clone(),
1337 item: *task,
1338 depends_on: edges.to_vec(),
1339 })
1340 .await
1341 .map_err(|error| refused(destination, error))?,
1342 Item::Project(project) => destination
1343 .source()
1344 .write_project(&ItemWrite {
1345 target: target.clone(),
1346 item: *project,
1347 depends_on: edges.to_vec(),
1348 })
1349 .await
1350 .map_err(|error| refused(destination, error))?,
1351 // No edges, and that is the contract: a document takes part in no dependency
1352 // graph, so there is nothing here for `depends_on` to carry.
1353 Item::Document(document) => destination
1354 .source()
1355 .write_document(&ItemWrite {
1356 target: target.clone(),
1357 item: *document,
1358 depends_on: Vec::new(),
1359 })
1360 .await
1361 .map_err(|error| refused(destination, error))?,
1362 };
1363 // A created item can only be journalled here: its id is what the write answers
1364 // with. A create that fails leaves nothing behind — §4.9 makes taking the item
1365 // back the source's own duty, because a write that refused must not leave an item
1366 // nobody asked for.
1367 if target.is_none() {
1368 journal.record(Undo::Created {
1369 kind: created_kind,
1370 id: landed.clone(),
1371 });
1372 }
1373 Ok(landed)
1374 }
1375
1376 /// A configured source that built, for reading an item out of.
1377 fn readable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
1378 let name = self.known(name)?;
1379 if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
1380 return Err(EngineError::SourceRefused {
1381 name: name.to_string(),
1382 error: unavailable.error().clone(),
1383 });
1384 }
1385 self.ready()
1386 .find(|source| source.name() == &name)
1387 .ok_or(EngineError::NoSources)
1388 }
1389}
1390
1391/// How many tasks of a project are read at once while walking it.
1392const PROJECT_PAGE: std::num::NonZeroU32 = std::num::NonZeroU32::new(50).expect("50 is not zero");
1393
1394/// One page request against `source`, at the largest page it will serve.
1395fn request_for(source: &ResolvedSource, cursor: Option<Cursor>) -> PageRequest {
1396 PageRequest {
1397 cursor,
1398 limit: source.source().capabilities().max_page_size.max(1),
1399 }
1400}
1401
1402/// Whether writing this item would change what the destination already holds.
1403///
1404/// A free function over the state already read rather than a method that reads it again:
1405/// the same answer is wanted where the item is landed and where a repeat copy of a project
1406/// decides whether it settled, and a second read there is a second round trip for nothing.
1407fn changes(
1408 held: Option<&Prior>,
1409 item: &Planned,
1410 target: &NativeId,
1411 project: &Option<NativeId>,
1412 edges: &[DependencyEdge],
1413) -> bool {
1414 let Some(held) = held else {
1415 return true;
1416 };
1417 let outgoing = outgoing(
1418 item,
1419 target.clone(),
1420 project.clone(),
1421 &recorded(item, Some(held)),
1422 );
1423 !same(&held.item, &outgoing) || !same_edges(&held.edges, edges)
1424}
1425
1426/// Remove one item this copy created, through the destination's own write interface.
1427async fn remove(
1428 destination: &ResolvedSource,
1429 kind: Level,
1430 id: &NativeId,
1431) -> Result<(), SourceError> {
1432 match kind {
1433 Level::Task => destination.source().delete_task(id).await,
1434 Level::Project => destination.source().delete_project(id).await,
1435 Level::Document => destination.source().delete_document(id).await,
1436 }
1437}
1438
1439/// Write one item back exactly as the destination held it before this copy.
1440async fn restore(
1441 destination: &ResolvedSource,
1442 id: &NativeId,
1443 prior: &Prior,
1444) -> Result<(), SourceError> {
1445 match &prior.item {
1446 Item::Task(task) => destination
1447 .source()
1448 .write_task(&ItemWrite {
1449 target: Some(id.clone()),
1450 item: (**task).clone(),
1451 depends_on: prior.edges.clone(),
1452 })
1453 .await
1454 .map(|_| ()),
1455 Item::Project(project) => destination
1456 .source()
1457 .write_project(&ItemWrite {
1458 target: Some(id.clone()),
1459 item: (**project).clone(),
1460 depends_on: prior.edges.clone(),
1461 })
1462 .await
1463 .map(|_| ()),
1464 Item::Document(document) => destination
1465 .source()
1466 .write_document(&ItemWrite {
1467 target: Some(id.clone()),
1468 item: (**document).clone(),
1469 depends_on: Vec::new(),
1470 })
1471 .await
1472 .map(|_| ()),
1473 }
1474}
1475
1476/// Refuse a document copy addressed to a source that declares it has none.
1477///
1478/// Read off the declaration rather than by asking, which is what "not asked" means: the
1479/// engine learned at the handshake that this source holds no documents, so it refuses
1480/// naming the source and its plugin instead of sending a read that would be refused there.
1481/// Applied at both ends of a copy — a source with no documents holds nothing to copy out,
1482/// and a destination with none has nowhere to put one.
1483fn documentary(source: &ResolvedSource) -> Result<(), EngineError> {
1484 if source.source().capabilities().documents.is_native() {
1485 return Ok(());
1486 }
1487 Err(EngineError::NoDocuments {
1488 name: source.name().to_string(),
1489 kind: source.kind().to_owned(),
1490 })
1491}
1492
1493/// One source failing while a copy was mid-flight.
1494fn refused(source: &ResolvedSource, error: SourceError) -> EngineError {
1495 EngineError::SourceRefused {
1496 name: source.name().to_string(),
1497 error,
1498 }
1499}
1500
1501/// Every forward edge at one item, walked to exhaustion one page at a time.
1502async fn forward_edges(
1503 source: &ResolvedSource,
1504 id: &NativeId,
1505 kind: Level,
1506) -> Result<Vec<DependencyEdge>, EngineError> {
1507 // A document has no edges to walk, and asking for them would mean asking a source for
1508 // a graph the contract says nothing may point into.
1509 if kind == Level::Document {
1510 return Ok(Vec::new());
1511 }
1512 let mut edges = Vec::new();
1513 let mut cursor: Option<Cursor> = None;
1514 loop {
1515 let asked = cursor.clone();
1516 let request = request_for(source, cursor);
1517 let page = match kind {
1518 Level::Task | Level::Document => {
1519 source
1520 .source()
1521 .task_dependencies(id, Direction::DependsOn, &request)
1522 .await
1523 }
1524 Level::Project => {
1525 source
1526 .source()
1527 .project_dependencies(id, Direction::DependsOn, &request)
1528 .await
1529 }
1530 }
1531 .map_err(|error| refused(source, error))?;
1532 fits(page.items.len(), request.limit).map_err(|error| refused(source, error))?;
1533 edges.extend(page.items);
1534 unrepeated(
1535 page.next.as_ref(),
1536 asked.as_ref(),
1537 "an item's dependencies were being read for a copy",
1538 )
1539 .map_err(|error| refused(source, error))?;
1540 match page.next {
1541 Some(next) => cursor = Some(next),
1542 None => return Ok(edges),
1543 }
1544 }
1545}
1546
1547/// The origin one item records, when it records a usable one.
1548fn origin_of(metadata: &BTreeMap<String, Value>) -> Option<GlobalId> {
1549 metadata
1550 .get(GlobalId::ORIGIN_KEY)?
1551 .as_str()?
1552 .parse::<GlobalId>()
1553 .ok()
1554}
1555
1556/// The title and metadata of either kind of item.
1557fn described(item: &Item) -> (&str, &BTreeMap<String, Value>) {
1558 match item {
1559 Item::Task(task) => (&task.title, &task.metadata),
1560 Item::Project(project) => (&project.title, &project.metadata),
1561 Item::Document(document) => (&document.title, &document.metadata),
1562 }
1563}
1564
1565/// The item as the destination should hold it.
1566///
1567/// `url`, `location`, `created_at` and `updated_at` are the destination's own and are
1568/// never written — where the *source* holds an item says nothing about where the
1569/// destination does, which is why a copied document does not arrive claiming the path or
1570/// the link its source reported. The two reserved keys this product encodes typed fields
1571/// under are removed, because those fields travel as themselves — leaving the encoding
1572/// beside them would have the destination hold one thing twice, and disagree with itself
1573/// the moment one changed.
1574fn outgoing(item: &Planned, id: NativeId, project: Option<NativeId>, origin: &Origin) -> Item {
1575 match &item.item {
1576 Item::Task(task) => Item::Task(Box::new(Task {
1577 id,
1578 url: None,
1579 location: None,
1580 created_at: None,
1581 updated_at: None,
1582 project,
1583 metadata: carried(&task.metadata, origin),
1584 ..(**task).clone()
1585 })),
1586 Item::Project(project) => Item::Project(Box::new(Project {
1587 id,
1588 url: None,
1589 location: None,
1590 created_at: None,
1591 updated_at: None,
1592 metadata: carried(&project.metadata, origin),
1593 ..(**project).clone()
1594 })),
1595 Item::Document(document) => Item::Document(Box::new(Document {
1596 id,
1597 url: None,
1598 location: None,
1599 created_at: None,
1600 updated_at: None,
1601 project,
1602 metadata: carried(&document.metadata, origin),
1603 ..(**document).clone()
1604 })),
1605 }
1606}
1607
1608/// The metadata a copy carries: the caller's own keys untouched, and the origin settled.
1609///
1610/// The key is removed before it is settled rather than overwritten, because the item being
1611/// copied carries an origin of its own and [`Origin::Keeps`] must not let it through.
1612fn carried(metadata: &BTreeMap<String, Value>, origin: &Origin) -> BTreeMap<String, Value> {
1613 let mut carried = metadata.clone();
1614 carried.remove(Repository::METADATA_KEY);
1615 carried.remove(DependencyEdge::RECORDED_KEY);
1616 carried.remove(GlobalId::ORIGIN_KEY);
1617 let held = match origin {
1618 Origin::Records(id) => Some(Value::String(id.to_string())),
1619 Origin::Keeps(held) => held.clone(),
1620 };
1621 if let Some(held) = held {
1622 carried.insert(GlobalId::ORIGIN_KEY.to_owned(), held);
1623 }
1624 carried
1625}
1626
1627/// What one landed item records at [`GlobalId::ORIGIN_KEY`].
1628enum Origin {
1629 /// The qualified id this item was copied from, as the id type rather than as its
1630 /// spelling: the key holds a [`GlobalId`] and nothing else may be recorded there.
1631 Records(GlobalId),
1632 /// Whatever the destination already holds there — `None` when it holds nothing, which
1633 /// is written as the key being absent rather than as a null.
1634 ///
1635 /// A [`Value`] and not a [`GlobalId`], because this variant does not interpret what it
1636 /// carries: it is the destination's own metadata entry, held for the length of one
1637 /// write and put back exactly as it was read. Parsing it would turn a value a
1638 /// destination holds and this engine cannot read into a value this engine deletes,
1639 /// which is the opposite of what keeping it means.
1640 Keeps(Option<Value>),
1641}
1642
1643/// Which of the two a copy of this item does.
1644///
1645/// A copy that reached its target by rule 1 is a copy-back: the item being copied names
1646/// the destination item, so the destination is the *original* and the id being copied
1647/// belongs to the copy that came out of it. Recording that id there would overwrite the
1648/// original's own provenance — and with it the correspondence every later copy from the
1649/// source it was authored in depends on. That copy would then match nothing and create a
1650/// second item beside the one it meant to update, which is the whole failure: nothing is
1651/// reported, and whoever reads that board now has two. So a copy-back leaves the
1652/// destination's origin exactly as the destination holds it, absent included, and every
1653/// other copy records the id it was copied from.
1654fn recorded(item: &Planned, held: Option<&Prior>) -> Origin {
1655 if let Target::Update {
1656 found: Found::Origin,
1657 ..
1658 } = &item.target
1659 {
1660 return Origin::Keeps(
1661 held.and_then(|held| described(&held.item).1.get(GlobalId::ORIGIN_KEY).cloned()),
1662 );
1663 }
1664 Origin::Records(item.source.clone())
1665}
1666
1667/// Whether the destination already reads exactly as this copy would leave it.
1668///
1669/// The destination's own `url` and timestamps are excluded because a copy never writes
1670/// them, so a difference there is not one this copy would close.
1671fn same(held: &Item, outgoing: &Item) -> bool {
1672 match (held, outgoing) {
1673 (Item::Task(held), Item::Task(outgoing)) => {
1674 held.title == outgoing.title
1675 && held.content == outgoing.content
1676 && held.status == outgoing.status
1677 && held.labels == outgoing.labels
1678 && held.project == outgoing.project
1679 && held.metadata == outgoing.metadata
1680 && held.repositories == outgoing.repositories
1681 }
1682 (Item::Project(held), Item::Project(outgoing)) => {
1683 held.title == outgoing.title
1684 && held.content == outgoing.content
1685 && held.status == outgoing.status
1686 && held.labels == outgoing.labels
1687 && held.metadata == outgoing.metadata
1688 && held.repositories == outgoing.repositories
1689 }
1690 // No status, because a document has none; no edges, because it is in no graph.
1691 (Item::Document(held), Item::Document(outgoing)) => {
1692 held.title == outgoing.title
1693 && held.content == outgoing.content
1694 && held.labels == outgoing.labels
1695 && held.project == outgoing.project
1696 && held.metadata == outgoing.metadata
1697 && held.repositories == outgoing.repositories
1698 }
1699 _ => false,
1700 }
1701}
1702
1703/// Whether the destination's forward edges already say what this copy would write.
1704fn same_edges(held: &[DependencyEdge], outgoing: &[DependencyEdge]) -> bool {
1705 let ends = |edges: &[DependencyEdge]| {
1706 let mut ends: Vec<(String, ItemKind, DependencyKind)> = edges
1707 .iter()
1708 .map(|edge| (edge.to.id().to_owned(), edge.to.kind, edge.kind))
1709 .collect();
1710 ends.sort_by(|left, right| left.0.cmp(&right.0));
1711 ends
1712 };
1713 ends(held) == ends(outgoing)
1714}
1715
1716/// Each read edge as the destination should record it, or `None` when its far end is a
1717/// member of this copy whose destination id is not known yet.
1718fn mapped_edges(
1719 edges: &[DependencyEdge],
1720 origin: &SourceName,
1721 destination: &ResolvedSource,
1722 copied: &[GlobalId],
1723 written: &BTreeMap<String, NativeId>,
1724) -> Vec<Option<DependencyEdge>> {
1725 edges
1726 .iter()
1727 .map(|edge| {
1728 let far = GlobalId::new(origin.clone(), NativeId(edge.to.id().to_owned()));
1729 let id = if let Some(native) = names(&edge.to, destination.name()) {
1730 // A far end already qualified to the destination's own source is that
1731 // source's own item, so it is written the way that source names its own:
1732 // unqualified. Leaving it qualified would have the destination hold an
1733 // edge into itself written as if it left, which is the one spelling the
1734 // reserved key exists to keep for edges that really do.
1735 Some(native)
1736 } else if edge.to.is_qualified() || origin == destination.name() {
1737 // Already naming a source of its own, or a copy inside one source where
1738 // the far end's own id is the destination's id.
1739 Some(edge.to.id().to_owned())
1740 } else if copied.contains(&far) {
1741 written.get(&far.to_string()).map(|native| native.0.clone())
1742 } else {
1743 Some(far.to_string())
1744 }?;
1745 DependencyEndpoint::new(id, edge.to.kind)
1746 .ok()
1747 .map(|to| DependencyEdge {
1748 from: edge.from.clone(),
1749 to,
1750 kind: edge.kind,
1751 })
1752 })
1753 .collect()
1754}
1755
1756/// The native id a qualified endpoint names at `destination`, when it names one there.
1757fn names(endpoint: &DependencyEndpoint, destination: &SourceName) -> Option<String> {
1758 if !endpoint.is_qualified() {
1759 return None;
1760 }
1761 let id: GlobalId = endpoint.id().parse().ok()?;
1762 (&id.source == destination).then_some(id.native.0)
1763}
1764
1765/// The edges that could be resolved, which is every one of them on the second pass.
1766fn resolved(edges: &[Option<DependencyEdge>]) -> Vec<DependencyEdge> {
1767 edges.iter().flatten().cloned().collect()
1768}
1769
1770/// Whether the destination holds an item with this id.
1771async fn exists(
1772 destination: &ResolvedSource,
1773 id: &NativeId,
1774 kind: Level,
1775) -> Result<bool, EngineError> {
1776 let found = match kind {
1777 Level::Task => destination
1778 .source()
1779 .get_task(id)
1780 .await
1781 .map_err(|error| refused(destination, error))?
1782 .is_some(),
1783 Level::Project => destination
1784 .source()
1785 .get_project(id)
1786 .await
1787 .map_err(|error| refused(destination, error))?
1788 .is_some(),
1789 Level::Document => destination
1790 .source()
1791 .get_document(id)
1792 .await
1793 .map_err(|error| refused(destination, error))?
1794 .is_some(),
1795 };
1796 Ok(found)
1797}