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