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