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, BTreeSet};
28
29use onetaskgraph_plugin_api::{
30 Cursor, DependencyEdge, DependencyEndpoint, DependencyKind, Direction, Document, DocumentQuery,
31 ItemKind, ItemWrite, Location, Metering, NativeId, Page, PageRequest, Project, ProjectQuery,
32 Repository, SourceError, SourceName, StatusCategory, Task, TaskQuery, TaskRef,
33};
34use schemars::JsonSchema;
35use serde::{Deserialize, Serialize};
36use serde_json::Value;
37
38use crate::GlobalId;
39use crate::resolve::ResolvedSource;
40
41use super::delivery::{Delivered, targets};
42use super::fetch::{fits, unrepeated};
43use super::local::ProjectSelector;
44use super::narrow::holds_priority;
45use super::{
46 DocumentFilters, DocumentRequest, Engine, EngineError, Filters, LeftBehind, Paging, Qualified,
47 TaskRequest,
48};
49
50/// A request to copy work into one configured destination.
51#[derive(Debug, Clone)]
52pub struct CopyRequest {
53 /// The qualified items to copy, in the order they were named.
54 pub items: CopyItems,
55 /// What those ids name, and what comes with them.
56 pub scope: CopyScope,
57 /// The configured source to copy into — a source name, never a qualified id.
58 pub destination: SourceName,
59 /// How to re-establish a correspondence the two origin rules cannot find.
60 pub match_by: Option<MatchBy>,
61 /// Whether an origin naming nothing at the destination falls through to the search
62 /// rule instead of refusing.
63 pub recreate: bool,
64 /// Whether to perform every read and no write.
65 pub dry_run: bool,
66}
67
68/// The items one copy names: at least one, because a copy naming none is not a copy.
69///
70/// A newtype rather than a bare `Vec`, for the reason [`Repository`] is one: the empty
71/// list is not a copy of nothing, it is a caller mistake, and a type that can hold it
72/// leaves every reader to decide what it means — a report with no entries, an error, a
73/// silent success. None of those is better than not being able to say it.
74#[derive(Debug, Clone, PartialEq, Eq)]
75pub struct CopyItems(Vec<GlobalId>);
76
77impl CopyItems {
78 /// The items a caller named, or `None` when they named none.
79 #[must_use]
80 pub fn new(items: Vec<GlobalId>) -> Option<Self> {
81 (!items.is_empty()).then_some(Self(items))
82 }
83
84 /// The items, in the order they were named.
85 #[must_use]
86 pub fn as_slice(&self) -> &[GlobalId] {
87 &self.0
88 }
89}
90
91/// What the ids a copy names are, and what travels with them.
92///
93/// One value rather than a kind beside a flag, because three of the four combinations
94/// those two would make are real and the fourth — tasks, with the tasks of each also
95/// copied — means nothing. The member list is a variant for the same reason: it narrows a
96/// project copy that carries its tasks, and it has nothing to say to the other three.
97#[derive(Debug, Clone, PartialEq, Eq)]
98pub enum CopyScope {
99 /// The ids name tasks, and only those tasks are copied.
100 Tasks,
101 /// The ids name projects.
102 Projects {
103 /// Whether the tasks in each project are copied too.
104 tasks: bool,
105 },
106 /// The ids name projects, and of the tasks in them exactly these are copied.
107 ///
108 /// A member the list does not name is not read at the destination, not compared, not
109 /// written and not reported, and no walk for what the copy left behind runs. A named
110 /// member with no counterpart there is still created: the list narrows which members
111 /// are read and written, never which outcomes are possible.
112 ///
113 /// An edge from a copied item to a member the list does not name resolves to the
114 /// destination id that member's own [`GlobalId::ORIGIN_KEY`] records at the source,
115 /// without reading the destination for it. When that member records none, the copy is
116 /// refused before anything is written.
117 Members(CopyItems),
118 /// The ids name documents, and only those documents are copied.
119 ///
120 /// Nothing travels with a document: it takes part in no dependency graph, and it holds
121 /// nothing of its own the way a project holds tasks.
122 Documents,
123}
124
125/// The caller-named escape for a correspondence neither origin rule can find.
126///
127/// A person editing Markdown who deletes or corrupts the origin key leaves an item rule 1
128/// cannot use and rule 2 cannot find, and the next copy would create a second item. This
129/// is how that is re-established without hand-editing ids.
130#[derive(Debug, Clone, PartialEq, Eq)]
131pub enum MatchBy {
132 /// Match the item whose title is the same.
133 Title,
134 /// Match the item whose value at this metadata key is the same.
135 Metadata(String),
136}
137
138impl MatchBy {
139 /// The spelling a caller types, `title` or any metadata key.
140 #[must_use]
141 pub fn parse(key: &str) -> Self {
142 if key == "title" {
143 Self::Title
144 } else {
145 Self::Metadata(key.to_owned())
146 }
147 }
148}
149
150/// What a copy did, one entry per item.
151///
152/// The same per-item outcomes reach every consumer: the machine-readable output renders
153/// this, the rendered output renders this, and a Rust caller is handed it.
154#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
155pub struct CopyReport {
156 /// One entry per item the copy considered, in the order it considered them.
157 pub items: Vec<CopyOutcome>,
158 // Three flat scalars rather than one nested object, and a `!skip_serializing_if` beside
159 // every skip: both are load-bearing for the generated SDKs rather than matters of
160 // taste, and AGENTS.md's note on what a copied document's references are pointed at is
161 // where that reasoning lives.
162 /// Reference occurrences the copy rewrote to the destination's own location for the
163 /// record they name.
164 ///
165 /// A silent bound is indistinguishable from a bug, so the copy says what it did to the
166 /// references the documents it carried hold. This and the two below are totals over the
167 /// whole invocation rather than figures per document, and all three default to zero, so
168 /// a consumer written against the output before they existed is unaffected.
169 ///
170 /// **What these figures do not claim.** The referent set is a document's own project,
171 /// so a reference to a record in a *different* project is never recognised at all and
172 /// cannot appear in [`Self::references_unresolved`] either. These are the references
173 /// the copy recognised; they are not a census of every reference a document holds.
174 /// Noticing an out-of-scope reference would need exactly the unbounded destination walk
175 /// this design refuses.
176 #[serde(default, skip_serializing_if = "nothing_to_report")]
177 #[schemars(!skip_serializing_if)]
178 pub references_rewritten: u64,
179 /// Reference occurrences the copy recognised and left byte-for-byte as they were,
180 /// because the correspondence could not be established.
181 #[serde(default, skip_serializing_if = "nothing_to_report")]
182 #[schemars(!skip_serializing_if)]
183 pub references_unresolved: u64,
184 /// How many of [`Self::references_unresolved`] were left alone because the
185 /// correspondence was **ambiguous** rather than merely absent. A sub-count, never
186 /// larger than it.
187 ///
188 /// Split out because the two mean different things to a reader. A reference with no
189 /// counterpart is ordinary and expected under the bound above — the design working. An
190 /// ambiguous one says the destination holds duplicate records for one work item, or the
191 /// source reports one location for two records, and re-running the copy will never
192 /// clear it.
193 #[serde(default, skip_serializing_if = "nothing_to_report")]
194 #[schemars(!skip_serializing_if)]
195 // llmlint: ignore[invalid_states_unrepresentable] JSON Schema cannot express an
196 // inequality between two numbers, so a private constructor here would hold this in one
197 // consumer of three while both SDKs' generated models went on admitting it. What holds
198 // it is `substitute`: `Resolution` has no variant that counts an occurrence ambiguous
199 // without counting it unresolved.
200 pub references_ambiguous: u64,
201 /// `delivers` entries the copy rewrote to the destination's own id for a member of the
202 /// copied set, over the whole invocation. Every other entry arrives qualified and is not
203 /// counted. Left out when zero, as the three figures above are.
204 #[serde(default, skip_serializing_if = "nothing_to_report")]
205 #[schemars(!skip_serializing_if)]
206 pub delivers_rewritten: u64,
207 /// One entry per delivered task the copy kept in step with a task it landed, after the
208 /// whole copy was complete — see `task status set`, which reports the same entries. Left
209 /// out when there were none.
210 ///
211 /// A failed entry does not undo the copy: the tasks it landed stay landed, and the
212 /// command exits `4`.
213 #[serde(default, skip_serializing_if = "Vec::is_empty")]
214 #[schemars(!skip_serializing_if)]
215 pub delivered: Vec<Delivered>,
216 /// What this copy spent, summed over the sources in the command that meter their own
217 /// requests — and absent, never zero, when none of them does.
218 ///
219 /// Omitted from the wire when absent, like the three figures above, so a consumer
220 /// written against the output before it existed reads the same document.
221 #[serde(default, skip_serializing_if = "Option::is_none")]
222 pub spent: Option<Spent>,
223}
224
225/// Whether one of [`CopyReport`]'s reference figures has anything to say.
226///
227/// A copy that recognised no reference reports that by leaving the figure out rather than
228/// by writing a nought, so the machine output of a task or project copy is exactly what it
229/// was before these figures existed. The human rendering says it in words either way,
230/// because a reader there needs to be told the copy looked.
231fn nothing_to_report(figure: &u64) -> bool {
232 *figure == 0
233}
234
235/// What one command spent, summed over the sources in it that meter their own requests.
236///
237/// **Source-owned.** Every figure is what a source said it sent and spent while the command
238/// ran, read through [`TaskSource::metering`](onetaskgraph_plugin_api::TaskSource::metering)
239/// before the command and again after it. The engine adds the differences up by name and
240/// interprets none of them, which is why the budget and unit names are open vocabulary.
241#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
242pub struct Spent {
243 /// How many HTTP requests those sources sent for this command.
244 pub requests: u64,
245 /// What those requests spent, one entry per budget and unit, ordered by budget name.
246 pub budgets: Vec<BudgetSpent>,
247}
248
249/// What one command spent against one budget.
250#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
251pub struct BudgetSpent {
252 /// The budget, as the source names it — `graphql`, `rest`.
253 // llmlint: ignore[invalid_states_unrepresentable] The contract this report lands states `budget` and `unit` as strings in an open vocabulary the engine interprets none of, and both SDKs are generated from this type's schema; the one value such a vocabulary can refuse, the empty name, never reaches here, because `difference` below treats a source reporting one as not metering.
254 pub budget: String,
255 /// What it is metered in — `points`, `requests`.
256 // llmlint: ignore[invalid_states_unrepresentable] As `budget` above.
257 pub unit: String,
258 /// How much was spent against it, in that unit.
259 pub amount: u64,
260 /// Whether any part of `amount` was modelled by a source rather than reported by its
261 /// backend or counted, which makes `amount` a lower bound on what the backend charged
262 /// rather than a measurement of it.
263 pub lower_bound: bool,
264}
265
266/// One document's reference figures, before they are folded into the invocation's.
267#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
268struct Counted {
269 /// Occurrences rewritten.
270 rewritten: u64,
271 /// Occurrences recognised and left alone.
272 unresolved: u64,
273 /// How many of those were ambiguous.
274 ambiguous: u64,
275}
276
277impl Counted {
278 /// Fold one document's figures into the invocation's.
279 fn add(&mut self, other: Self) {
280 self.rewritten += other.rewritten;
281 self.unresolved += other.unresolved;
282 self.ambiguous += other.ambiguous;
283 }
284}
285
286/// What happened to one item.
287///
288/// `action` and `destination` are one value rather than two fields side by side: an
289/// updated item without a destination id, or an orphan without one, are states this type
290/// must not be able to say — the id *is* what those outcomes are about. The one outcome
291/// that legitimately has none is a dry run that would create, because nothing was
292/// created and there is no id to report.
293#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
294pub struct CopyOutcome {
295 /// The qualified id the item was read from.
296 pub source: GlobalId,
297 /// What happened to it, and where.
298 #[serde(flatten)]
299 pub action: CopyAction,
300}
301
302impl CopyOutcome {
303 /// The qualified id this outcome landed on, when it landed on one.
304 #[must_use]
305 pub fn destination(&self) -> Option<&GlobalId> {
306 self.action.destination()
307 }
308}
309
310/// The four things a copy can do to one item, and the id each of them is about.
311#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
312#[serde(tag = "action", rename_all = "kebab-case")]
313pub enum CopyAction {
314 // llmlint: ignore[names_match_behavior] `created` is Contract D's serialized action
315 // for both a completed create and a dry run that would create; the optional destination
316 // distinguishes those cases, and renaming this public variant would break Rust callers.
317 /// The destination held no counterpart, so one was created.
318 Created {
319 /// The id it was created under, or `null` for a dry run that would have created
320 /// one — there is no id, because nothing was.
321 destination: Option<GlobalId>,
322 },
323 /// The destination held a counterpart and it now reads as the source does.
324 Updated {
325 /// The item that was updated.
326 destination: GlobalId,
327 },
328 /// The destination held a counterpart that already read that way; nothing was written.
329 Unchanged {
330 /// The item that already said it.
331 destination: GlobalId,
332 },
333 /// The destination holds a counterpart the source no longer does. A copy never
334 /// deletes, so it was left exactly as it is.
335 Orphaned {
336 /// The item that was left alone.
337 destination: GlobalId,
338 },
339}
340
341impl CopyAction {
342 /// The qualified id this action is about, when there is one.
343 #[must_use]
344 pub fn destination(&self) -> Option<&GlobalId> {
345 match self {
346 Self::Created { destination } => destination.as_ref(),
347 Self::Updated { destination }
348 | Self::Unchanged { destination }
349 | Self::Orphaned { destination } => Some(destination),
350 }
351 }
352
353 /// The word this action serializes as, taken from its own `Serialize`.
354 ///
355 /// Read back off the wire form rather than written out again in a `match`, for the
356 /// reason `render::wire` gives: a second spelling of `unchanged` would be a second
357 /// place for it to drift from the one a caller reads.
358 #[must_use]
359 pub fn name(&self) -> String {
360 serde_json::to_value(self).expect("a contract enum serialises")["action"]
361 .as_str()
362 .expect("an internally tagged enum carries its tag")
363 .to_owned()
364 }
365}
366
367/// What one item points at, resolved as far as the copy has got: its forward edges and the
368/// tasks it delivers.
369struct Pointing<'a> {
370 /// Its forward edges, `None` where the far end is a member not landed yet.
371 edges: &'a [Option<DependencyEdge>],
372 /// The tasks it delivers that can already be named at the destination.
373 delivers: &'a [TaskRef],
374}
375
376/// Where one item is going at the destination.
377enum Target {
378 /// Update the destination item with this id, reached by the rule named.
379 Update {
380 /// The destination item this copy updates.
381 id: NativeId,
382 /// Which rule found it.
383 found: Found,
384 },
385 /// Create one.
386 Create,
387}
388
389/// Which of the rules above found the destination item a copy is updating.
390///
391/// The two are the same instruction — update that item — and a different answer about the
392/// origin, which is why the distinction is carried this far rather than dropped where it
393/// is made. See [`recorded`].
394#[derive(Clone, Copy, PartialEq, Eq)]
395enum Found {
396 /// Rule 1: the item being copied already named it, so this copy is a copy-back.
397 Origin,
398 /// Rule 2 or the caller's matching escape: the destination was searched for it.
399 Search,
400}
401
402/// What a scan of the destination is looking for.
403enum Wanted {
404 /// An item recording this qualified id as its origin.
405 Origin(String),
406 /// An item whose title is this.
407 Title(String),
408 /// An item holding this value at this metadata key.
409 Metadata(String, Value),
410}
411
412impl Wanted {
413 /// Whether one destination item is the one being looked for.
414 fn found(&self, title: &str, metadata: &BTreeMap<String, Value>) -> bool {
415 match self {
416 Self::Origin(id) => {
417 metadata.get(GlobalId::ORIGIN_KEY) == Some(&Value::String(id.clone()))
418 }
419 Self::Title(wanted) => title == wanted,
420 Self::Metadata(key, value) => metadata.get(key) == Some(value),
421 }
422 }
423}
424
425/// What the destination held before this copy touched one item.
426///
427/// Read once, in [`Engine::land`], and used three times over: to decide whether the write
428/// would change anything, to repair the item's edges once the rest of the copy has landed,
429/// and — if the copy cannot finish — to put the item back exactly as it was.
430#[derive(Clone)]
431struct Prior {
432 /// The item as the destination held it.
433 item: Item,
434 /// Its forward edges there.
435 edges: Vec<DependencyEdge>,
436}
437
438/// One item that landed with an edge whose far end was not written yet.
439///
440/// Held until every item of the whole copy has landed, because the far end may be in
441/// another project of the same command: a copy of two projects at once is one copied set,
442/// not two, and an edge across them is remapped rather than written as a foreign id.
443struct Deferred {
444 /// The item, as it was read and resolved.
445 item: Planned,
446 /// The destination project it was filed under.
447 filed: Option<NativeId>,
448 /// Where it landed.
449 destination: NativeId,
450 /// What the destination held there before, when it held anything.
451 prior: Option<Prior>,
452}
453
454/// What one item's undo has to do to put the destination back.
455enum Undo {
456 /// The copy created it, so undoing means removing it.
457 Created {
458 /// Which write interface removes it.
459 kind: Level,
460 /// The destination id it was created under.
461 id: NativeId,
462 },
463 /// The copy overwrote something, so undoing means writing that something back.
464 ///
465 /// No `kind` beside the id, unlike the variant above: what was there says which of the
466 /// two write interfaces takes it back, and a second spelling of that could disagree
467 /// with it.
468 Updated {
469 /// The destination id that was overwritten.
470 id: NativeId,
471 /// What was there before.
472 prior: Prior,
473 },
474}
475
476impl Undo {
477 /// The destination id this entry is about.
478 fn id(&self) -> &NativeId {
479 match self {
480 Self::Created { id, .. } | Self::Updated { id, .. } => id,
481 }
482 }
483
484 /// Which of the destination's three write interfaces this entry belongs to.
485 ///
486 /// An id alone does not identify a destination item: nothing stops a destination
487 /// numbering its tasks and its projects in one namespace, and a local-Markdown store
488 /// filing `alpha.md` under both is the ordinary case rather than the contrived one.
489 /// So this pairs with `id` wherever one entry has to be told from another.
490 fn kind(&self) -> Level {
491 match self {
492 Self::Created { kind, .. } => *kind,
493 // Read off what was there, for the reason the variant carries no `kind` of
494 // its own: two spellings of one fact can disagree, and this one cannot.
495 Self::Updated { prior, .. } => prior.item.level(),
496 }
497 }
498}
499
500/// Everything one copy has written, in the order it wrote it, so a copy that cannot finish
501/// can undo its own writes.
502///
503/// This is not state the engine keeps: it lives for the length of one `copy` call and is
504/// dropped with it, so the invariant that nothing of a user's work is written down outside
505/// the plugin that owns it is untouched.
506#[derive(Default)]
507struct Journal {
508 /// One entry per destination item this copy first touched, in that order.
509 entries: Vec<Undo>,
510}
511
512impl Journal {
513 /// Record what has to happen to put one destination item back.
514 ///
515 /// The *first* entry for an id is the one that matters and later ones are dropped: an
516 /// item written twice — once as it lands, once when its edges are repaired — was only
517 /// ever one thing before this copy started, and that is what undoing it restores.
518 fn record(&mut self, entry: Undo) {
519 if self
520 .entries
521 .iter()
522 .any(|held| held.kind() == entry.kind() && held.id() == entry.id())
523 {
524 return;
525 }
526 self.entries.push(entry);
527 }
528}
529
530/// What one copy invocation carries from the item it lands to the next.
531///
532/// Like the [`Journal`] beside it, this lives for the length of one `copy` call and is
533/// dropped with it: nothing here is state the engine keeps, and nothing in it is read back
534/// to answer a later query.
535#[derive(Default)]
536struct Running {
537 /// Every item an edge of this command resolves to a destination id rather than writing as
538 /// it was read: the ids named, what travels with them, and — for a member copy — each
539 /// member the copy does not carry whose own recorded origin already names its
540 /// destination item. That last kind is resolvable without being copied.
541 resolvable: Vec<GlobalId>,
542 /// The destination item each of those corresponds to, as soon as it is known — whether
543 /// this command found it, landed it, or read it off a member's recorded origin.
544 ///
545 /// Keyed by the qualified id's own rendering, which is what a recorded origin holds
546 /// anyway — making `GlobalId` orderable for a local map would put an ordering on a
547 /// contract type for a reason no caller of it has.
548 counterparts: BTreeMap<String, NativeId>,
549 /// The items held back for [`Engine::repair`], across the whole request.
550 deferred: Vec<Deferred>,
551 /// The reference figures, one total for the whole invocation rather than one per
552 /// document.
553 references: Counted,
554 /// The destination project each source project corresponds to, once this command has
555 /// looked for it — so a second task filed under the same project does not walk the
556 /// destination for it again.
557 filings: BTreeMap<String, Option<NativeId>>,
558 /// `delivers` entries naming a member of the copied set, over the whole invocation.
559 delivers_rewritten: u64,
560 /// Every task this copy landed, for the relation rule once the whole copy is complete.
561 landed: Vec<LandedTask>,
562}
563
564/// One task a copy landed, and what the relation rule reads of it once the copy is complete.
565struct LandedTask {
566 /// Where it landed.
567 destination: GlobalId,
568 /// The source it was read from, which a bare `delivers` entry names a task of.
569 origin: SourceName,
570 /// Its `delivers`, as its source reported them.
571 delivers: Vec<TaskRef>,
572 /// The `delivers` the destination held there before this copy, qualified.
573 before: Vec<GlobalId>,
574 /// Its status category, as the copy wrote it.
575 category: StatusCategory,
576}
577
578/// A task a copy landed, with the tasks it delivers now and delivered before, qualified.
579struct Deliverer {
580 destination: GlobalId,
581 category: StatusCategory,
582 now: Vec<GlobalId>,
583 before: Vec<GlobalId>,
584}
585
586/// One item, read and resolved, on its way into the destination.
587struct Planned {
588 /// Where it came from.
589 source: GlobalId,
590 /// The item as its source reported it.
591 item: Item,
592 /// Its forward edges, as its source reported them.
593 edges: Vec<DependencyEdge>,
594 /// Where it is going.
595 target: Target,
596 /// What the destination held at that target, read once where the target was found.
597 ///
598 /// The read that decided an item exists is the read of what it holds, so the two are
599 /// one round trip rather than two against a hosted destination.
600 held: Option<Prior>,
601}
602
603/// A task, a project or a document, so the copy path is written once.
604#[derive(Clone)]
605enum Item {
606 /// A task.
607 Task(Box<Task>),
608 /// A project.
609 Project(Box<Project>),
610 /// A document.
611 Document(Box<Document>),
612}
613
614impl Item {
615 fn id(&self) -> &NativeId {
616 match self {
617 Self::Task(task) => &task.id,
618 Self::Project(project) => &project.id,
619 Self::Document(document) => &document.id,
620 }
621 }
622
623 fn level(&self) -> Level {
624 match self {
625 Self::Task(_) => Level::Task,
626 Self::Project(_) => Level::Project,
627 Self::Document(_) => Level::Document,
628 }
629 }
630}
631
632/// Which of a destination's three read-and-write interfaces one item belongs to.
633///
634/// Deliberately not [`ItemKind`]: that enum names what a *dependency endpoint* points at,
635/// and the contract gives it no document variant because nothing may point at a document.
636/// This one names which pair of methods reads and writes an item, which is a different
637/// question with a third answer.
638///
639/// Ordered so it can key a map of what a destination holds, per interface: an id alone
640/// does not identify a destination item, for the reason [`Undo::kind`] records.
641#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
642enum Level {
643 /// `get_task`, `write_task`, `delete_task`.
644 Task,
645 /// `get_project`, `write_project`, `delete_project`.
646 Project,
647 /// `get_document`, `write_document`, `delete_document`.
648 Document,
649}
650
651/// One record filed under a document's own project, and the location string its source
652/// reports for it.
653///
654/// A reference is a **literal occurrence, in a document's content, of the exact location
655/// string the source reports for a related record** — the `String` inside
656/// [`Location::Path`] or [`Location::Url`]. Both ends of a rewrite come from the plugins'
657/// own reported [`Location`]: nothing here composes an address out of a name, an id or a
658/// root, because a source that reports a canonical absolute path and a source that reports
659/// an issue link are the two things the contract lets this ask about.
660#[derive(Clone)]
661struct Referent {
662 /// Its qualified id at the source.
663 id: GlobalId,
664 /// The origin it records in its own metadata, when it records one.
665 origin: Option<GlobalId>,
666 /// Which of the destination's three interfaces its counterpart would be read from.
667 level: Level,
668 /// The non-empty location string its source reports for it.
669 location: String,
670}
671
672impl Referent {
673 /// The two keys a destination record's own recorded origin is matched against.
674 ///
675 /// **A destination record is this referent's counterpart when its
676 /// [`GlobalId::ORIGIN_KEY`] equals either this referent's own qualified source id, or
677 /// the origin this referent itself records.** In plainer terms: when the destination
678 /// record was copied directly from this referent, or directly from the one predecessor
679 /// this referent itself records.
680 ///
681 /// That reach is exactly one recorded hop of ancestry on each side, and nothing more:
682 ///
683 /// - **A single hop resolves.** The destination record came straight from the referent.
684 /// - **A one-level fan-out resolves.** A record copied from one store into two, with
685 /// the document arriving by one route and naming records that arrived by the other,
686 /// so both sides trace to one common predecessor. That is what the second key buys,
687 /// and it is the only thing it buys.
688 /// - **A chain of two or more hops does not resolve**, on either side, and that is
689 /// permanent rather than pending: only one origin is ever recorded and every hop
690 /// overwrites it. [`carried`] removes the key from the incoming metadata outright and
691 /// [`recorded`] writes the id at the *immediate* source on every path but the
692 /// copy-back, so A → B → C leaves C keyed by B and A's id gone.
693 ///
694 /// The second key costs no read — the referent's metadata is already in hand. Chasing
695 /// the chain further would need the intermediate stores configured and reachable, which
696 /// would put a third party's availability inside a copy; a durable lineage id would
697 /// identify only records written after it landed, so it would resolve nothing already
698 /// on a destination.
699 fn keys(&self) -> Vec<String> {
700 let mut keys = vec![self.id.to_string()];
701 if let Some(origin) = &self.origin {
702 let recorded = origin.to_string();
703 if recorded != keys[0] {
704 keys.push(recorded);
705 }
706 }
707 keys
708 }
709}
710
711/// What every whole-reference occurrence of one location string becomes.
712///
713/// Three variants rather than a rewrite beside a flag, because the two ways of leaving an
714/// occurrence alone are what the two figures a copy reports are *about*: one is the design
715/// working and the other says the destination or the source holds something a re-run will
716/// never fix. A bool would let a reader of this type read them as the same outcome.
717enum Resolution {
718 /// The destination's own location string for the counterpart.
719 Rewrite(String),
720 /// The destination holds no counterpart, or holds one it reports no location for, so
721 /// the occurrence is left byte-for-byte. Ordinary and expected under the bound this
722 /// design works to.
723 NoCounterpart,
724 /// The correspondence could not be established **confidently** — more than one
725 /// destination record matches the referent's two keys, or two referents report this one
726 /// location string. The occurrence is left byte-for-byte, no record is chosen, and a
727 /// re-run will never clear it.
728 Ambiguous,
729}
730
731/// Every destination record that records an origin, read **once per copy invocation**.
732///
733/// Several documents in one project is the ordinary case, and a walk per document would
734/// multiply reads against a rate limiter for no gain — so this is built once, for the
735/// levels the invocation's own documents really name, and every document and every
736/// referent of that invocation is answered out of it. A copy whose documents hold no
737/// candidate reference builds none at all.
738///
739/// **This is a stricter discipline than [`Engine::scan`], deliberately.** That lookup takes
740/// the first hit and stops, and every consumer of the copy already depends on it doing so;
741/// it chooses the copy's own target, where a caller named the item. This one edits the
742/// content of somebody's document, where a wrong answer is silent corruption of prose a
743/// person will act on — so where more than one record matches, it chooses none. The two
744/// lookups answer different questions and are meant to disagree on a destination holding
745/// duplicates.
746#[derive(Default)]
747struct Counterparts {
748 /// The records at one interface recording one origin.
749 by_origin: BTreeMap<(Level, String), Vec<Held>>,
750}
751
752/// One record a destination walk found: where it is there, and the location string that
753/// destination reports for it — `None` when it reports none.
754type Held = (NativeId, Option<String>);
755
756impl Counterparts {
757 /// Record one destination item, when it records an origin at all.
758 fn note(
759 &mut self,
760 level: Level,
761 id: &NativeId,
762 location: Option<&Location>,
763 metadata: &BTreeMap<String, Value>,
764 ) {
765 let Some(origin) = origin_of(metadata) else {
766 return;
767 };
768 self.by_origin
769 .entry((level, origin.to_string()))
770 .or_default()
771 .push((id.clone(), located(location)));
772 }
773
774 /// What one referent's occurrences become, by the two-key rule.
775 ///
776 /// Where the correspondence cannot be established **confidently**, the text is left
777 /// exactly as it is and no record is chosen — see the note on this type.
778 fn resolve(&self, referent: &Referent) -> Resolution {
779 let mut candidates: Vec<&Held> = Vec::new();
780 for key in referent.keys() {
781 for record in self
782 .by_origin
783 .get(&(referent.level, key))
784 .into_iter()
785 .flatten()
786 {
787 // One destination record matching both keys is one record, not two.
788 if !candidates.iter().any(|held| held.0 == record.0) {
789 candidates.push(record);
790 }
791 }
792 }
793 match candidates.as_slice() {
794 [] => Resolution::NoCounterpart,
795 // A counterpart the destination reports no location for names nowhere a reader
796 // could go, so the source's own string is left standing rather than removed.
797 [(_, location)] => location
798 .clone()
799 .map_or(Resolution::NoCounterpart, Resolution::Rewrite),
800 _ => Resolution::Ambiguous,
801 }
802 }
803}
804
805impl Engine {
806 /// Copy every item a request names into one configured destination.
807 ///
808 /// This is the whole of the verb, and the command line drives exactly this: a copy a
809 /// Rust caller makes and a copy typed at a shell are the same call, so the two cannot
810 /// answer the same copy differently.
811 ///
812 /// # Errors
813 ///
814 /// Returns [`EngineError`] when the destination is not configured, cannot be built,
815 /// cannot be written, or — for a document copy — declares it has no documents; when an
816 /// id names nothing; when an origin names an item the
817 /// destination no longer holds and `--recreate` was not given; and when the
818 /// destination refuses the write — including a field or a metadata key it cannot
819 /// carry, which it names rather than dropping.
820 pub async fn copy(&self, request: &CopyRequest) -> Result<CopyReport, EngineError> {
821 let destination = self.writable(&request.destination)?;
822 // Before anything is read, and from the declaration rather than from a failed
823 // write: a destination that says it has no documents has nowhere to put one.
824 if request.scope == CopyScope::Documents {
825 documentary(destination)?;
826 }
827 // The sources this command names, each read once before and once after, so what it
828 // spent is the difference between two of each one's own running totals.
829 let mut metered = vec![destination];
830 for id in request.items.as_slice() {
831 if let Some(source) = self.ready().find(|source| source.name() == &id.source)
832 && !metered.iter().any(|held| held.name() == source.name())
833 {
834 metered.push(source);
835 }
836 }
837 let before = readings(&metered).await;
838 let mut journal = Journal::default();
839 match self.copy_all(destination, request, &mut journal).await {
840 Ok((mut report, deliverers)) => {
841 // After the whole copy is complete and outside the journal: what the relation
842 // rule writes is the delivered tasks' own, and a failure there is reported
843 // against that task rather than undoing a copy that landed.
844 for deliverer in &deliverers {
845 report.delivered.extend(
846 self.deliver(
847 &deliverer.destination,
848 deliverer.category,
849 &deliverer.now,
850 &deliverer.before,
851 )
852 .await,
853 );
854 }
855 report.spent = spent_between(&before, &readings(&metered).await);
856 Ok(report)
857 }
858 Err(error) => Err(self.undo(destination, journal, error).await),
859 }
860 }
861
862 /// The copy itself, with everything it writes recorded so a failure can be undone.
863 ///
864 /// The ids named together are **one** copied set, and that is what makes an edge
865 /// between any two of them a real edge at the destination: a copy of two projects at
866 /// once knows that a task in the first depends on a task in the second, and a task
867 /// knows that the project it belongs to is being created beside it. Copying them one
868 /// at a time could not, and wrote the far end as the id it had at its *source* — a
869 /// dangling reference to somewhere the destination has never heard of.
870 async fn copy_all(
871 &self,
872 destination: &ResolvedSource,
873 request: &CopyRequest,
874 journal: &mut Journal,
875 ) -> Result<(CopyReport, Vec<Deliverer>), EngineError> {
876 let mut running = Running::default();
877 // The whole copied set, established before anything is written. For a project
878 // copy that means reading every named project's membership first: the set is the
879 // whole request rather than one project of it.
880 let items = match &request.scope {
881 CopyScope::Tasks | CopyScope::Documents => {
882 let kind = if request.scope == CopyScope::Documents {
883 Level::Document
884 } else {
885 Level::Task
886 };
887 running
888 .resolvable
889 .extend(request.items.as_slice().iter().cloned());
890 let mut planned = Vec::new();
891 for id in request.items.as_slice() {
892 planned.push(self.plan(destination, request, kind, id).await?);
893 }
894 self.copy_items(destination, request, planned, None, &mut running, journal)
895 .await?
896 }
897 CopyScope::Projects { tasks } => {
898 let mut projects = Vec::new();
899 for id in request.items.as_slice() {
900 let members = if *tasks {
901 self.project_members(id).await?
902 } else {
903 Vec::new()
904 };
905 running.resolvable.push(id.clone());
906 running.resolvable.extend(members.iter().cloned());
907 projects.push((id.clone(), members));
908 }
909 self.copy_projects(destination, request, &projects, &[], &mut running, journal)
910 .await?
911 }
912 CopyScope::Members(named) => {
913 let (projects, unrecorded) = self
914 .named_members(destination, request.items.as_slice(), named, &mut running)
915 .await?;
916 self.copy_projects(
917 destination,
918 request,
919 &projects,
920 &unrecorded,
921 &mut running,
922 journal,
923 )
924 .await?
925 }
926 };
927 let references = running.references;
928 let delivers_rewritten = running.delivers_rewritten;
929 // Every member's destination id is known once the first pass has landed it, so the
930 // tasks each landed task delivers are settled here rather than after the repair.
931 let deliverers: Vec<Deliverer> = running
932 .landed
933 .iter()
934 .map(|task| Deliverer {
935 destination: task.destination.clone(),
936 category: task.category,
937 now: targets(
938 &resolved_entries(&mapped_delivers(
939 &task.delivers,
940 &task.origin,
941 destination,
942 &running.resolvable,
943 &running.counterparts,
944 )),
945 destination.name(),
946 ),
947 before: task.before.clone(),
948 })
949 .collect();
950 self.repair(destination, request, running, journal).await?;
951 Ok((
952 CopyReport {
953 items,
954 references_rewritten: references.rewritten,
955 references_unresolved: references.unresolved,
956 references_ambiguous: references.ambiguous,
957 delivers_rewritten,
958 // Filled in by `copy`, once the copy is complete.
959 delivered: Vec::new(),
960 // Filled in by `copy`, which is the one place both readings are taken.
961 spent: None,
962 },
963 deliverers,
964 ))
965 }
966
967 /// Write every deferred item again, now that every destination id is known.
968 ///
969 /// This is the second half of the two passes an edge between two items of one copy
970 /// needs: the far end's destination id does not exist until it has been created, so
971 /// the item that points at it lands first without that edge and is completed here.
972 /// It runs once for the whole request rather than once per project, because a far end
973 /// may be in a project this copy has not reached yet.
974 async fn repair(
975 &self,
976 destination: &ResolvedSource,
977 request: &CopyRequest,
978 running: Running,
979 journal: &mut Journal,
980 ) -> Result<(), EngineError> {
981 if request.dry_run {
982 return Ok(());
983 }
984 let Running {
985 resolvable,
986 counterparts,
987 deferred,
988 ..
989 } = running;
990 for entry in deferred {
991 let edges = mapped_edges(
992 &entry.item.edges,
993 &entry.item.source.source,
994 destination,
995 &resolvable,
996 &counterparts,
997 );
998 let delivers = delivers_of(&entry.item, destination, &resolvable, &counterparts);
999 self.write(
1000 destination,
1001 &entry.item,
1002 Some(entry.destination),
1003 entry.filed,
1004 &resolved(&edges),
1005 &resolved_entries(&delivers),
1006 entry.prior,
1007 journal,
1008 )
1009 .await?;
1010 }
1011 Ok(())
1012 }
1013
1014 /// Put the destination back the way this copy found it, then report why it failed.
1015 ///
1016 /// Undone in reverse, and an item this copy created is removed rather than restored —
1017 /// the entry recording what it looked like a moment after creation is not a state
1018 /// anybody asked for. When the destination cannot take one of them back, the refusal
1019 /// says so and names what is still there, because a user told "the copy failed" about
1020 /// a destination that is not as they left it will copy again over a tree nobody
1021 /// described.
1022 async fn undo(
1023 &self,
1024 destination: &ResolvedSource,
1025 journal: Journal,
1026 error: EngineError,
1027 ) -> EngineError {
1028 let created: Vec<(Level, &NativeId)> = journal
1029 .entries
1030 .iter()
1031 .filter_map(|entry| match entry {
1032 Undo::Created { kind, id } => Some((*kind, id)),
1033 Undo::Updated { .. } => None,
1034 })
1035 .collect();
1036 // The ids and the refusal are one value rather than two, because they are one
1037 // fact: an item is only left behind because the destination refused to take it
1038 // back, so the first refusal carries the first id and neither half can be
1039 // recorded without the other.
1040 let mut unrestored: Option<(LeftBehind, SourceError)> = None;
1041 for entry in journal.entries.iter().rev() {
1042 let outcome = match entry {
1043 Undo::Created { kind, id } => remove(destination, *kind, id).await,
1044 Undo::Updated { id, prior, .. } if !created.contains(&(prior.item.level(), id)) => {
1045 restore(destination, id, prior).await
1046 }
1047 Undo::Updated { .. } => Ok(()),
1048 };
1049 if let Err(problem) = outcome {
1050 let id = GlobalId::new(destination.name().clone(), entry.id().clone());
1051 match &mut unrestored {
1052 Some((left_behind, _)) => left_behind.push(id),
1053 None => unrestored = Some((LeftBehind::new(id), problem)),
1054 }
1055 }
1056 }
1057 match unrestored {
1058 None => error,
1059 Some((left_behind, refusal)) => EngineError::CopyNotUndone {
1060 error: Box::new(error),
1061 left_behind,
1062 refusal,
1063 },
1064 }
1065 }
1066
1067 /// The destination source, once it is established it exists and can be written.
1068 fn writable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
1069 let name = self.known(name)?;
1070 if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
1071 return Err(EngineError::DestinationUnavailable {
1072 name: name.to_string(),
1073 error: unavailable.error().clone(),
1074 });
1075 }
1076 let source = self
1077 .ready()
1078 .find(|source| source.name() == &name)
1079 .ok_or(EngineError::NoSources)?;
1080 if !source.source().writes().is_supported() {
1081 return Err(EngineError::NotWritable {
1082 name: name.to_string(),
1083 kind: source.kind().to_owned(),
1084 });
1085 }
1086 Ok(source)
1087 }
1088
1089 /// Copy each project and what travels with it: every task in it, the members a member
1090 /// copy names, or nothing at all.
1091 ///
1092 /// Every item of every project is read and its target decided **before any of them is
1093 /// written, and each exactly once.** That is what lets a member copy refuse an edge it
1094 /// cannot resolve while the destination is still as it was found, and what stops a
1095 /// whole copy resolving one target twice — once to learn the ids the project's own
1096 /// edges name, and again to land it — which against a hosted destination is a read per
1097 /// item spent on an answer the command already had.
1098 async fn copy_projects(
1099 &self,
1100 destination: &ResolvedSource,
1101 request: &CopyRequest,
1102 projects: &[(GlobalId, Vec<GlobalId>)],
1103 unrecorded: &[GlobalId],
1104 running: &mut Running,
1105 journal: &mut Journal,
1106 ) -> Result<Vec<CopyOutcome>, EngineError> {
1107 let mut plans = Vec::new();
1108 for (id, members) in projects {
1109 let project = self.plan(destination, request, Level::Project, id).await?;
1110 let mut tasks = Vec::new();
1111 for member in members {
1112 tasks.push(self.plan(destination, request, Level::Task, member).await?);
1113 }
1114 plans.push((project, tasks));
1115 }
1116 for (project, tasks) in &plans {
1117 for item in std::iter::once(project).chain(tasks) {
1118 unrecorded_far_end(item, destination, unrecorded)?;
1119 }
1120 }
1121 let mut outcomes = Vec::new();
1122 for (project, tasks) in plans {
1123 outcomes.extend(
1124 self.copy_project(destination, request, project, tasks, running, journal)
1125 .await?,
1126 );
1127 }
1128 Ok(outcomes)
1129 }
1130
1131 /// Land one planned project, then the tasks planned beside it.
1132 async fn copy_project(
1133 &self,
1134 destination: &ResolvedSource,
1135 request: &CopyRequest,
1136 project: Planned,
1137 tasks: Vec<Planned>,
1138 running: &mut Running,
1139 journal: &mut Journal,
1140 ) -> Result<Vec<CopyOutcome>, EngineError> {
1141 let (carries_tasks, walks_orphans) = match request.scope {
1142 CopyScope::Projects { tasks } => (tasks, tasks),
1143 CopyScope::Members(_) => (true, false),
1144 CopyScope::Tasks | CopyScope::Documents => (false, false),
1145 };
1146 let id = project.source.clone();
1147 let members: Vec<GlobalId> = tasks.iter().map(|task| task.source.clone()).collect();
1148 // The project lands with every edge it can already resolve — its members' targets
1149 // are known from their plans — so a project that has not changed is not written at
1150 // all. What it *reports* is read the way it always has been: against the ids known
1151 // before its own members' were, unless every edge resolves and nothing differs.
1152 // Reading it any other way would move the word a repeat copy reports for a project
1153 // whose only difference is an edge, which is not this change's to move.
1154 let mut before_members = running.counterparts.clone();
1155 if let Target::Update { id: target, .. } = &project.target {
1156 before_members.insert(id.to_string(), target.clone());
1157 }
1158 for item in std::iter::once(&project).chain(&tasks) {
1159 if let Target::Update { id: target, .. } = &item.target {
1160 running
1161 .counterparts
1162 .insert(item.source.to_string(), target.clone());
1163 }
1164 }
1165 let unchanged = match &project.target {
1166 Target::Update { id: target, .. } => {
1167 let settled = mapped_edges(
1168 &project.edges,
1169 &id.source,
1170 destination,
1171 &running.resolvable,
1172 &running.counterparts,
1173 );
1174 let first = mapped_edges(
1175 &project.edges,
1176 &id.source,
1177 destination,
1178 &running.resolvable,
1179 &before_members,
1180 );
1181 let held = project.held.as_ref();
1182 let unchanged_with = |edges: &[Option<DependencyEdge>]| {
1183 !changes(
1184 held,
1185 &project,
1186 target,
1187 &None,
1188 &resolved(edges),
1189 &[],
1190 destination.name(),
1191 )
1192 };
1193 Some(
1194 (!settled.iter().any(Option::is_none) && unchanged_with(&settled))
1195 || unchanged_with(&first),
1196 )
1197 }
1198 Target::Create => None,
1199 };
1200 let mut outcomes = self
1201 .copy_items(destination, request, vec![project], None, running, journal)
1202 .await?;
1203 if let (Some(unchanged), Some(landed)) = (unchanged, outcomes[0].destination().cloned()) {
1204 outcomes[0].action = if unchanged {
1205 CopyAction::Unchanged {
1206 destination: landed,
1207 }
1208 } else {
1209 CopyAction::Updated {
1210 destination: landed,
1211 }
1212 };
1213 }
1214 if !carries_tasks {
1215 return Ok(outcomes);
1216 }
1217 // `None` when a dry run would have created the project: nothing was written, so
1218 // there is no destination project id to file the tasks under. Every task is still
1219 // read and still reported, because that is what a dry run is for.
1220 let filed = outcomes.first().and_then(CopyOutcome::destination).cloned();
1221 outcomes.extend(
1222 self.copy_items(
1223 destination,
1224 request,
1225 tasks,
1226 filed.as_ref().map(|project| project.native.clone()),
1227 running,
1228 journal,
1229 )
1230 .await?,
1231 );
1232 // A member copy was told which members it carries, so it cannot tell a member it
1233 // left out from one the source no longer holds, and does not walk for either.
1234 if walks_orphans && let Some(filed) = filed {
1235 outcomes.extend(
1236 self.orphans(destination, &id, &filed.native, &members)
1237 .await?,
1238 );
1239 }
1240 Ok(outcomes)
1241 }
1242
1243 /// The members a member copy names, per project and in the order they were named, and
1244 /// every member it does not name that records no destination id.
1245 ///
1246 /// Settled before anything is read at the destination. A named id that is a member of
1247 /// none of the projects is refused here, naming it. An unnamed member whose recorded
1248 /// origin names the destination joins the copied set with that id already known, so an
1249 /// edge to it resolves exactly as an edge to a copied item does, and costs no read. An
1250 /// unnamed member recording none is returned, so an edge to it is refused before
1251 /// anything is written — see [`unrecorded_far_end`].
1252 async fn named_members(
1253 &self,
1254 destination: &ResolvedSource,
1255 projects: &[GlobalId],
1256 named: &CopyItems,
1257 running: &mut Running,
1258 ) -> Result<(Vec<(GlobalId, Vec<GlobalId>)>, Vec<GlobalId>), EngineError> {
1259 let mut carried = Vec::new();
1260 let mut unrecorded = Vec::new();
1261 for project in projects {
1262 let held = self.project_member_tasks(project).await?;
1263 let mut members: Vec<GlobalId> = Vec::new();
1264 for id in named.as_slice() {
1265 if held.iter().any(|task| &task.id == id) && !members.contains(id) {
1266 members.push(id.clone());
1267 }
1268 }
1269 for task in held {
1270 if members.contains(&task.id) {
1271 continue;
1272 }
1273 match origin_of(&task.item.metadata)
1274 .filter(|origin| &origin.source == destination.name())
1275 {
1276 Some(origin) => {
1277 running
1278 .counterparts
1279 .insert(task.id.to_string(), origin.native);
1280 running.resolvable.push(task.id);
1281 }
1282 None => unrecorded.push(task.id),
1283 }
1284 }
1285 running.resolvable.push(project.clone());
1286 running.resolvable.extend(members.iter().cloned());
1287 carried.push((project.clone(), members));
1288 }
1289 if let Some(stray) = named
1290 .as_slice()
1291 .iter()
1292 .find(|id| !carried.iter().any(|(_, members)| members.contains(id)))
1293 {
1294 return Err(EngineError::NotAMember {
1295 id: stray.clone(),
1296 projects: projects.to_vec(),
1297 });
1298 }
1299 Ok((carried, unrecorded))
1300 }
1301
1302 /// Every task the source holds in `project`, by qualified id.
1303 async fn project_members(&self, project: &GlobalId) -> Result<Vec<GlobalId>, EngineError> {
1304 Ok(self
1305 .project_member_tasks(project)
1306 .await?
1307 .into_iter()
1308 .map(|task| task.id)
1309 .collect())
1310 }
1311
1312 /// Every task the source holds in `project`, as the source reported it.
1313 ///
1314 /// The ids alone are what a copy files under a project; the whole task is what a
1315 /// document's references need, because the location a reference names is a field of it.
1316 async fn project_member_tasks(
1317 &self,
1318 project: &GlobalId,
1319 ) -> Result<Vec<Qualified<Task>>, EngineError> {
1320 let mut request = TaskRequest {
1321 sources: vec![project.source.clone()],
1322 filters: Filters::default(),
1323 project: ProjectSelector::Qualified(project.clone()),
1324 priorities: Vec::new(),
1325 commented_since: None,
1326 paging: Paging {
1327 limit: PROJECT_PAGE,
1328 token: None,
1329 },
1330 };
1331 let mut members = Vec::new();
1332 // Pages by this engine's own token rather than by a source cursor, and the
1333 // asymmetry with the three walks below is deliberate. A source answering two
1334 // cursors with each other advances on every page, so `unrepeated` under the list
1335 // verb never fires; the cycle shows only as a token handed back unchanged, which
1336 // is what this loop pages by. Point it at the source and a project copy spins.
1337 //
1338 // No page bound here for the same reason: `Engine::tasks` merges under the budget
1339 // it was asked, so nothing longer than `PROJECT_PAGE` can arrive. `fits` in
1340 // `fetch::walk` refuses the page a source can really overrun, its own, while
1341 // these very members are read.
1342 let misbehaved = |error| EngineError::SourceRefused {
1343 name: project.source.to_string(),
1344 error,
1345 };
1346 loop {
1347 let asked = request.paging.token.clone();
1348 let response = self.tasks(&request).await?;
1349 if let Some(failure) = response.errors.first() {
1350 return Err(EngineError::SourceRefused {
1351 name: failure.source.to_string(),
1352 error: failure.error.clone(),
1353 });
1354 }
1355 unrepeated(
1356 response.next.as_ref(),
1357 asked.as_ref(),
1358 "the tasks of a project were being read for a copy",
1359 )
1360 .map_err(misbehaved)?;
1361 members.extend(response.items);
1362 match response.next {
1363 Some(token) => request.paging.token = Some(token),
1364 None => return Ok(members),
1365 }
1366 }
1367 }
1368
1369 /// Every document the source holds in `project`, as the source reported it.
1370 ///
1371 /// Paged by this engine's own token for the reason the member walk above is, and the
1372 /// note there says why.
1373 async fn project_documents(
1374 &self,
1375 project: &GlobalId,
1376 ) -> Result<Vec<Qualified<Document>>, EngineError> {
1377 let mut request = DocumentRequest {
1378 sources: vec![project.source.clone()],
1379 filters: DocumentFilters::default(),
1380 project: ProjectSelector::Qualified(project.clone()),
1381 paging: Paging {
1382 limit: PROJECT_PAGE,
1383 token: None,
1384 },
1385 };
1386 let mut held = Vec::new();
1387 let misbehaved = |error| EngineError::SourceRefused {
1388 name: project.source.to_string(),
1389 error,
1390 };
1391 loop {
1392 let asked = request.paging.token.clone();
1393 let response = self.documents(&request).await?;
1394 if let Some(failure) = response.errors.first() {
1395 return Err(EngineError::SourceRefused {
1396 name: failure.source.to_string(),
1397 error: failure.error.clone(),
1398 });
1399 }
1400 unrepeated(
1401 response.next.as_ref(),
1402 asked.as_ref(),
1403 "the documents of a project were being read for a copy",
1404 )
1405 .map_err(misbehaved)?;
1406 held.extend(response.items);
1407 match response.next {
1408 Some(token) => request.paging.token = Some(token),
1409 None => return Ok(held),
1410 }
1411 }
1412 }
1413
1414 /// Point every reference the documents of this copy hold at the destination's own
1415 /// records, and say how many it could not.
1416 ///
1417 /// A document copied out of a local Markdown store used to arrive naming absolute paths
1418 /// under one checkout on one machine, dead for the only reader the copy exists for,
1419 /// while the destination held its own record for every one of them the whole time. This
1420 /// is what closes that, and it is deliberately **not** a Markdown-link parser: the
1421 /// artifact that motivated it holds bare absolute paths inside backticks in a table
1422 /// cell, which `[text](target)` matching would have left exactly as it found them.
1423 ///
1424 /// The correspondence is re-established from what the *destination* records at
1425 /// [`GlobalId::ORIGIN_KEY`], not from the mapping this copy holds. That mapping is not
1426 /// available when it is needed: `project copy` and `document copy` are separate verbs,
1427 /// one invocation carries one [`CopyScope`], and a project copy carries no documents —
1428 /// so by the time the document is copied, the tasks were written by a process that has
1429 /// exited. Reading the destination is also what makes a document copied on its own
1430 /// work, which a same-run mapping never could.
1431 ///
1432 /// Tasks and projects are not touched. Only a document's content is rewritten, and
1433 /// nothing else about it changes.
1434 async fn rewrite_references(
1435 &self,
1436 destination: &ResolvedSource,
1437 planned: &mut [Planned],
1438 counts: &mut Counted,
1439 ) -> Result<(), EngineError> {
1440 // Read at the source, once per project rather than once per document: several
1441 // documents of one project is the ordinary case.
1442 let mut by_project: BTreeMap<String, Vec<Referent>> = BTreeMap::new();
1443 let mut named: Vec<Vec<Referent>> = Vec::new();
1444 for item in planned.iter() {
1445 named.push(self.named_referents(item, &mut by_project).await?);
1446 }
1447 // The destination is walked only for a copy that really names something, and only
1448 // for the interfaces those referents are read from.
1449 let mut levels: Vec<Level> = Vec::new();
1450 for referent in named.iter().flatten() {
1451 if !levels.contains(&referent.level) {
1452 levels.push(referent.level);
1453 }
1454 }
1455 if levels.is_empty() {
1456 return Ok(());
1457 }
1458 let counterparts = self.counterparts(destination, &levels).await?;
1459 for (item, referents) in planned.iter_mut().zip(named) {
1460 let Item::Document(document) = &mut item.item else {
1461 continue;
1462 };
1463 let Some(content) = &document.content else {
1464 continue;
1465 };
1466 let (rewritten, made) = substitute(content, &table_for(&referents, &counterparts));
1467 document.content = Some(rewritten);
1468 counts.add(made);
1469 }
1470 Ok(())
1471 }
1472
1473 /// The records of one document's own project whose location string its content really
1474 /// holds, as a whole reference.
1475 ///
1476 /// A document with no content, or with no project at the source, names nothing: the
1477 /// referent set is the document's own project, its tasks and its other documents, and
1478 /// there is no such set without a project.
1479 async fn named_referents(
1480 &self,
1481 item: &Planned,
1482 by_project: &mut BTreeMap<String, Vec<Referent>>,
1483 ) -> Result<Vec<Referent>, EngineError> {
1484 let Item::Document(document) = &item.item else {
1485 return Ok(Vec::new());
1486 };
1487 let (Some(content), Some(project)) =
1488 (document.content.as_deref(), document.project.as_ref())
1489 else {
1490 return Ok(Vec::new());
1491 };
1492 if content.is_empty() {
1493 return Ok(Vec::new());
1494 }
1495 let project = GlobalId::new(item.source.source.clone(), project.clone());
1496 let key = project.to_string();
1497 if !by_project.contains_key(&key) {
1498 let read = self.referents(&project).await?;
1499 by_project.insert(key.clone(), read);
1500 }
1501 Ok(by_project[&key]
1502 .iter()
1503 // A document does not name itself: the referent set is every *other* record
1504 // filed under the project. Told by the interface as well as the id, because an
1505 // id alone does not identify a record — a folder of Markdown filing `A.md`
1506 // under both `tasks/` and `documents/` is the ordinary case rather than the
1507 // contrived one, and excluding by id alone would drop that task from the set.
1508 .filter(|referent| referent.level != Level::Document || referent.id != item.source)
1509 .filter(|referent| holds(content, &referent.location))
1510 .cloned()
1511 .collect())
1512 }
1513
1514 /// Every record filed under one project at its own source, with the location string
1515 /// that source reports for it.
1516 ///
1517 /// The project record itself, every task filed under it, and every document filed under
1518 /// it. All three are reads this engine already knows how to make.
1519 async fn referents(&self, project: &GlobalId) -> Result<Vec<Referent>, EngineError> {
1520 let source = self.readable(&project.source)?;
1521 let mut referents = Vec::new();
1522 if let Some(held) = source
1523 .source()
1524 .get_project(&project.native)
1525 .await
1526 .map_err(|error| refused(source, error))?
1527 {
1528 note(
1529 &mut referents,
1530 project.clone(),
1531 Level::Project,
1532 held.location.as_ref(),
1533 &held.metadata,
1534 );
1535 }
1536 for task in self.project_member_tasks(project).await? {
1537 note(
1538 &mut referents,
1539 task.id,
1540 Level::Task,
1541 task.item.location.as_ref(),
1542 &task.item.metadata,
1543 );
1544 }
1545 for document in self.project_documents(project).await? {
1546 note(
1547 &mut referents,
1548 document.id,
1549 Level::Document,
1550 document.item.location.as_ref(),
1551 &document.item.metadata,
1552 );
1553 }
1554 Ok(referents)
1555 }
1556
1557 /// Walk the destination once for every record it holds at the levels named, one page
1558 /// at a time.
1559 ///
1560 /// One page is *read* at a time, and what is kept from each is three fields of each
1561 /// record — its id, the origin it records and the location the destination reports —
1562 /// never the page. That is more than [`Engine::scan`] keeps, and deliberately: the whole
1563 /// point of this walk is that one pass answers every document and every referent of the
1564 /// invocation, so what it learns has to outlive the page it learned it from. Nothing is
1565 /// written down and the index is dropped with the call.
1566 ///
1567 /// The other difference from `scan` is the *answer*: this one keeps every match, so a
1568 /// destination holding two records for one work item is reported as ambiguous rather
1569 /// than resolved to the first.
1570 async fn counterparts(
1571 &self,
1572 destination: &ResolvedSource,
1573 levels: &[Level],
1574 ) -> Result<Counterparts, EngineError> {
1575 let mut found = Counterparts::default();
1576 for level in levels {
1577 // Every cursor this level has already been sent. `unrepeated` below catches a
1578 // source that hands back the cursor it was just given, and its own note says
1579 // why it catches no more than that: a source cycling through two cursors
1580 // advances on every page, and seeing it needs memory the walks that share that
1581 // helper do not keep. This walk does keep memory — it is building an index that
1582 // outlives each page — so here the memory exists and the cycle is caught. The
1583 // walks one level up catch the same defect as a page token handed back
1584 // unchanged; this one pages by the source's own cursor and has no level above
1585 // it, so nothing else would.
1586 let mut asked_before: BTreeSet<String> = BTreeSet::new();
1587 let mut cursor: Option<Cursor> = None;
1588 loop {
1589 if let Some(next) = &cursor
1590 && !asked_before.insert(next.0.clone())
1591 {
1592 return Err(refused(
1593 destination,
1594 SourceError::Malformed {
1595 message: "the source returned a cursor it had already been \
1596 given while the destination was being walked for the \
1597 records a document's references name, so the walk \
1598 would never end"
1599 .to_owned(),
1600 },
1601 ));
1602 }
1603 let asked = cursor.clone();
1604 let request = request_for(destination, cursor);
1605 let next = match level {
1606 Level::Task => {
1607 let page = destination
1608 .source()
1609 .query_tasks(&TaskQuery::default(), &request)
1610 .await
1611 .map_err(|error| refused(destination, error))?;
1612 fits(page.items.len(), request.limit)
1613 .map_err(|error| refused(destination, error))?;
1614 for task in &page.items {
1615 found.note(*level, &task.id, task.location.as_ref(), &task.metadata);
1616 }
1617 page.next
1618 }
1619 Level::Project => {
1620 let page = destination
1621 .source()
1622 .query_projects(&ProjectQuery::default(), &request)
1623 .await
1624 .map_err(|error| refused(destination, error))?;
1625 fits(page.items.len(), request.limit)
1626 .map_err(|error| refused(destination, error))?;
1627 for project in &page.items {
1628 found.note(
1629 *level,
1630 &project.id,
1631 project.location.as_ref(),
1632 &project.metadata,
1633 );
1634 }
1635 page.next
1636 }
1637 Level::Document => {
1638 let page = destination
1639 .source()
1640 .query_documents(&DocumentQuery::default(), &request)
1641 .await
1642 .map_err(|error| refused(destination, error))?;
1643 fits(page.items.len(), request.limit)
1644 .map_err(|error| refused(destination, error))?;
1645 for document in &page.items {
1646 found.note(
1647 *level,
1648 &document.id,
1649 document.location.as_ref(),
1650 &document.metadata,
1651 );
1652 }
1653 page.next
1654 }
1655 };
1656 unrepeated(
1657 next.as_ref(),
1658 asked.as_ref(),
1659 "the destination was being walked for the records a document's \
1660 references name",
1661 )
1662 .map_err(|error| refused(destination, error))?;
1663 match next {
1664 Some(next) => cursor = Some(next),
1665 None => break,
1666 }
1667 }
1668 }
1669 Ok(found)
1670 }
1671
1672 /// Destination tasks filed under the copied project whose origin the source no longer
1673 /// holds.
1674 ///
1675 /// A copy never deletes, so each is left exactly as it is and reported.
1676 async fn orphans(
1677 &self,
1678 destination: &ResolvedSource,
1679 project: &GlobalId,
1680 at_destination: &NativeId,
1681 copied: &[GlobalId],
1682 ) -> Result<Vec<CopyOutcome>, EngineError> {
1683 let mut orphans = Vec::new();
1684 let mut cursor: Option<Cursor> = None;
1685 loop {
1686 let asked = cursor.clone();
1687 let request = request_for(destination, cursor);
1688 let page: Page<Task> = destination
1689 .source()
1690 .query_tasks(&TaskQuery::default(), &request)
1691 .await
1692 .map_err(|error| refused(destination, error))?;
1693 fits(page.items.len(), request.limit).map_err(|error| refused(destination, error))?;
1694 for task in &page.items {
1695 if task.project.as_ref() != Some(at_destination) {
1696 continue;
1697 }
1698 let Some(origin) = origin_of(&task.metadata) else {
1699 continue;
1700 };
1701 if origin.source != project.source || copied.contains(&origin) {
1702 continue;
1703 }
1704 orphans.push(CopyOutcome {
1705 source: origin,
1706 action: CopyAction::Orphaned {
1707 destination: GlobalId::new(destination.name().clone(), task.id.clone()),
1708 },
1709 });
1710 }
1711 unrepeated(
1712 page.next.as_ref(),
1713 asked.as_ref(),
1714 "the destination was being read for items the copy left behind",
1715 )
1716 .map_err(|error| refused(destination, error))?;
1717 match page.next {
1718 Some(next) => cursor = Some(next),
1719 None => return Ok(orphans),
1720 }
1721 }
1722 }
1723
1724 /// Resolve and write every item planned, holding back the ones whose edges are not
1725 /// resolvable yet.
1726 ///
1727 /// An edge between two items of one copy can point at a member whose destination id
1728 /// does not exist until it has been created, so the item that points at it lands
1729 /// without that edge and is handed to `deferred`. [`Engine::repair`] finishes it once
1730 /// the *whole* request has landed — not once this call has, because the far end may
1731 /// be in another project of the same command.
1732 async fn copy_items(
1733 &self,
1734 destination: &ResolvedSource,
1735 request: &CopyRequest,
1736 mut planned: Vec<Planned>,
1737 project: Option<NativeId>,
1738 running: &mut Running,
1739 journal: &mut Journal,
1740 ) -> Result<Vec<CopyOutcome>, EngineError> {
1741 // Only a document's own content names other records, and only once every document
1742 // of this call has been read: the destination is walked once for all of them, and
1743 // the content the rest of this call lands is the rewritten one — which is what
1744 // makes a repeat copy of an already-rewritten document report `unchanged`.
1745 if planned
1746 .iter()
1747 .any(|item| item.item.level() == Level::Document)
1748 {
1749 self.rewrite_references(destination, &mut planned, &mut running.references)
1750 .await?;
1751 }
1752
1753 for item in &planned {
1754 if let Target::Update { id, .. } = &item.target {
1755 running
1756 .counterparts
1757 .insert(item.source.to_string(), id.clone());
1758 }
1759 if let Item::Task(task) = &item.item {
1760 running.delivers_rewritten +=
1761 members_named(&task.delivers, &item.source.source, &running.resolvable);
1762 }
1763 }
1764
1765 // Resolved once per item, and used by both passes: the repair pass writes the
1766 // same item again, and re-deriving this there could file it somewhere else.
1767 let mut filed = Vec::new();
1768 for item in &planned {
1769 filed.push(
1770 self.filed(destination, item, project.clone(), running)
1771 .await?,
1772 );
1773 }
1774
1775 let mut outcomes = Vec::new();
1776 let mut unresolved = Vec::new();
1777 let mut priors = Vec::new();
1778 for (index, item) in planned.iter().enumerate() {
1779 let edges = mapped_edges(
1780 &item.edges,
1781 &item.source.source,
1782 destination,
1783 &running.resolvable,
1784 &running.counterparts,
1785 );
1786 let delivers = delivers_of(
1787 item,
1788 destination,
1789 &running.resolvable,
1790 &running.counterparts,
1791 );
1792 if edges.iter().any(Option::is_none) || delivers.iter().any(Option::is_none) {
1793 unresolved.push(index);
1794 }
1795 let resolvable_delivers = resolved_entries(&delivers);
1796 let (outcome, prior) = self
1797 .land(
1798 destination,
1799 request,
1800 item,
1801 filed[index].clone(),
1802 Pointing {
1803 edges: &edges,
1804 delivers: &resolvable_delivers,
1805 },
1806 journal,
1807 )
1808 .await?;
1809 if let Some(id) = outcome.destination() {
1810 running
1811 .counterparts
1812 .insert(item.source.to_string(), id.native.clone());
1813 }
1814 if !request.dry_run
1815 && let (Item::Task(task), Some(landed)) = (&item.item, outcome.destination())
1816 {
1817 running.landed.push(LandedTask {
1818 destination: landed.clone(),
1819 origin: item.source.source.clone(),
1820 delivers: task.delivers.clone(),
1821 before: match prior.as_ref().map(|prior| &prior.item) {
1822 Some(Item::Task(held)) => targets(&held.delivers, destination.name()),
1823 _ => Vec::new(),
1824 },
1825 category: task.status.category,
1826 });
1827 }
1828 outcomes.push(outcome);
1829 priors.push(prior);
1830 }
1831
1832 if !request.dry_run {
1833 for (index, item) in planned.into_iter().enumerate() {
1834 if !unresolved.contains(&index) {
1835 continue;
1836 }
1837 // Every item a copy that is not a dry run lands has a destination id: the
1838 // one outcome without one is a dry run that would have created, and this
1839 // block does not run for a dry run.
1840 let id = outcomes[index]
1841 .destination()
1842 .expect("a copy that writes lands every item it planned")
1843 .clone();
1844 running.deferred.push(Deferred {
1845 item,
1846 filed: filed[index].clone(),
1847 destination: id.native,
1848 prior: priors[index].clone(),
1849 });
1850 }
1851 }
1852 Ok(outcomes)
1853 }
1854
1855 /// Read one item and its forward edges, and decide where it is going.
1856 async fn plan(
1857 &self,
1858 destination: &ResolvedSource,
1859 request: &CopyRequest,
1860 kind: Level,
1861 id: &GlobalId,
1862 ) -> Result<Planned, EngineError> {
1863 let source = self.readable(&id.source)?;
1864 if kind == Level::Document {
1865 documentary(source)?;
1866 }
1867 let item = match kind {
1868 Level::Task => source
1869 .source()
1870 .get_task(&id.native)
1871 .await
1872 .map_err(|error| refused(source, error))?
1873 .map(|task| Item::Task(Box::new(task))),
1874 Level::Project => source
1875 .source()
1876 .get_project(&id.native)
1877 .await
1878 .map_err(|error| refused(source, error))?
1879 .map(|project| Item::Project(Box::new(project))),
1880 Level::Document => source
1881 .source()
1882 .get_document(&id.native)
1883 .await
1884 .map_err(|error| refused(source, error))?
1885 .map(|document| Item::Document(Box::new(document))),
1886 }
1887 .ok_or_else(|| EngineError::NoSuchItem { id: id.to_string() })?;
1888 // Before the destination is read, and so before it is written: a destination that
1889 // holds no priority has nowhere to put one, and every item of a copy is planned before
1890 // any of them lands. A task carrying `none` passes and writes exactly as it always did.
1891 if let Item::Task(task) = &item {
1892 holds_priority(destination, &id.to_string(), task.priority)?;
1893 }
1894 let edges = forward_edges(source, &id.native, item.level()).await?;
1895 let (target, held) = self.target(destination, request, id, &item).await?;
1896 Ok(Planned {
1897 source: id.clone(),
1898 item,
1899 edges,
1900 target,
1901 held,
1902 })
1903 }
1904
1905 /// Which destination item this one corresponds to, by the two origin rules and the
1906 /// caller's escape, and what the destination holds there.
1907 async fn target(
1908 &self,
1909 destination: &ResolvedSource,
1910 request: &CopyRequest,
1911 id: &GlobalId,
1912 item: &Item,
1913 ) -> Result<(Target, Option<Prior>), EngineError> {
1914 let (title, metadata) = described(item);
1915 if let Some(origin) = origin_of(metadata)
1916 && &origin.source == destination.name()
1917 {
1918 // The read that says the origin still names something is the read of what it
1919 // holds, so rule 1 costs one round trip rather than two.
1920 if let Some(held) = self
1921 .prior(destination, item.level(), &origin.native)
1922 .await?
1923 {
1924 return Ok((
1925 Target::Update {
1926 id: origin.native,
1927 found: Found::Origin,
1928 },
1929 Some(held),
1930 ));
1931 }
1932 if !request.recreate {
1933 return Err(EngineError::StaleOrigin {
1934 item: id.to_string(),
1935 origin: origin.to_string(),
1936 });
1937 }
1938 }
1939 if let Some(found) = self
1940 .scan(destination, item.level(), &Wanted::Origin(id.to_string()))
1941 .await?
1942 {
1943 let held = self.prior(destination, item.level(), &found).await?;
1944 return Ok((
1945 Target::Update {
1946 id: found,
1947 found: Found::Search,
1948 },
1949 held,
1950 ));
1951 }
1952 let wanted = match &request.match_by {
1953 Some(MatchBy::Title) => Some(Wanted::Title(title.to_owned())),
1954 Some(MatchBy::Metadata(key)) => metadata
1955 .get(key)
1956 .map(|value| Wanted::Metadata(key.clone(), value.clone())),
1957 None => None,
1958 };
1959 if let Some(wanted) = wanted
1960 && let Some(found) = self.scan(destination, item.level(), &wanted).await?
1961 {
1962 let held = self.prior(destination, item.level(), &found).await?;
1963 return Ok((
1964 Target::Update {
1965 id: found,
1966 found: Found::Search,
1967 },
1968 held,
1969 ));
1970 }
1971 Ok((Target::Create, None))
1972 }
1973
1974 /// Walk the destination one page at a time, looking for `wanted`.
1975 ///
1976 /// One page is held at a time and nothing is written down, which is the same bound
1977 /// every other compensation in this engine works under.
1978 async fn scan(
1979 &self,
1980 destination: &ResolvedSource,
1981 kind: Level,
1982 wanted: &Wanted,
1983 ) -> Result<Option<NativeId>, EngineError> {
1984 let mut cursor: Option<Cursor> = None;
1985 loop {
1986 let asked = cursor.clone();
1987 let request = request_for(destination, cursor);
1988 let next = match kind {
1989 Level::Task => {
1990 let page = destination
1991 .source()
1992 .query_tasks(&TaskQuery::default(), &request)
1993 .await
1994 .map_err(|error| refused(destination, error))?;
1995 fits(page.items.len(), request.limit)
1996 .map_err(|error| refused(destination, error))?;
1997 for task in &page.items {
1998 if wanted.found(&task.title, &task.metadata) {
1999 return Ok(Some(task.id.clone()));
2000 }
2001 }
2002 page.next
2003 }
2004 Level::Project => {
2005 let page = destination
2006 .source()
2007 .query_projects(&ProjectQuery::default(), &request)
2008 .await
2009 .map_err(|error| refused(destination, error))?;
2010 fits(page.items.len(), request.limit)
2011 .map_err(|error| refused(destination, error))?;
2012 for project in &page.items {
2013 if wanted.found(&project.title, &project.metadata) {
2014 return Ok(Some(project.id.clone()));
2015 }
2016 }
2017 page.next
2018 }
2019 Level::Document => {
2020 let page = destination
2021 .source()
2022 .query_documents(&DocumentQuery::default(), &request)
2023 .await
2024 .map_err(|error| refused(destination, error))?;
2025 fits(page.items.len(), request.limit)
2026 .map_err(|error| refused(destination, error))?;
2027 for document in &page.items {
2028 if wanted.found(&document.title, &document.metadata) {
2029 return Ok(Some(document.id.clone()));
2030 }
2031 }
2032 page.next
2033 }
2034 };
2035 unrepeated(
2036 next.as_ref(),
2037 asked.as_ref(),
2038 "the destination was being scanned for the item to update",
2039 )
2040 .map_err(|error| refused(destination, error))?;
2041 match next {
2042 Some(next) => cursor = Some(next),
2043 None => return Ok(None),
2044 }
2045 }
2046 }
2047
2048 /// Write one planned item, or say what a dry run would have done.
2049 ///
2050 /// Answers with what the destination held there beforehand as well, which is what
2051 /// makes an item written twice restorable to what it was rather than to what this
2052 /// copy's first pass left.
2053 async fn land(
2054 &self,
2055 destination: &ResolvedSource,
2056 request: &CopyRequest,
2057 item: &Planned,
2058 project: Option<NativeId>,
2059 pointing: Pointing<'_>,
2060 journal: &mut Journal,
2061 ) -> Result<(CopyOutcome, Option<Prior>), EngineError> {
2062 let Pointing { edges, delivers } = pointing;
2063 let target = match &item.target {
2064 Target::Update { id, .. } => Some(id.clone()),
2065 Target::Create => None,
2066 };
2067 // The one read of the destination item, made where its target was found, used to
2068 // decide whether the write changes anything and — if the copy cannot finish — to
2069 // put that item back.
2070 let prior = item.held.clone();
2071 let edges = resolved(edges);
2072 let qualified = |native: NativeId| GlobalId::new(destination.name().clone(), native);
2073 if let Some(id) = &target
2074 && !changes(
2075 prior.as_ref(),
2076 item,
2077 id,
2078 &project,
2079 &edges,
2080 delivers,
2081 destination.name(),
2082 )
2083 {
2084 return Ok((
2085 CopyOutcome {
2086 source: item.source.clone(),
2087 action: CopyAction::Unchanged {
2088 destination: qualified(id.clone()),
2089 },
2090 },
2091 prior,
2092 ));
2093 }
2094 if request.dry_run {
2095 return Ok((
2096 CopyOutcome {
2097 source: item.source.clone(),
2098 action: match target {
2099 Some(id) => CopyAction::Updated {
2100 destination: qualified(id),
2101 },
2102 // Null only here: nothing was created, so there is no id to report.
2103 None => CopyAction::Created { destination: None },
2104 },
2105 },
2106 prior,
2107 ));
2108 }
2109 let updating = target.is_some();
2110 let written = qualified(
2111 self.write(
2112 destination,
2113 item,
2114 target,
2115 project,
2116 &edges,
2117 delivers,
2118 prior.clone(),
2119 journal,
2120 )
2121 .await?,
2122 );
2123 Ok((
2124 CopyOutcome {
2125 source: item.source.clone(),
2126 action: if updating {
2127 CopyAction::Updated {
2128 destination: written,
2129 }
2130 } else {
2131 CopyAction::Created {
2132 destination: Some(written),
2133 }
2134 },
2135 },
2136 prior,
2137 ))
2138 }
2139
2140 /// Which destination project this item is filed under, when it is filed at all.
2141 ///
2142 /// A task copied as part of a project copy is filed under that project's counterpart,
2143 /// which the copy has just established. A task copied on its own has to find it.
2144 async fn filed(
2145 &self,
2146 destination: &ResolvedSource,
2147 item: &Planned,
2148 project: Option<NativeId>,
2149 running: &mut Running,
2150 ) -> Result<Option<NativeId>, EngineError> {
2151 match (&item.item, project) {
2152 (Item::Task(task), None) => {
2153 self.counterpart(destination, item, task.project.as_ref(), running)
2154 .await
2155 }
2156 (Item::Document(document), None) => {
2157 self.counterpart(destination, item, document.project.as_ref(), running)
2158 .await
2159 }
2160 (Item::Task(_) | Item::Document(_), filed) => Ok(filed),
2161 (Item::Project(_), _) => Ok(None),
2162 }
2163 }
2164
2165 /// The destination project this task's own project corresponds to, when there is one.
2166 ///
2167 /// A task copied on its own keeps its source's project id when the destination holds
2168 /// no counterpart: the field is opaque to this engine, and dropping it would lose
2169 /// what the source said.
2170 ///
2171 /// Looked for once per source project per command. A project this command itself
2172 /// copied is already known, and one an earlier task of this command was filed under
2173 /// was already looked for; walking the destination again for either would spend a
2174 /// scan per task on an answer the command holds.
2175 async fn counterpart(
2176 &self,
2177 destination: &ResolvedSource,
2178 item: &Planned,
2179 project: Option<&NativeId>,
2180 running: &mut Running,
2181 ) -> Result<Option<NativeId>, EngineError> {
2182 let Some(project) = project else {
2183 return Ok(None);
2184 };
2185 let qualified = GlobalId::new(item.source.source.clone(), project.clone()).to_string();
2186 if let Some(landed) = running.counterparts.get(&qualified) {
2187 return Ok(Some(landed.clone()));
2188 }
2189 if let Some(looked) = running.filings.get(&qualified) {
2190 return Ok(looked.clone());
2191 }
2192 let found = self
2193 .scan(
2194 destination,
2195 Level::Project,
2196 &Wanted::Origin(qualified.clone()),
2197 )
2198 .await?;
2199 let filed = Some(found.unwrap_or_else(|| project.clone()));
2200 running.filings.insert(qualified, filed.clone());
2201 Ok(filed)
2202 }
2203
2204 /// What the destination holds at one id, item and forward edges together.
2205 ///
2206 /// One read for both purposes it serves — deciding whether a write changes anything,
2207 /// and putting the item back if the copy cannot finish — because a second read of the
2208 /// same item is a second round trip against a hosted destination for nothing.
2209 async fn prior(
2210 &self,
2211 destination: &ResolvedSource,
2212 kind: Level,
2213 id: &NativeId,
2214 ) -> Result<Option<Prior>, EngineError> {
2215 let held = match kind {
2216 Level::Task => destination
2217 .source()
2218 .get_task(id)
2219 .await
2220 .map_err(|error| refused(destination, error))?
2221 .map(|task| Item::Task(Box::new(task))),
2222 Level::Project => destination
2223 .source()
2224 .get_project(id)
2225 .await
2226 .map_err(|error| refused(destination, error))?
2227 .map(|project| Item::Project(Box::new(project))),
2228 Level::Document => destination
2229 .source()
2230 .get_document(id)
2231 .await
2232 .map_err(|error| refused(destination, error))?
2233 .map(|document| Item::Document(Box::new(document))),
2234 };
2235 let Some(item) = held else {
2236 return Ok(None);
2237 };
2238 let edges = forward_edges(destination, id, kind).await?;
2239 Ok(Some(Prior { item, edges }))
2240 }
2241
2242 /// Hand one item to the destination's own write interface, recording how to take it
2243 /// back.
2244 // llmlint: ignore[suppressions_justified] A write is the item, where it is going, what
2245 // it is filed under, its edges, what was there before and the journal that records how
2246 // to put it back. Each is a distinct decision made by a different part of the copy, and
2247 // grouping them would only move the argument list to a constructor.
2248 #[allow(clippy::too_many_arguments)]
2249 async fn write(
2250 &self,
2251 destination: &ResolvedSource,
2252 item: &Planned,
2253 target: Option<NativeId>,
2254 project: Option<NativeId>,
2255 edges: &[DependencyEdge],
2256 delivers: &[TaskRef],
2257 prior: Option<Prior>,
2258 journal: &mut Journal,
2259 ) -> Result<NativeId, EngineError> {
2260 let created_kind = item.item.level();
2261 let suggested = target
2262 .clone()
2263 .unwrap_or_else(|| created_id(&item.item, project.as_ref()));
2264 // Settled before the journal takes `prior`, and from that same read: what the
2265 // destination holds at the origin key is what a copy-back leaves there, and what it
2266 // holds as `delivered_by` is what the item keeps.
2267 let origin = recorded(item, prior.as_ref());
2268 let landing = outgoing(item, suggested, project, &origin, delivers, prior.as_ref());
2269 // Recorded *before* the write rather than after it. A destination's own write is
2270 // several calls — `docs/plugin-protocol.md` §4.9 — and one of them failing leaves
2271 // the ones before it applied. No source can put those back, because only this
2272 // journal holds what was there; recorded after a successful write, an update that
2273 // stopped part way was the one way a copy could end and leave the destination
2274 // altered. A restore of an item the write never reached rewrites what is already
2275 // there, which costs one mutation and is what "either complete or it never
2276 // happened" is worth.
2277 if let (Some(id), Some(prior)) = (target.clone(), prior) {
2278 journal.record(Undo::Updated { id, prior });
2279 }
2280 let landed = match landing {
2281 Item::Task(task) => destination
2282 .source()
2283 .write_task(&ItemWrite {
2284 target: target.clone(),
2285 item: *task,
2286 depends_on: edges.to_vec(),
2287 })
2288 .await
2289 .map_err(|error| refused(destination, error))?,
2290 Item::Project(project) => destination
2291 .source()
2292 .write_project(&ItemWrite {
2293 target: target.clone(),
2294 item: *project,
2295 depends_on: edges.to_vec(),
2296 })
2297 .await
2298 .map_err(|error| refused(destination, error))?,
2299 // No edges, and that is the contract: a document takes part in no dependency
2300 // graph, so there is nothing here for `depends_on` to carry.
2301 Item::Document(document) => destination
2302 .source()
2303 .write_document(&ItemWrite {
2304 target: target.clone(),
2305 item: *document,
2306 depends_on: Vec::new(),
2307 })
2308 .await
2309 .map_err(|error| refused(destination, error))?,
2310 };
2311 // A created item can only be journalled here: its id is what the write answers
2312 // with. A create that fails leaves nothing behind — §4.9 makes taking the item
2313 // back the source's own duty, because a write that refused must not leave an item
2314 // nobody asked for.
2315 if target.is_none() {
2316 journal.record(Undo::Created {
2317 kind: created_kind,
2318 id: landed.clone(),
2319 });
2320 }
2321 Ok(landed)
2322 }
2323
2324 /// A configured source that built, for reading an item out of.
2325 fn readable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
2326 let name = self.known(name)?;
2327 if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
2328 return Err(EngineError::SourceRefused {
2329 name: name.to_string(),
2330 error: unavailable.error().clone(),
2331 });
2332 }
2333 self.ready()
2334 .find(|source| source.name() == &name)
2335 .ok_or(EngineError::NoSources)
2336 }
2337}
2338
2339/// How many tasks of a project are read at once while walking it.
2340const PROJECT_PAGE: std::num::NonZeroU32 = std::num::NonZeroU32::new(50).expect("50 is not zero");
2341
2342/// Each source's own running totals, in the order the sources were given.
2343///
2344/// A reading a source could not take is read as that source not metering: what a command
2345/// spent is a report about the work, and a failed reading must not become a failure of the
2346/// work itself. That holds when only one of a source's two readings failed as well, since
2347/// a difference needs both ends.
2348pub(super) async fn readings(sources: &[&ResolvedSource]) -> Vec<Option<Metering>> {
2349 let mut read = Vec::with_capacity(sources.len());
2350 for source in sources {
2351 read.push(source.source().metering().await.ok().flatten());
2352 }
2353 read
2354}
2355
2356/// What the sources spent between two readings of them, or `None` when none of them meters.
2357///
2358/// A source that does not meter contributes nothing and does not make the total zero: a
2359/// command whose sources all declined reports that it cannot say, not that it spent nothing.
2360/// A budget is keyed by its name and unit together, so two sources naming one budget add up
2361/// and two units of one budget stay apart.
2362///
2363/// A source's two readings are a plugin's word, so they are held to [`Metering`]'s contract
2364/// before either is believed, and a pair that breaks it is that source not metering — see
2365/// [`difference`].
2366pub(super) fn spent_between(
2367 before: &[Option<Metering>],
2368 after: &[Option<Metering>],
2369) -> Option<Spent> {
2370 let mut metered = false;
2371 let mut requests = 0_u64;
2372 let mut budgets: BTreeMap<(String, String), (u64, u64)> = BTreeMap::new();
2373 for (before, after) in before.iter().zip(after) {
2374 let (Some(before), Some(after)) = (before, after) else {
2375 continue;
2376 };
2377 let Some((sent, spent)) = difference(before, after) else {
2378 continue;
2379 };
2380 metered = true;
2381 requests = requests.saturating_add(sent);
2382 for (key, measured, modelled) in spent {
2383 let total = budgets.entry(key).or_default();
2384 total.0 = total.0.saturating_add(measured).saturating_add(modelled);
2385 total.1 = total.1.saturating_add(modelled);
2386 }
2387 }
2388 metered.then(|| Spent {
2389 requests,
2390 budgets: budgets
2391 .into_iter()
2392 .map(|((budget, unit), (amount, modelled))| BudgetSpent {
2393 budget,
2394 unit,
2395 amount,
2396 lower_bound: modelled > 0,
2397 })
2398 .collect(),
2399 })
2400}
2401
2402/// One budget's name and unit, and the measured and modelled amounts spent against it.
2403type BudgetDifference = ((String, String), u64, u64);
2404
2405/// What one source sent and spent between two of its readings, or `None` when the pair
2406/// cannot be a running total's.
2407///
2408/// A running total never falls, never drops a budget it has named, and names every budget
2409/// it keeps, once; a pair that breaks any of that is a source that reset its figures or
2410/// reported something else, and a difference taken over it would be a number that measures
2411/// nothing. Such a source is reported as not metering rather than as having spent a clamped
2412/// zero.
2413fn difference(before: &Metering, after: &Metering) -> Option<(u64, Vec<BudgetDifference>)> {
2414 let sent = after.requests.checked_sub(before.requests)?;
2415 for reading in [before, after] {
2416 let mut named = BTreeSet::new();
2417 for budget in &reading.budgets {
2418 if budget.budget.is_empty()
2419 || budget.unit.is_empty()
2420 || !named.insert((&budget.budget, &budget.unit))
2421 {
2422 return None;
2423 }
2424 }
2425 }
2426 let held = |from: &Metering, budget: &onetaskgraph_plugin_api::Metered| {
2427 from.budgets
2428 .iter()
2429 .find(|held| held.budget == budget.budget && held.unit == budget.unit)
2430 .cloned()
2431 };
2432 if before
2433 .budgets
2434 .iter()
2435 .any(|budget| held(after, budget).is_none())
2436 {
2437 return None;
2438 }
2439 let mut spent = Vec::with_capacity(after.budgets.len());
2440 for budget in &after.budgets {
2441 let earlier = held(before, budget);
2442 let measured = budget
2443 .measured
2444 .checked_sub(earlier.as_ref().map_or(0, |held| held.measured))?;
2445 let modelled = budget
2446 .modelled
2447 .checked_sub(earlier.as_ref().map_or(0, |held| held.modelled))?;
2448 spent.push((
2449 (budget.budget.clone(), budget.unit.clone()),
2450 measured,
2451 modelled,
2452 ));
2453 }
2454 Some((sent, spent))
2455}
2456
2457/// One page request against `source`, at the largest page it will serve.
2458fn request_for(source: &ResolvedSource, cursor: Option<Cursor>) -> PageRequest {
2459 PageRequest {
2460 cursor,
2461 limit: source.source().capabilities().max_page_size.max(1),
2462 }
2463}
2464
2465/// Whether writing this item would change what the destination already holds.
2466///
2467/// A free function over the state already read rather than a method that reads it again:
2468/// the same answer is wanted where the item is landed and where a repeat copy of a project
2469/// decides whether it settled, and a second read there is a second round trip for nothing.
2470fn changes(
2471 held: Option<&Prior>,
2472 item: &Planned,
2473 target: &NativeId,
2474 project: &Option<NativeId>,
2475 edges: &[DependencyEdge],
2476 delivers: &[TaskRef],
2477 destination: &SourceName,
2478) -> bool {
2479 let Some(held) = held else {
2480 return true;
2481 };
2482 let outgoing = outgoing(
2483 item,
2484 target.clone(),
2485 project.clone(),
2486 &recorded(item, Some(held)),
2487 delivers,
2488 Some(held),
2489 );
2490 !same(&held.item, &outgoing, destination) || !same_edges(&held.edges, edges)
2491}
2492
2493/// Remove one item this copy created, through the destination's own write interface.
2494async fn remove(
2495 destination: &ResolvedSource,
2496 kind: Level,
2497 id: &NativeId,
2498) -> Result<(), SourceError> {
2499 match kind {
2500 Level::Task => destination.source().delete_task(id).await,
2501 Level::Project => destination.source().delete_project(id).await,
2502 Level::Document => destination.source().delete_document(id).await,
2503 }
2504}
2505
2506/// Write one item back exactly as the destination held it before this copy.
2507async fn restore(
2508 destination: &ResolvedSource,
2509 id: &NativeId,
2510 prior: &Prior,
2511) -> Result<(), SourceError> {
2512 match &prior.item {
2513 Item::Task(task) => destination
2514 .source()
2515 .write_task(&ItemWrite {
2516 target: Some(id.clone()),
2517 item: (**task).clone(),
2518 depends_on: prior.edges.clone(),
2519 })
2520 .await
2521 .map(|_| ()),
2522 Item::Project(project) => destination
2523 .source()
2524 .write_project(&ItemWrite {
2525 target: Some(id.clone()),
2526 item: (**project).clone(),
2527 depends_on: prior.edges.clone(),
2528 })
2529 .await
2530 .map(|_| ()),
2531 Item::Document(document) => destination
2532 .source()
2533 .write_document(&ItemWrite {
2534 target: Some(id.clone()),
2535 item: (**document).clone(),
2536 depends_on: Vec::new(),
2537 })
2538 .await
2539 .map(|_| ()),
2540 }
2541}
2542
2543/// Refuse a document copy addressed to a source that declares it has none.
2544///
2545/// Read off the declaration rather than by asking, which is what "not asked" means: the
2546/// engine learned at the handshake that this source holds no documents, so it refuses
2547/// naming the source and its plugin instead of sending a read that would be refused there.
2548/// Applied at both ends of a copy — a source with no documents holds nothing to copy out,
2549/// and a destination with none has nowhere to put one.
2550fn documentary(source: &ResolvedSource) -> Result<(), EngineError> {
2551 if source.source().capabilities().documents.is_native() {
2552 return Ok(());
2553 }
2554 Err(EngineError::NoDocuments {
2555 name: source.name().to_string(),
2556 kind: source.kind().to_owned(),
2557 })
2558}
2559
2560/// One source failing while a copy was mid-flight.
2561fn refused(source: &ResolvedSource, error: SourceError) -> EngineError {
2562 EngineError::SourceRefused {
2563 name: source.name().to_string(),
2564 error,
2565 }
2566}
2567
2568/// Every forward edge at one item, walked to exhaustion one page at a time.
2569async fn forward_edges(
2570 source: &ResolvedSource,
2571 id: &NativeId,
2572 kind: Level,
2573) -> Result<Vec<DependencyEdge>, EngineError> {
2574 // A document has no edges to walk, and asking for them would mean asking a source for
2575 // a graph the contract says nothing may point into.
2576 if kind == Level::Document {
2577 return Ok(Vec::new());
2578 }
2579 let mut edges = Vec::new();
2580 let mut cursor: Option<Cursor> = None;
2581 loop {
2582 let asked = cursor.clone();
2583 let request = request_for(source, cursor);
2584 let page = match kind {
2585 Level::Task | Level::Document => {
2586 source
2587 .source()
2588 .task_dependencies(id, Direction::DependsOn, &request)
2589 .await
2590 }
2591 Level::Project => {
2592 source
2593 .source()
2594 .project_dependencies(id, Direction::DependsOn, &request)
2595 .await
2596 }
2597 }
2598 .map_err(|error| refused(source, error))?;
2599 fits(page.items.len(), request.limit).map_err(|error| refused(source, error))?;
2600 edges.extend(page.items);
2601 unrepeated(
2602 page.next.as_ref(),
2603 asked.as_ref(),
2604 "an item's dependencies were being read for a copy",
2605 )
2606 .map_err(|error| refused(source, error))?;
2607 match page.next {
2608 Some(next) => cursor = Some(next),
2609 None => return Ok(edges),
2610 }
2611 }
2612}
2613
2614/// The location string one record reports, when it reports a usable one.
2615///
2616/// Either variant's own `String`, and `None` for a record the source gave no location for
2617/// or gave an empty string for: there is nothing to look for in a document's content and
2618/// nothing to point a reader at.
2619fn located(location: Option<&Location>) -> Option<String> {
2620 let (Location::Path(held) | Location::Url(held)) = location?;
2621 (!held.is_empty()).then(|| held.clone())
2622}
2623
2624/// Record one candidate referent, when its source said where it is.
2625fn note(
2626 into: &mut Vec<Referent>,
2627 id: GlobalId,
2628 level: Level,
2629 location: Option<&Location>,
2630 metadata: &BTreeMap<String, Value>,
2631) {
2632 if let Some(location) = located(location) {
2633 into.push(Referent {
2634 id,
2635 origin: origin_of(metadata),
2636 level,
2637 location,
2638 });
2639 }
2640}
2641
2642/// What every location string this document names becomes, longest first.
2643///
2644/// Longest first because a shorter location may start where a longer one does — a
2645/// project's directory and a task's file under it — and the longer of the two is the
2646/// record that occurrence names.
2647fn table_for(referents: &[Referent], counterparts: &Counterparts) -> Vec<(String, Resolution)> {
2648 let mut table: Vec<(String, Resolution)> = Vec::new();
2649 for referent in referents {
2650 // Two referents reporting one location string: an occurrence of it cannot be
2651 // attributed to either, and a rewrite would be *confidently wrong* rather than
2652 // merely unhelpful. So neither is chosen and both occurrences are counted.
2653 if let Some(held) = table
2654 .iter_mut()
2655 .find(|(location, _)| location == &referent.location)
2656 {
2657 held.1 = Resolution::Ambiguous;
2658 continue;
2659 }
2660 table.push((referent.location.clone(), counterparts.resolve(referent)));
2661 }
2662 table.sort_by_key(|(location, _)| std::cmp::Reverse(location.len()));
2663 table
2664}
2665
2666/// Whether `content` holds `location` at least once, stopped on both sides.
2667fn holds(content: &str, location: &str) -> bool {
2668 (0..content.len()).any(|at| delimited_at(content, at, location))
2669}
2670
2671/// Whether `location` occurs at `at` **stopped on both sides** — by
2672/// [`stops_a_location`], or by the end of the content — rather than as part of a longer
2673/// location-like string.
2674///
2675/// A location string occurring inside a longer one is a different string naming a
2676/// different record: `/…/tasks/p/t.md` must not be rewritten inside `/…/tasks/p/t.md.bak`,
2677/// `https://example.invalid/1` must not be rewritten inside `https://example.invalid/12`,
2678/// and a project's location that is a directory prefix of a task's must not be rewritten
2679/// inside that task's. What deciding it this way costs is stated on [`stops_a_location`].
2680fn delimited_at(content: &str, at: usize, location: &str) -> bool {
2681 if !content.is_char_boundary(at) || !content[at..].starts_with(location) {
2682 return false;
2683 }
2684 let before = content[..at].chars().next_back();
2685 let after = content[at + location.len()..].chars().next();
2686 stops_a_location(before) && stops_a_location(after)
2687}
2688
2689/// Whether a character cannot continue a path or a link, so a location string beside one
2690/// ends there.
2691///
2692/// Stated as what *stops* a location rather than as what one may contain, because the
2693/// second list is unbounded — a path may hold very nearly any byte, and a URL more. Every
2694/// character not named here continues, which is what leaves the three cases above alone;
2695/// the end of the content counts as a stop. The set is what the artifact this exists for
2696/// really wraps a bare path in — a backtick in a table cell — plus the delimiters prose
2697/// and Markdown put next to one.
2698///
2699/// **Sentence punctuation is deliberately absent, and that is a stated cost rather than an
2700/// oversight.** `.`, `!`, `?` and `:` each equally *continue* a real location — `/…/t.md`
2701/// and `/…/t.md.bak` are two files, `…/1` and `…/1?q=2` two pages — so admitting them as
2702/// stops would rewrite one record's location into another's. The price is that a location
2703/// written bare at the end of a sentence is not recognised at all: its text is left
2704/// byte-for-byte and it is counted in neither figure, exactly as a reference to another
2705/// project's record is. That is the direction to be wrong in, because this edits the
2706/// content of somebody's document, where a confidently wrong rewrite is worse than one
2707/// that never happens.
2708fn stops_a_location(character: Option<char>) -> bool {
2709 match character {
2710 None => true,
2711 Some(character) => character.is_whitespace() || "`\"'()[]{}<>|,;".contains(character),
2712 }
2713}
2714
2715/// One document's content with every whole reference rewritten, and what that took.
2716///
2717/// A location string with no confident counterpart is left **byte-for-byte** as it was
2718/// rather than removed or guessed at, and so is every character of the content that is not
2719/// a rewritten reference. Nothing is added and nothing is reformatted.
2720fn substitute(content: &str, table: &[(String, Resolution)]) -> (String, Counted) {
2721 let mut written = String::with_capacity(content.len());
2722 let mut counts = Counted::default();
2723 let mut at = 0;
2724 while at < content.len() {
2725 if let Some((location, resolution)) = table
2726 .iter()
2727 .find(|(location, _)| delimited_at(content, at, location))
2728 {
2729 match resolution {
2730 Resolution::Rewrite(there) => {
2731 written.push_str(there);
2732 counts.rewritten += 1;
2733 }
2734 // Both left-alone outcomes count as unresolved on the branch that counts
2735 // them, which is what really holds `ambiguous` at or below `unresolved`.
2736 Resolution::NoCounterpart => {
2737 written.push_str(location);
2738 counts.unresolved += 1;
2739 }
2740 Resolution::Ambiguous => {
2741 written.push_str(location);
2742 counts.unresolved += 1;
2743 counts.ambiguous += 1;
2744 }
2745 }
2746 at += location.len();
2747 continue;
2748 }
2749 let character = content[at..]
2750 .chars()
2751 .next()
2752 .expect("a character at a boundary this walk only ever lands on");
2753 written.push(character);
2754 at += character.len_utf8();
2755 }
2756 (written, counts)
2757}
2758
2759/// The origin one item records, when it records a usable one.
2760fn origin_of(metadata: &BTreeMap<String, Value>) -> Option<GlobalId> {
2761 metadata
2762 .get(GlobalId::ORIGIN_KEY)?
2763 .as_str()?
2764 .parse::<GlobalId>()
2765 .ok()
2766}
2767
2768/// The title and metadata of either kind of item.
2769fn described(item: &Item) -> (&str, &BTreeMap<String, Value>) {
2770 match item {
2771 Item::Task(task) => (&task.title, &task.metadata),
2772 Item::Project(project) => (&project.title, &project.metadata),
2773 Item::Document(document) => (&document.title, &document.metadata),
2774 }
2775}
2776
2777/// The item as the destination should hold it.
2778///
2779/// `url`, `location`, `created_at` and `updated_at` are the destination's own and are
2780/// never written — where the *source* holds an item says nothing about where the
2781/// destination does, which is why a copied document does not arrive claiming the path or
2782/// the link its source reported. The two reserved keys this product encodes typed fields
2783/// under are removed, because those fields travel as themselves — leaving the encoding
2784/// beside them would have the destination hold one thing twice, and disagree with itself
2785/// the moment one changed.
2786///
2787/// A task's `delivers` is `delivers`, already resolved against the destination, and its
2788/// `delivered_by` is the one the destination holds — never the source's, and empty for an
2789/// item this copy creates: that list is the store's to keep, at the destination.
2790fn outgoing(
2791 item: &Planned,
2792 id: NativeId,
2793 project: Option<NativeId>,
2794 origin: &Origin,
2795 delivers: &[TaskRef],
2796 held: Option<&Prior>,
2797) -> Item {
2798 match &item.item {
2799 Item::Task(task) => Item::Task(Box::new(Task {
2800 id,
2801 url: None,
2802 location: None,
2803 created_at: None,
2804 updated_at: None,
2805 project,
2806 metadata: carried(&task.metadata, origin),
2807 delivers: delivers.to_vec(),
2808 delivered_by: match held.map(|held| &held.item) {
2809 Some(Item::Task(held)) => held.delivered_by.clone(),
2810 _ => Vec::new(),
2811 },
2812 ..(**task).clone()
2813 })),
2814 Item::Project(project) => Item::Project(Box::new(Project {
2815 id,
2816 url: None,
2817 location: None,
2818 created_at: None,
2819 updated_at: None,
2820 metadata: carried(&project.metadata, origin),
2821 ..(**project).clone()
2822 })),
2823 Item::Document(document) => Item::Document(Box::new(Document {
2824 id,
2825 url: None,
2826 location: None,
2827 created_at: None,
2828 updated_at: None,
2829 project,
2830 metadata: carried(&document.metadata, origin),
2831 ..(**document).clone()
2832 })),
2833 }
2834}
2835
2836/// The id a created item is offered to the destination under.
2837///
2838/// A task whose id is `<its project's id>/<rest>` is offered as `<rest>` under the
2839/// destination project it is filed in, so a destination whose ids are paths files it in that
2840/// project rather than the source's. Any other id, and every update, is left as it is. The
2841/// `a_project_copied_*` journeys in `crates/onetaskgraph/tests/e2e/copy.rs` hold the Markdown
2842/// plugin's id shape to this rule.
2843fn created_id(item: &Item, filed: Option<&NativeId>) -> NativeId {
2844 if let (Item::Task(task), Some(filed)) = (item, filed)
2845 && let Some(own) = &task.project
2846 && let Some(rest) = task
2847 .id
2848 .as_str()
2849 .strip_prefix(own.as_str())
2850 .and_then(|rest| rest.strip_prefix('/'))
2851 && !rest.is_empty()
2852 {
2853 return NativeId(format!("{}/{rest}", filed.as_str()));
2854 }
2855 item.id().clone()
2856}
2857
2858/// The metadata a copy carries: the caller's own keys untouched, and the origin settled.
2859///
2860/// The key is removed before it is settled rather than overwritten, because the item being
2861/// copied carries an origin of its own and [`Origin::Keeps`] must not let it through.
2862fn carried(metadata: &BTreeMap<String, Value>, origin: &Origin) -> BTreeMap<String, Value> {
2863 let mut carried = metadata.clone();
2864 carried.remove(Repository::METADATA_KEY);
2865 carried.remove(DependencyEdge::RECORDED_KEY);
2866 carried.remove(TaskRef::DELIVERS_KEY);
2867 carried.remove(TaskRef::DELIVERED_BY_KEY);
2868 carried.remove(GlobalId::ORIGIN_KEY);
2869 let held = match origin {
2870 Origin::Records(id) => Some(Value::String(id.to_string())),
2871 Origin::Keeps(held) => held.clone(),
2872 };
2873 if let Some(held) = held {
2874 carried.insert(GlobalId::ORIGIN_KEY.to_owned(), held);
2875 }
2876 carried
2877}
2878
2879/// What one landed item records at [`GlobalId::ORIGIN_KEY`].
2880enum Origin {
2881 /// The qualified id this item was copied from, as the id type rather than as its
2882 /// spelling: the key holds a [`GlobalId`] and nothing else may be recorded there.
2883 Records(GlobalId),
2884 /// Whatever the destination already holds there — `None` when it holds nothing, which
2885 /// is written as the key being absent rather than as a null.
2886 ///
2887 /// A [`Value`] and not a [`GlobalId`], because this variant does not interpret what it
2888 /// carries: it is the destination's own metadata entry, held for the length of one
2889 /// write and put back exactly as it was read. Parsing it would turn a value a
2890 /// destination holds and this engine cannot read into a value this engine deletes,
2891 /// which is the opposite of what keeping it means.
2892 Keeps(Option<Value>),
2893}
2894
2895/// Which of the two a copy of this item does.
2896///
2897/// A copy that reached its target by rule 1 is a copy-back: the item being copied names
2898/// the destination item, so the destination is the *original* and the id being copied
2899/// belongs to the copy that came out of it. Recording that id there would overwrite the
2900/// original's own provenance — and with it the correspondence every later copy from the
2901/// source it was authored in depends on. That copy would then match nothing and create a
2902/// second item beside the one it meant to update, which is the whole failure: nothing is
2903/// reported, and whoever reads that board now has two. So a copy-back leaves the
2904/// destination's origin exactly as the destination holds it, absent included, and every
2905/// other copy records the id it was copied from.
2906fn recorded(item: &Planned, held: Option<&Prior>) -> Origin {
2907 if let Target::Update {
2908 found: Found::Origin,
2909 ..
2910 } = &item.target
2911 {
2912 return Origin::Keeps(
2913 held.and_then(|held| described(&held.item).1.get(GlobalId::ORIGIN_KEY).cloned()),
2914 );
2915 }
2916 Origin::Records(item.source.clone())
2917}
2918
2919/// Whether the destination already reads exactly as this copy would leave it.
2920///
2921/// The destination's own `url` and timestamps are excluded because a copy never writes
2922/// them, so a difference there is not one this copy would close.
2923fn same(held: &Item, outgoing: &Item, destination: &SourceName) -> bool {
2924 match (held, outgoing) {
2925 (Item::Task(held), Item::Task(outgoing)) => {
2926 // Qualified before they are compared, so `T-1` and `folder:T-1` at the destination
2927 // `folder` are the one entry they are.
2928 targets(&held.delivers, destination) == targets(&outgoing.delivers, destination)
2929 && targets(&held.delivered_by, destination)
2930 == targets(&outgoing.delivered_by, destination)
2931 && held.title == outgoing.title
2932 && held.content == outgoing.content
2933 && held.status == outgoing.status
2934 && held.priority == outgoing.priority
2935 && held.labels == outgoing.labels
2936 && held.project == outgoing.project
2937 && held.metadata == outgoing.metadata
2938 && held.repositories == outgoing.repositories
2939 }
2940 (Item::Project(held), Item::Project(outgoing)) => {
2941 held.title == outgoing.title
2942 && held.content == outgoing.content
2943 && held.status == outgoing.status
2944 && held.labels == outgoing.labels
2945 && held.metadata == outgoing.metadata
2946 && held.repositories == outgoing.repositories
2947 }
2948 // No status, because a document has none; no edges, because it is in no graph.
2949 (Item::Document(held), Item::Document(outgoing)) => {
2950 held.title == outgoing.title
2951 && held.content == outgoing.content
2952 && held.labels == outgoing.labels
2953 && held.project == outgoing.project
2954 && held.metadata == outgoing.metadata
2955 && held.repositories == outgoing.repositories
2956 }
2957 _ => false,
2958 }
2959}
2960
2961/// Whether the destination's forward edges already say what this copy would write.
2962fn same_edges(held: &[DependencyEdge], outgoing: &[DependencyEdge]) -> bool {
2963 let ends = |edges: &[DependencyEdge]| {
2964 let mut ends: Vec<(String, ItemKind, DependencyKind)> = edges
2965 .iter()
2966 .map(|edge| (edge.to.id().to_owned(), edge.to.kind, edge.kind))
2967 .collect();
2968 ends.sort_by(|left, right| left.0.cmp(&right.0));
2969 ends
2970 };
2971 ends(held) == ends(outgoing)
2972}
2973
2974/// Each read edge as the destination should record it, or `None` when its far end is a
2975/// member of this copy whose destination id is not known yet.
2976fn mapped_edges(
2977 edges: &[DependencyEdge],
2978 origin: &SourceName,
2979 destination: &ResolvedSource,
2980 copied: &[GlobalId],
2981 written: &BTreeMap<String, NativeId>,
2982) -> Vec<Option<DependencyEdge>> {
2983 edges
2984 .iter()
2985 .map(|edge| {
2986 let far = GlobalId::new(origin.clone(), NativeId(edge.to.id().to_owned()));
2987 let id = if let Some(native) = names(&edge.to, destination.name()) {
2988 // A far end already qualified to the destination's own source is that
2989 // source's own item, so it is written the way that source names its own:
2990 // unqualified. Leaving it qualified would have the destination hold an
2991 // edge into itself written as if it left, which is the one spelling the
2992 // reserved key exists to keep for edges that really do.
2993 Some(native)
2994 } else if !edge.to.is_qualified() && copied.contains(&far) {
2995 // A member of this copy, inside one source included: a copy there lands
2996 // beside the item it was read from, so the edge names the copy rather than
2997 // the original the copy's own project does not hold.
2998 written.get(&far.to_string()).map(|native| native.0.clone())
2999 } else if edge.to.is_qualified() || origin == destination.name() {
3000 // Already naming a source of its own, or a copy inside one source where
3001 // the far end's own id is the destination's id.
3002 Some(edge.to.id().to_owned())
3003 } else {
3004 Some(far.to_string())
3005 }?;
3006 DependencyEndpoint::new(id, edge.to.kind)
3007 .ok()
3008 .map(|to| DependencyEdge {
3009 from: edge.from.clone(),
3010 to,
3011 kind: edge.kind,
3012 })
3013 })
3014 .collect()
3015}
3016
3017/// Each `delivers` entry of one planned task as the destination should hold it, or `None`
3018/// where it names a member of this copy whose destination id is not known yet. Empty for
3019/// anything but a task.
3020fn delivers_of(
3021 item: &Planned,
3022 destination: &ResolvedSource,
3023 copied: &[GlobalId],
3024 written: &BTreeMap<String, NativeId>,
3025) -> Vec<Option<TaskRef>> {
3026 match &item.item {
3027 Item::Task(task) => mapped_delivers(
3028 &task.delivers,
3029 &item.source.source,
3030 destination,
3031 copied,
3032 written,
3033 ),
3034 Item::Project(_) | Item::Document(_) => Vec::new(),
3035 }
3036}
3037
3038/// Each `delivers` entry as the destination should hold it, or `None` when it names a member
3039/// of this copy whose destination id is not known yet.
3040///
3041/// An entry naming a member of the copied set becomes that member's own id at the
3042/// destination — bare, because the member is the destination's, unless the id holds a colon
3043/// a bare entry would be misread by. Every other entry is carried through qualified, so a
3044/// bare one read at `origin` goes on naming a task of `origin`.
3045fn mapped_delivers(
3046 entries: &[TaskRef],
3047 origin: &SourceName,
3048 destination: &ResolvedSource,
3049 copied: &[GlobalId],
3050 written: &BTreeMap<String, NativeId>,
3051) -> Vec<Option<TaskRef>> {
3052 entries
3053 .iter()
3054 .map(|entry| {
3055 let qualified = entry.in_source(origin);
3056 let Ok(far) = qualified.as_str().parse::<GlobalId>() else {
3057 return Some(qualified);
3058 };
3059 if !copied.contains(&far) {
3060 return Some(qualified);
3061 }
3062 let native = written.get(&far.to_string())?;
3063 Some(if native.as_str().contains(':') {
3064 TaskRef::qualified(destination.name(), native)
3065 } else {
3066 TaskRef::new(native.as_str())
3067 .unwrap_or_else(|_| TaskRef::qualified(destination.name(), native))
3068 })
3069 })
3070 .collect()
3071}
3072
3073/// How many of one task's `delivers` entries name a member of the copied set.
3074fn members_named(entries: &[TaskRef], origin: &SourceName, copied: &[GlobalId]) -> u64 {
3075 let named = entries
3076 .iter()
3077 .filter(|entry| {
3078 entry
3079 .in_source(origin)
3080 .as_str()
3081 .parse::<GlobalId>()
3082 .is_ok_and(|far| copied.contains(&far))
3083 })
3084 .count();
3085 u64::try_from(named).unwrap_or(u64::MAX)
3086}
3087
3088/// The entries that could be resolved, which is every one of them on the second pass.
3089fn resolved_entries(entries: &[Option<TaskRef>]) -> Vec<TaskRef> {
3090 entries.iter().flatten().cloned().collect()
3091}
3092
3093/// The native id a qualified endpoint names at `destination`, when it names one there.
3094fn names(endpoint: &DependencyEndpoint, destination: &SourceName) -> Option<String> {
3095 if !endpoint.is_qualified() {
3096 return None;
3097 }
3098 let id: GlobalId = endpoint.id().parse().ok()?;
3099 (&id.source == destination).then_some(id.native.0)
3100}
3101
3102/// The edges that could be resolved, which is every one of them on the second pass.
3103fn resolved(edges: &[Option<DependencyEdge>]) -> Vec<DependencyEdge> {
3104 edges.iter().flatten().cloned().collect()
3105}
3106
3107/// Refuse an item of a member copy whose edge names a member that copy does not carry
3108/// and whose destination id nothing records.
3109///
3110/// `unrecorded` is empty for every other copy, which is what makes this a no-op there. An
3111/// edge already naming a source of its own, and every edge of a copy inside one source,
3112/// is written as it was read by [`mapped_edges`], so neither can need a recorded origin.
3113fn unrecorded_far_end(
3114 item: &Planned,
3115 destination: &ResolvedSource,
3116 unrecorded: &[GlobalId],
3117) -> Result<(), EngineError> {
3118 if &item.source.source == destination.name() {
3119 return Ok(());
3120 }
3121 for edge in &item.edges {
3122 if edge.to.is_qualified() {
3123 continue;
3124 }
3125 let far = GlobalId::new(
3126 item.source.source.clone(),
3127 NativeId(edge.to.id().to_owned()),
3128 );
3129 if unrecorded.contains(&far) {
3130 return Err(EngineError::UnrecordedMember {
3131 item: item.source.clone(),
3132 member: far,
3133 destination: destination.name().clone(),
3134 });
3135 }
3136 }
3137 Ok(())
3138}