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