//! The copy verb: one item out of one source and into another, by the rules that make a
//! second copy an update rather than a duplicate.
//!
//! Correspondence lives on the items and never in a table. A copied item carries
//! [`GlobalId::ORIGIN_KEY`], whose value is the qualified id it was copied from, and the
//! item it was copied from carries [`MetadataKey::COPIES_KEY`], an object naming its
//! counterpart at each destination it was copied into. The rules below read exactly those —
//! so nothing here is written down outside the plugin that owns the item, and the invariant
//! this engine is built around is untouched.
//!
//! 0. **Follow the link.** An item whose `onetaskgraph.copies` names an item at the
//! destination has that item read by id, and when that item's own origin names this one
//! back it is the target: one read, and no scan. A link naming nothing there any more is
//! refused as `stale-link` unless `--recreate` is given, for the reason a stale origin
//! is; a link naming an item that no longer names this one back — somebody re-pointed it
//! — is ignored, and the rules below decide.
//! 1. **Follow the origin.** An item already carrying an origin whose source half is the
//! destination names the destination item *directly*, and the copy updates it. This is
//! the half that makes an edit's copy-back an update: the local file came from the
//! remote item and knows which one.
//! 2. **Search by origin.** Otherwise the destination is scanned, one page at a time, for
//! an item whose origin is the id being copied. Found, the copy updates it; not found,
//! the copy creates one carrying that origin.
//!
//! Which rule found the item decides what the copy records there. A copy that got its
//! target from rule 1 is a copy-back — the destination is the *original*, and the item
//! being copied is the one that came from it — so the destination keeps the origin it
//! already holds, holding none included. Every other copy records the id it was copied
//! from. See [`recorded`] for what stamping a copy-back's own id there costs.
//!
//! The same distinction decides the link. Once the whole copy has landed, every item whose
//! target was found by any rule but rule 1 — or was created — has its link for this
//! destination recorded or refreshed through its source's narrow metadata write, and that
//! write joins the undo journal like every destination write. Rule 1 records nothing: that
//! correspondence is already written down, on the destination item.
//!
//! A destination write is at the user's explicit request, names its destination, goes
//! through that source's own write interface into that source's own store, and is never
//! read back to answer a query. That is what makes it a write and not a cache.
use std::collections::{BTreeMap, BTreeSet};
use onetaskgraph_plugin_api::{
AssetPayload, AssetUploads, AssetWrite, Capabilities, Cursor, DependencyEdge,
DependencyEndpoint, DependencyKind, Direction, Document, DocumentQuery, ItemKind, ItemWrite,
Location, MetadataKey, MetadataMatch, Metering, NativeId, Page, PageRequest, Project,
ProjectQuery, Repository, SourceError, SourceName, StatusCategory, Task, TaskQuery, TaskRef,
TextFields, TextQuery,
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use crate::GlobalId;
use crate::config::Placement;
use crate::resolve::ResolvedSource;
use crate::template::{Sha256Digest, TemplateProvenance, body_digest};
use super::assets;
use super::delivery::{Delivered, targets};
use super::fetch::{fits, unrepeated};
use super::local::ProjectSelector;
use super::narrow::holds_priority;
use super::{
DocumentFilters, DocumentRequest, Engine, EngineError, Filters, LeftBehind, Paging, Qualified,
TaskRequest,
};
/// A request to copy work into one configured destination.
#[derive(Debug, Clone)]
pub struct CopyRequest {
/// The qualified items to copy, in the order they were named.
pub items: CopyItems,
/// What those ids name, and what comes with them.
pub scope: CopyScope,
/// The configured source to copy into — a source name, never a qualified id.
pub destination: SourceName,
/// How to re-establish a correspondence the two origin rules cannot find.
pub match_by: Option<MatchBy>,
/// Whether an origin naming nothing at the destination falls through to the search
/// rule instead of refusing.
pub recreate: bool,
/// Whether the caller asserts the destination holds no counterpart of any item named,
/// so each is created there without the correspondence lookup — no origin search of the
/// destination at all.
///
/// Sound for a caller that has just asked the destination itself: the origin query is
/// an index that lags a write, so repeating it here would be behind by exactly as much as
/// the caller's own was and could find nothing the caller's did not. Refused beside
/// [`match_by`](Self::match_by) and [`recreate`](Self::recreate), which are ways of
/// looking, and for an item that itself records a counterpart at the destination — a
/// link or an origin naming it — which is a carrier the caller's assertion overlooked.
// llmlint: ignore-block[invalid_states_unrepresentable] One enum of the ways a copy finds its target would make `create` beside `match_by` or `recreate` unrepresentable, but only by replacing `match_by` and `recreate`, which every Rust caller of this request sets by name and the command line maps flag for flag — rewriting what each existing caller writes. `create` is one more field beside them, as `recreate` and `dry_run` were, and the one combination it makes possible that means nothing is refused by `Engine::copy`, as `EngineError::CreateWith`, before anything is read.
pub create: bool,
// llmlint: ignore-end[invalid_states_unrepresentable]
/// Whether to perform every read and no write.
pub dry_run: bool,
}
/// A way of looking for a counterpart that [`CopyRequest::create`] says not to take, named as
/// the command line spells it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CopyLookup {
/// [`CopyRequest::match_by`], `--match-by`.
MatchBy,
/// [`CopyRequest::recreate`], `--recreate`.
Recreate,
}
impl std::fmt::Display for CopyLookup {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(match self {
Self::MatchBy => "--match-by",
Self::Recreate => "--recreate",
})
}
}
/// The items one copy names: at least one, because a copy naming none is not a copy.
///
/// A newtype rather than a bare `Vec`, for the reason [`Repository`] is one: the empty
/// list is not a copy of nothing, it is a caller mistake, and a type that can hold it
/// leaves every reader to decide what it means — a report with no entries, an error, a
/// silent success. None of those is better than not being able to say it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CopyItems(Vec<GlobalId>);
impl CopyItems {
/// The items a caller named, or `None` when they named none.
#[must_use]
pub fn new(items: Vec<GlobalId>) -> Option<Self> {
(!items.is_empty()).then_some(Self(items))
}
/// The items, in the order they were named.
#[must_use]
pub fn as_slice(&self) -> &[GlobalId] {
&self.0
}
}
/// What the ids a copy names are, and what travels with them.
///
/// One value rather than a kind beside a flag, because three of the four combinations
/// those two would make are real and the fourth — tasks, with the tasks of each also
/// copied — means nothing. The member list is a variant for the same reason: it narrows a
/// project copy that carries its tasks, and it has nothing to say to the other three.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CopyScope {
/// The ids name tasks, and only those tasks are copied.
Tasks,
/// The ids name projects.
Projects {
/// Whether the tasks in each project are copied too.
tasks: bool,
},
/// The ids name projects, and of the tasks in them exactly these are copied.
///
/// A member the list does not name is not read at the destination, not compared, not
/// written and not reported, and no walk for what the copy left behind runs. A named
/// member with no counterpart there is still created: the list narrows which members
/// are read and written, never which outcomes are possible.
///
/// An edge from a copied item to a member the list does not name resolves to the
/// destination id that member's own [`GlobalId::ORIGIN_KEY`] records at the source,
/// without reading the destination for it — or, for a copy into a source with `routes`,
/// the id its own `onetaskgraph.copies` link records in any source the copy reaches.
/// When that member records neither, the copy is refused before anything is written.
Members(CopyItems),
/// The ids name documents, and only those documents are copied.
///
/// Nothing travels with a document: it takes part in no dependency graph, and it holds
/// nothing of its own the way a project holds tasks.
Documents,
}
/// The caller-named escape for a correspondence neither origin rule can find.
///
/// A person editing Markdown who deletes or corrupts the origin key leaves an item rule 1
/// cannot use and rule 2 cannot find, and the next copy would create a second item. This
/// is how that is re-established without hand-editing ids.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MatchBy {
/// Match the item whose title is the same.
Title,
/// Match the item whose value at this metadata key is the same.
Metadata(String),
}
impl MatchBy {
/// The spelling a caller types, `title` or any metadata key.
#[must_use]
pub fn parse(key: &str) -> Self {
if key == "title" {
Self::Title
} else {
Self::Metadata(key.to_owned())
}
}
}
/// What a copy did, one entry per item.
///
/// The same per-item outcomes reach every consumer: the machine-readable output renders
/// this, the rendered output renders this, and a Rust caller is handed it.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct CopyReport {
/// One entry per item the copy considered, in the order it considered them.
pub items: Vec<CopyOutcome>,
// Three flat scalars rather than one nested object, and a `!skip_serializing_if` beside
// every skip: both are load-bearing for the generated SDKs rather than matters of
// taste, and AGENTS.md's note on what a copied document's references are pointed at is
// where that reasoning lives.
/// Reference occurrences the copy rewrote to the destination's own location for the
/// record they name.
///
/// A silent bound is indistinguishable from a bug, so the copy says what it did to the
/// references the documents it carried hold. This and the two below are totals over the
/// whole invocation rather than figures per document, and all three default to zero, so
/// a consumer written against the output before they existed is unaffected.
///
/// **What these figures do not claim.** The referent set is a document's own project,
/// so a reference to a record in a *different* project is never recognised at all and
/// cannot appear in [`Self::references_unresolved`] either. These are the references
/// the copy recognised; they are not a census of every reference a document holds.
/// Noticing an out-of-scope reference would need exactly the unbounded destination walk
/// this design refuses.
#[serde(default, skip_serializing_if = "nothing_to_report")]
#[schemars(!skip_serializing_if)]
pub references_rewritten: u64,
/// Reference occurrences the copy recognised and left byte-for-byte as they were,
/// because the correspondence could not be established.
#[serde(default, skip_serializing_if = "nothing_to_report")]
#[schemars(!skip_serializing_if)]
pub references_unresolved: u64,
/// How many of [`Self::references_unresolved`] were left alone because the
/// correspondence was **ambiguous** rather than merely absent. A sub-count, never
/// larger than it.
///
/// Split out because the two mean different things to a reader. A reference with no
/// counterpart is ordinary and expected under the bound above — the design working. An
/// ambiguous one says the destination holds duplicate records for one work item, or the
/// source reports one location for two records, and re-running the copy will never
/// clear it.
#[serde(default, skip_serializing_if = "nothing_to_report")]
#[schemars(!skip_serializing_if)]
// llmlint: ignore[invalid_states_unrepresentable] JSON Schema cannot express an
// inequality between two numbers, so a private constructor here would hold this in one
// consumer of three while both SDKs' generated models went on admitting it. What holds
// it is `substitute`: `Resolution` has no variant that counts an occurrence ambiguous
// without counting it unresolved.
pub references_ambiguous: u64,
/// `delivers` entries the copy rewrote to the destination's own id for a member of the
/// copied set, over the whole invocation. Every other entry arrives qualified and is not
/// counted. Left out when zero, as the three figures above are.
#[serde(default, skip_serializing_if = "nothing_to_report")]
#[schemars(!skip_serializing_if)]
pub delivers_rewritten: u64,
/// One entry per delivered task the copy kept in step with a task it landed, after the
/// whole copy was complete — see `task status set`, which reports the same entries. Left
/// out when there were none.
///
/// A failed entry does not undo the copy: the tasks it landed stay landed, and the
/// command exits `4`.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
#[schemars(!skip_serializing_if)]
pub delivered: Vec<Delivered>,
/// What this copy spent, summed over the sources in the command that meter their own
/// requests — and absent, never zero, when none of them does.
///
/// Omitted from the wire when absent, like the three figures above, so a consumer
/// written against the output before it existed reads the same document.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub spent: Option<Spent>,
}
/// Whether one of [`CopyReport`]'s reference figures has anything to say.
///
/// A copy that recognised no reference reports that by leaving the figure out rather than
/// by writing a nought, so the machine output of a task or project copy is exactly what it
/// was before these figures existed. The human rendering says it in words either way,
/// because a reader there needs to be told the copy looked.
fn nothing_to_report(figure: &u64) -> bool {
*figure == 0
}
/// What one command spent, summed over the sources in it that meter their own requests.
///
/// **Source-owned.** Every figure is what a source said it sent and spent while the command
/// ran, read through [`TaskSource::metering`](onetaskgraph_plugin_api::TaskSource::metering)
/// before the command and again after it. The engine adds the differences up by name and
/// interprets none of them, which is why the budget and unit names are open vocabulary.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct Spent {
/// How many HTTP requests those sources sent for this command.
pub requests: u64,
/// What those requests spent, one entry per budget and unit, ordered by budget name.
pub budgets: Vec<BudgetSpent>,
}
/// What one command spent against one budget.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct BudgetSpent {
/// The budget, as the source names it — `graphql`, `rest`.
// 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.
pub budget: String,
/// What it is metered in — `points`, `requests`.
// llmlint: ignore[invalid_states_unrepresentable] As `budget` above.
pub unit: String,
/// How much was spent against it, in that unit.
pub amount: u64,
/// Whether any part of `amount` was modelled by a source rather than reported by its
/// backend or counted, which makes `amount` a lower bound on what the backend charged
/// rather than a measurement of it.
pub lower_bound: bool,
}
/// One document's reference figures, before they are folded into the invocation's.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
struct Counted {
/// Occurrences rewritten.
rewritten: u64,
/// Occurrences recognised and left alone.
unresolved: u64,
/// How many of those were ambiguous.
ambiguous: u64,
}
impl Counted {
/// Fold one document's figures into the invocation's.
fn add(&mut self, other: Self) {
self.rewritten += other.rewritten;
self.unresolved += other.unresolved;
self.ambiguous += other.ambiguous;
}
}
/// What happened to one item.
///
/// `action` and `destination` are one value rather than two fields side by side: an
/// updated item without a destination id, or an orphan without one, are states this type
/// must not be able to say — the id *is* what those outcomes are about. The one outcome
/// that legitimately has none is a dry run that would create, because nothing was
/// created and there is no id to report.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
pub struct CopyOutcome {
/// The qualified id the item was read from.
pub source: GlobalId,
/// What happened to it, and where.
#[serde(flatten)]
pub action: CopyAction,
/// Which source a routed copy placed it in, and the index of the route entry that
/// matched — `null` there when none did.
///
/// Present for every item a copy into a source declaring `routes` landed, a dry run's
/// included, so a would-be create still says where it would go; absent for a copy into
/// a source with none, whose output is what it always was, and for an orphan, which was
/// not placed at all.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub placed: Option<Placement>,
}
/// Which rule found the destination counterpart an item was updated at, or found already
/// reading as it does.
///
/// Tried in this order, and the first that answers is the one reported: the link the item
/// records, its own origin, a search of the destination for an item recording it as its
/// origin, and the caller's `--match-by`. When none answers, the item is created and says
/// so as [`NoCounterpart`] instead.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum CopyVia {
/// The item's `onetaskgraph.copies` entry for the destination named an item whose own
/// origin names this one back. Found by one read by id, and no scan.
Link,
/// The item's own `onetaskgraph.origin` named the destination item: a copy-back.
Origin,
/// A search of the destination for an item recording this one as its origin.
Scan,
/// A search of the destination by the caller's `--match-by`.
Match,
}
/// What a created item reports as its `via`: no rule found a counterpart.
///
/// One word, and a type of its own rather than a fifth [`CopyVia`], so an item updated at a
/// counterpart cannot say none was found and a created one cannot name a rule that found
/// one.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum NoCounterpart {
/// No counterpart was found, so one was created — or, in a dry run, would have been.
#[default]
Created,
}
/// What a copy did to the `onetaskgraph.copies` entry the copied item holds for the
/// destination.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "kebab-case")]
pub enum CopyLink {
/// The entry was written, because the item held none for this destination or held one
/// naming another item.
Recorded,
/// Nothing was written: the entry already named this destination item, or the item was
/// found by its own origin, where the correspondence is already recorded on the other
/// side.
Unchanged,
/// The item's source cannot hold the entry — it has no write side, or it refuses the
/// narrow metadata write — so the copy landed without recording it.
Unrecorded,
}
impl CopyOutcome {
/// The qualified id this outcome landed on, when it landed on one.
#[must_use]
pub fn destination(&self) -> Option<&GlobalId> {
self.action.destination()
}
}
/// The four things a copy can do to one item, and the id each of them is about.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "action", rename_all = "kebab-case")]
pub enum CopyAction {
// llmlint: ignore[names_match_behavior] `created` is Contract D's serialized action
// for both a completed create and a dry run that would create; the optional destination
// distinguishes those cases, and renaming this public variant would break Rust callers.
/// The destination held no counterpart, so one was created.
Created {
/// The id it was created under, or `null` for a dry run that would have created
/// one — there is no id, because nothing was.
destination: Option<GlobalId>,
/// That no rule found a counterpart. The one word it can be, so a report written
/// before there was a `via` reads as saying it.
#[serde(default)]
via: NoCounterpart,
/// What the copy did to the link the copied item records for this destination at
/// `onetaskgraph.copies`; absent for a dry run, which writes nothing.
#[serde(default, skip_serializing_if = "Option::is_none")]
// llmlint: ignore[invalid_states_unrepresentable] Whether a copy is a dry run is a
// property of the request, which the report does not carry, and the contract this
// lands — shared with the follow-up tooling built against it — says a dry run leaves
// `link` out rather than naming a sixth word for it. What holds it is `Engine::link`:
// the one writer of this field, run for every item a copy that writes has landed and
// for none of a dry run's, which `a_dry_run_says_which_rule_would_answer_and_records_no_link`
// and every `link` assertion in tests/copy_link.rs drive.
link: Option<CopyLink>,
},
/// The destination held a counterpart and it now reads as the source does.
Updated {
/// The item that was updated.
destination: GlobalId,
/// Which rule found it.
via: CopyVia,
/// What the copy did to the copied item's link; absent for a dry run.
#[serde(default, skip_serializing_if = "Option::is_none")]
// llmlint: ignore[invalid_states_unrepresentable] As `link` on `Created` above.
link: Option<CopyLink>,
},
/// The destination held a counterpart that already read that way; nothing was written.
Unchanged {
/// The item that already said it.
destination: GlobalId,
/// Which rule found it.
via: CopyVia,
/// What the copy did to the copied item's link; absent for a dry run.
#[serde(default, skip_serializing_if = "Option::is_none")]
// llmlint: ignore[invalid_states_unrepresentable] As `link` on `Created` above.
link: Option<CopyLink>,
},
/// The destination holds a counterpart the source no longer does. A copy never
/// deletes, so it was left exactly as it is — and no rule was asked where it goes, and
/// no link was touched, because it was not copied.
Orphaned {
/// The item that was left alone.
destination: GlobalId,
},
}
impl CopyAction {
/// The qualified id this action is about, when there is one.
#[must_use]
pub fn destination(&self) -> Option<&GlobalId> {
match self {
Self::Created { destination, .. } => destination.as_ref(),
Self::Updated { destination, .. }
| Self::Unchanged { destination, .. }
| Self::Orphaned { destination } => Some(destination),
}
}
/// Which rule found the counterpart, or `None` when none did — the item was created —
/// and for an orphan, which no rule was asked about.
#[must_use]
pub fn found_by(&self) -> Option<CopyVia> {
match self {
Self::Updated { via, .. } | Self::Unchanged { via, .. } => Some(*via),
Self::Created { .. } | Self::Orphaned { .. } => None,
}
}
/// What became of the copied item's link, or `None` for a dry run and for an orphan.
#[must_use]
pub fn link(&self) -> Option<CopyLink> {
match self {
Self::Created { link, .. }
| Self::Updated { link, .. }
| Self::Unchanged { link, .. } => *link,
Self::Orphaned { .. } => None,
}
}
/// The link slot of an item this copy landed, `None` for an orphan.
fn link_slot(&mut self) -> Option<&mut Option<CopyLink>> {
match self {
Self::Created { link, .. }
| Self::Updated { link, .. }
| Self::Unchanged { link, .. } => Some(link),
Self::Orphaned { .. } => None,
}
}
/// The word this action serializes as, taken from its own `Serialize`.
///
/// Read back off the wire form rather than written out again in a `match`, for the
/// reason `render::wire` gives: a second spelling of `unchanged` would be a second
/// place for it to drift from the one a caller reads.
#[must_use]
pub fn name(&self) -> String {
serde_json::to_value(self).expect("a contract enum serialises")["action"]
.as_str()
.expect("an internally tagged enum carries its tag")
.to_owned()
}
}
/// What one item points at, resolved as far as the copy has got: its forward edges and the
/// tasks it delivers.
struct Pointing<'a> {
/// Its forward edges, `None` where the far end is a member not landed yet.
edges: &'a [Option<DependencyEdge>],
/// The tasks it delivers that can already be named at the destination.
delivers: &'a [TaskRef],
}
/// Where one item is going at the destination.
enum Target {
/// Update the destination item with this id, reached by the rule named.
Update {
/// The destination item this copy updates.
id: NativeId,
/// Which rule found it.
found: Found,
},
/// Create one.
Create,
}
/// Which of the rules above found the destination item a copy is updating.
///
/// Every one is the same instruction — update that item — and rule 1 is a different answer
/// about the origin, which is why the distinction is carried this far rather than dropped
/// where it is made. See [`recorded`]. It is also what the report's [`CopyVia`] says and
/// what decides whether the copied item's link is written.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Found {
/// The item's own link named it, and it names the item back.
Link,
/// Rule 1: the item being copied already named it, so this copy is a copy-back.
Origin,
/// Rule 2: the destination was searched for an item recording this one as its origin.
Scan,
/// The caller's matching escape.
Match,
}
impl Found {
/// The word the report says it in.
fn via(self) -> CopyVia {
match self {
Self::Link => CopyVia::Link,
Self::Origin => CopyVia::Origin,
Self::Scan => CopyVia::Scan,
Self::Match => CopyVia::Match,
}
}
}
impl Target {
/// Which rule found this target, or `None` for one this copy creates.
fn found(&self) -> Option<Found> {
match self {
Self::Update { found, .. } => Some(*found),
Self::Create => None,
}
}
}
/// What a scan of the destination is looking for.
enum Wanted {
/// An item recording this qualified id as its origin.
Origin(String),
/// An item whose title is this.
Title(String),
/// An item holding this value at this metadata key.
Metadata(String, Value),
}
impl Wanted {
/// Whether one destination item is the one being looked for.
fn found(&self, title: &str, metadata: &BTreeMap<String, Value>) -> bool {
match self {
Self::Origin(id) => {
metadata.get(GlobalId::ORIGIN_KEY) == Some(&Value::String(id.clone()))
}
Self::Title(wanted) => title == wanted,
Self::Metadata(key, value) => metadata.get(key) == Some(value),
}
}
/// The task query that asks `capabilities`' source for this item rather than for all of
/// them, when that source applies the predicate itself.
///
/// Only ever a narrowing the source declared it applies, so a source that cannot is
/// asked for everything exactly as before; and never the last word, because
/// [`found`](Self::found) still confirms each row. A text search is wider than an equal
/// title, and a source may answer a match it cannot yet see as absent — which is the
/// same answer the whole-store walk gave for an item its own listing was behind on.
///
/// A title or a value with no letter or digit is never pushed down: a source whose search
/// indexes words cannot answer it and may refuse it, and the walk it replaces answered it
/// already, so it is asked for everything as it always was.
fn task_query(&self, capabilities: &Capabilities) -> TaskQuery {
match self {
Self::Origin(id) if capabilities.filter_by_origin.is_native() => TaskQuery {
origin: Some(id.clone()),
..TaskQuery::default()
},
Self::Title(title) if capabilities.search_title.is_native() && has_words(title) => {
TaskQuery {
text: Some(TextQuery {
terms: title.clone(),
fields: TextFields::Title,
}),
..TaskQuery::default()
}
}
Self::Metadata(key, Value::String(value))
if capabilities.filter_by_metadata.is_native() && has_words(value) =>
{
MetadataMatch::new(key.clone(), Vec::new(), value.clone())
.map(|wanted| TaskQuery {
metadata: vec![wanted],
..TaskQuery::default()
})
.unwrap_or_default()
}
_ => TaskQuery::default(),
}
}
}
/// Whether `terms` holds a letter or a digit — a word a search that indexes words can find.
fn has_words(terms: &str) -> bool {
terms.chars().any(char::is_alphanumeric)
}
/// What the destination held before this copy touched one item.
///
/// Read once, in [`Engine::land`], and used three times over: to decide whether the write
/// would change anything, to repair the item's edges once the rest of the copy has landed,
/// and — if the copy cannot finish — to put the item back exactly as it was.
#[derive(Clone)]
struct Prior {
/// The item as the destination held it.
item: Item,
/// Its forward edges there.
edges: Vec<DependencyEdge>,
/// The image assets it held there, with the bytes of each its own record does not
/// already say the destination serves — `None` when it held none and no write of this
/// copy carried any, so putting it back writes it exactly as it did before assets.
assets: Option<Vec<AssetPayload>>,
/// What its `onetaskgraph.assets` says the destination serves, checked where it was read.
recorded: Option<AssetUploads>,
}
/// One item that landed with an edge whose far end was not written yet.
///
/// Held until every item of the whole copy has landed, because the far end may be in
/// another project of the same command: a copy of two projects at once is one copied set,
/// not two, and an edge across them is remapped rather than written as a foreign id.
struct Deferred {
/// The item, as it was read and resolved.
item: Planned,
/// The destination project it was filed under.
filed: Option<NativeId>,
/// Where it landed.
destination: NativeId,
/// What the destination held there before, when it held anything.
prior: Option<Prior>,
}
/// What one item's undo has to do to put the destination back.
///
/// Each entry names the source it was written to, because a routed copy writes into more
/// than one, and it is undone in every one of them.
enum Undo {
/// The copy created it, so undoing means removing it.
Created {
/// The source it was created in.
at: SourceName,
/// Which write interface removes it.
kind: Level,
/// The destination id it was created under.
id: NativeId,
},
/// The copy overwrote something, so undoing means writing that something back.
///
/// No `kind` beside the id, unlike the variant above: what was there says which of the
/// two write interfaces takes it back, and a second spelling of that could disagree
/// with it.
Updated {
/// The source it was overwritten in.
at: SourceName,
/// The destination id that was overwritten.
id: NativeId,
/// What was there before.
prior: Prior,
},
}
impl Undo {
/// The source this entry was written to.
fn at(&self) -> &SourceName {
match self {
Self::Created { at, .. } | Self::Updated { at, .. } => at,
}
}
/// The destination id this entry is about.
fn id(&self) -> &NativeId {
match self {
Self::Created { id, .. } | Self::Updated { id, .. } => id,
}
}
/// Which of the destination's three write interfaces this entry belongs to.
///
/// An id alone does not identify a destination item: nothing stops a destination
/// numbering its tasks and its projects in one namespace, and a local-Markdown store
/// filing `alpha.md` under both is the ordinary case rather than the contrived one.
/// So this pairs with `id` wherever one entry has to be told from another.
fn kind(&self) -> Level {
match self {
Self::Created { kind, .. } => *kind,
// Read off what was there, for the reason the variant carries no `kind` of
// its own: two spellings of one fact can disagree, and this one cannot.
Self::Updated { prior, .. } => prior.item.level(),
}
}
}
/// Refuse a write of `item` whose status `destination` has no name for, before the write: a
/// document carries no status, so it asks nothing.
async fn check_status(
destination: &ResolvedSource,
item: &Item,
target: Option<&NativeId>,
) -> Result<(), EngineError> {
let (kind, category) = match item {
Item::Task(task) => (ItemKind::Task, task.status.category),
Item::Project(project) => (ItemKind::Project, project.status.category),
Item::Document(_) => return Ok(()),
};
destination
.source()
.check_status_write(kind, category, target)
.await
.map_err(|error| refused(destination, error))
}
/// Everything one copy has written, in the order it wrote it, so a copy that cannot finish
/// can undo its own writes.
///
/// This is not state the engine keeps: it lives for the length of one `copy` call and is
/// dropped with it, so the invariant that nothing of a user's work is written down outside
/// the plugin that owns it is untouched.
#[derive(Default)]
struct Journal {
/// One entry per destination item this copy first touched, in that order.
entries: Vec<Undo>,
/// One entry per copied item whose `onetaskgraph.copies` link this copy wrote, in that
/// order.
///
/// Kept apart from `entries` because these are the *sources'* items rather than the
/// destination's, and an id alone does not say which store it is in: a copy inside one
/// source can write a link on an item and land on another item with the same kind.
links: Vec<Unlink>,
}
/// How to take back one link this copy wrote on a copied item.
struct Unlink {
/// The item, at its own source.
item: GlobalId,
/// Which of that source's interfaces holds it.
level: Level,
/// What it held at `onetaskgraph.copies` before, or `None` when it held nothing there.
before: Option<Value>,
}
impl Journal {
/// Record what has to happen to put one destination item back.
///
/// The *first* entry for an id is the one that matters and later ones are dropped: an
/// item written twice — once as it lands, once when its edges are repaired — was only
/// ever one thing before this copy started, and that is what undoing it restores.
fn record(&mut self, entry: Undo) {
if self.entries.iter().any(|held| {
held.at() == entry.at() && held.kind() == entry.kind() && held.id() == entry.id()
}) {
return;
}
self.entries.push(entry);
}
}
/// What one copy invocation carries from the item it lands to the next.
///
/// Like the [`Journal`] beside it, this lives for the length of one `copy` call and is
/// dropped with it: nothing here is state the engine keeps, and nothing in it is read back
/// to answer a later query.
#[derive(Default)]
struct Running {
/// Every item an edge of this command resolves to a destination id rather than writing as
/// it was read: the ids named, what travels with them, and — for a member copy — each
/// member the copy does not carry whose own recorded origin already names its
/// destination item. That last kind is resolvable without being copied.
resolvable: Vec<GlobalId>,
/// The destination item each of those corresponds to, as soon as it is known — whether
/// this command found it, landed it, or read it off a member's recorded origin.
///
/// Keyed by the qualified id's own rendering, which is what a recorded origin holds
/// anyway — making `GlobalId` orderable for a local map would put an ordering on a
/// contract type for a reason no caller of it has.
///
/// Qualified, because a routed copy lands items in more than one source, and whether an
/// edge to one is written bare or qualified depends on where both of its ends landed.
counterparts: BTreeMap<String, GlobalId>,
/// The items held back for [`Engine::repair`], across the whole request.
deferred: Vec<Deferred>,
/// The reference figures, one total for the whole invocation rather than one per
/// document.
references: Counted,
/// The destination project each source project corresponds to, once this command has
/// looked for it — so a second task filed under the same project does not walk the
/// destination for it again.
///
/// Keyed by the source project and the source the item lands in, because a routed task
/// lands in its home's member project rather than in the home.
filings: BTreeMap<(String, SourceName), Option<NativeId>>,
/// Every home project this copy has learned the member projects of, by the home's
/// qualified id: which members it holds, and whether this copy added one.
homes: BTreeMap<String, Home>,
/// `delivers` entries naming a member of the copied set, over the whole invocation.
delivers_rewritten: u64,
/// Every task this copy landed, for the relation rule once the whole copy is complete.
landed: Vec<LandedTask>,
/// Every item this copy landed, in the order it landed them, for the link each records
/// once the whole copy has landed.
linking: Vec<Linking>,
}
/// The member project a routed write was filed under, and what finding it wrote.
pub(super) struct Filed {
/// The member project, or `None` when the home is not there to have one.
pub(super) project: Option<NativeId>,
/// The member created and the home's list rewritten, if either was.
journal: Journal,
}
/// One home project, and the member projects it names at `onetaskgraph.members`.
struct Home {
/// The home, where it landed.
id: GlobalId,
/// Its title and status, which a member created for it takes.
title: String,
/// See `title`.
status: onetaskgraph_plugin_api::Status,
/// The members it names, at most one per source.
members: Vec<GlobalId>,
/// Whether this copy added or replaced one, so the home's own list is written once the
/// copy has landed.
grew: bool,
}
impl Home {
/// A home as its source holds it, with the members it already names.
///
/// # Errors
///
/// The home's own source refusing, as malformed, a member list [`members_of`] cannot
/// read: a copy filing into a plan it cannot read whole would add a second member beside
/// one it failed to see.
fn of(id: GlobalId, held: &Project) -> Result<Self, EngineError> {
let members =
members_of(&id, &held.metadata).map_err(|message| EngineError::SourceRefused {
name: id.source.to_string(),
error: SourceError::Malformed { message },
})?;
Ok(Self {
id,
title: held.title.clone(),
status: held.status.clone(),
members,
grew: false,
})
}
}
/// The member projects the home `home` records at [`MetadataKey::MEMBERS_KEY`]; none when it
/// records the key not at all.
///
/// # Errors
///
/// Why, when the key holds anything but a list of qualified ids, each in a different source
/// from the home and from the others — the one shape the routed write that keeps it writes.
pub(crate) fn members_of(
home: &GlobalId,
metadata: &BTreeMap<String, Value>,
) -> Result<Vec<GlobalId>, String> {
let Some(value) = metadata.get(MetadataKey::MEMBERS_KEY) else {
return Ok(Vec::new());
};
let refuse = |why: String| {
format!(
"{home} records {key} as {value}, which {why}; a home's member list is a list of \
qualified project ids, at most one per source and none in the home's own",
key = MetadataKey::MEMBERS_KEY
)
};
let Value::Array(entries) = value else {
return Err(refuse("is not a list".to_owned()));
};
let mut members: Vec<GlobalId> = Vec::new();
for entry in entries {
let member = entry
.as_str()
.and_then(|entry| entry.parse::<GlobalId>().ok())
.ok_or_else(|| refuse(format!("holds {entry}, which is not a qualified id")))?;
if member.source == home.source || members.iter().any(|kept| kept.source == member.source) {
return Err(refuse(format!(
"names {member}, a second project in {}",
member.source
)));
}
members.push(member);
}
Ok(members)
}
/// One item a copy landed, and what settling its `onetaskgraph.copies` link needs.
struct Linking {
/// The item, at its own source.
item: GlobalId,
/// Which of that source's interfaces holds it.
level: Level,
/// Which rule found where it landed, or `None` when this copy created it.
found: Option<Found>,
/// Where it landed.
destination: GlobalId,
/// What it held at `onetaskgraph.copies` when it was read, if anything.
held: Option<Value>,
}
/// One task a copy landed, and what the relation rule reads of it once the copy is complete.
struct LandedTask {
/// Where it landed.
destination: GlobalId,
/// The source it was read from, which a bare `delivers` entry names a task of.
origin: SourceName,
/// Its `delivers`, as its source reported them.
delivers: Vec<TaskRef>,
/// The `delivers` the destination held there before this copy, qualified.
before: Vec<GlobalId>,
/// Its status category, as the copy wrote it.
category: StatusCategory,
}
/// A task a copy landed, with the tasks it delivers now and delivered before, qualified.
struct Deliverer {
destination: GlobalId,
category: StatusCategory,
now: Vec<GlobalId>,
before: Vec<GlobalId>,
}
/// One item, read and resolved, on its way into the destination.
struct Planned {
/// Where it came from.
source: GlobalId,
/// The item as its source reported it.
item: Item,
/// Its forward edges, as its source reported them.
edges: Vec<DependencyEdge>,
/// Where it is going.
target: Target,
/// What the destination held at that target, read once where the target was found.
///
/// The read that decided an item exists is the read of what it holds, so the two are
/// one round trip rather than two against a hosted destination.
held: Option<Prior>,
/// The source it lands in: the copy's destination, or the source a route sent it to.
to: SourceName,
/// What the report says placed it there, for a copy into a source with routes.
placed: Option<Placement>,
/// The image assets its content references, with their bytes as its source holds them,
/// in the order the content first references them; none for a project.
assets: Vec<AssetPayload>,
}
/// One item read out of its source, before anything decides where it lands.
struct Read {
/// Where it came from.
source: GlobalId,
/// The item as its source reported it.
item: Item,
}
/// A task, a project or a document, so the copy path is written once.
#[derive(Clone)]
enum Item {
/// A task.
Task(Box<Task>),
/// A project.
Project(Box<Project>),
/// A document.
Document(Box<Document>),
}
impl Item {
fn id(&self) -> &NativeId {
match self {
Self::Task(task) => &task.id,
Self::Project(project) => &project.id,
Self::Document(document) => &document.id,
}
}
fn level(&self) -> Level {
match self {
Self::Task(_) => Level::Task,
Self::Project(_) => Level::Project,
Self::Document(_) => Level::Document,
}
}
}
/// Which of a destination's three read-and-write interfaces one item belongs to.
///
/// Deliberately not [`ItemKind`]: that enum names what a *dependency endpoint* points at,
/// and the contract gives it no document variant because nothing may point at a document.
/// This one names which pair of methods reads and writes an item, which is a different
/// question with a third answer.
///
/// Ordered so it can key a map of what a destination holds, per interface: an id alone
/// does not identify a destination item, for the reason [`Undo::kind`] records.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub(super) enum Level {
/// `get_task`, `write_task`, `delete_task`.
Task,
/// `get_project`, `write_project`, `delete_project`.
Project,
/// `get_document`, `write_document`, `delete_document`.
Document,
}
/// One record filed under a document's own project, and the location string its source
/// reports for it.
///
/// A reference is a **literal occurrence, in a document's content, of the exact location
/// string the source reports for a related record** — the `String` inside
/// [`Location::Path`] or [`Location::Url`]. Both ends of a rewrite come from the plugins'
/// own reported [`Location`]: nothing here composes an address out of a name, an id or a
/// root, because a source that reports a canonical absolute path and a source that reports
/// an issue link are the two things the contract lets this ask about.
#[derive(Clone)]
struct Referent {
/// Its qualified id at the source.
id: GlobalId,
/// The origin it records in its own metadata, when it records one.
origin: Option<GlobalId>,
/// Which of the destination's three interfaces its counterpart would be read from.
level: Level,
/// The non-empty location string its source reports for it.
location: String,
}
impl Referent {
/// The two keys a destination record's own recorded origin is matched against.
///
/// **A destination record is this referent's counterpart when its
/// [`GlobalId::ORIGIN_KEY`] equals either this referent's own qualified source id, or
/// the origin this referent itself records.** In plainer terms: when the destination
/// record was copied directly from this referent, or directly from the one predecessor
/// this referent itself records.
///
/// That reach is exactly one recorded hop of ancestry on each side, and nothing more:
///
/// - **A single hop resolves.** The destination record came straight from the referent.
/// - **A one-level fan-out resolves.** A record copied from one store into two, with
/// the document arriving by one route and naming records that arrived by the other,
/// so both sides trace to one common predecessor. That is what the second key buys,
/// and it is the only thing it buys.
/// - **A chain of two or more hops does not resolve**, on either side, and that is
/// permanent rather than pending: only one origin is ever recorded and every hop
/// overwrites it. [`carried`] removes the key from the incoming metadata outright and
/// [`recorded`] writes the id at the *immediate* source on every path but the
/// copy-back, so A → B → C leaves C keyed by B and A's id gone.
///
/// The second key costs no read — the referent's metadata is already in hand. Chasing
/// the chain further would need the intermediate stores configured and reachable, which
/// would put a third party's availability inside a copy; a durable lineage id would
/// identify only records written after it landed, so it would resolve nothing already
/// on a destination.
fn keys(&self) -> Vec<String> {
let mut keys = vec![self.id.to_string()];
if let Some(origin) = &self.origin {
let recorded = origin.to_string();
if recorded != keys[0] {
keys.push(recorded);
}
}
keys
}
}
/// What every whole-reference occurrence of one location string becomes.
///
/// Three variants rather than a rewrite beside a flag, because the two ways of leaving an
/// occurrence alone are what the two figures a copy reports are *about*: one is the design
/// working and the other says the destination or the source holds something a re-run will
/// never fix. A bool would let a reader of this type read them as the same outcome.
enum Resolution {
/// The destination's own location string for the counterpart.
Rewrite(String),
/// The destination holds no counterpart, or holds one it reports no location for, so
/// the occurrence is left byte-for-byte. Ordinary and expected under the bound this
/// design works to.
NoCounterpart,
/// The correspondence could not be established **confidently** — more than one
/// destination record matches the referent's two keys, or two referents report this one
/// location string. The occurrence is left byte-for-byte, no record is chosen, and a
/// re-run will never clear it.
Ambiguous,
}
/// Every destination record that records an origin, read **once per copy invocation**.
///
/// Several documents in one project is the ordinary case, and a walk per document would
/// multiply reads against a rate limiter for no gain — so this is built once, for the
/// levels the invocation's own documents really name, and every document and every
/// referent of that invocation is answered out of it. A copy whose documents hold no
/// candidate reference builds none at all.
///
/// **This is a stricter discipline than [`Engine::scan`], deliberately.** That lookup takes
/// the first hit and stops, and every consumer of the copy already depends on it doing so;
/// it chooses the copy's own target, where a caller named the item. This one edits the
/// content of somebody's document, where a wrong answer is silent corruption of prose a
/// person will act on — so where more than one record matches, it chooses none. The two
/// lookups answer different questions and are meant to disagree on a destination holding
/// duplicates.
#[derive(Default)]
struct Counterparts {
/// The records at one interface recording one origin.
by_origin: BTreeMap<(Level, String), Vec<Held>>,
}
/// One record a destination walk found: where it is there, and the location string that
/// destination reports for it — `None` when it reports none.
type Held = (NativeId, Option<String>);
impl Counterparts {
/// Record one destination item, when it records an origin at all.
fn note(
&mut self,
level: Level,
id: &NativeId,
location: Option<&Location>,
metadata: &BTreeMap<String, Value>,
) {
let Some(origin) = origin_of(metadata) else {
return;
};
self.by_origin
.entry((level, origin.to_string()))
.or_default()
.push((id.clone(), located(location)));
}
/// What one referent's occurrences become, by the two-key rule.
///
/// Where the correspondence cannot be established **confidently**, the text is left
/// exactly as it is and no record is chosen — see the note on this type.
fn resolve(&self, referent: &Referent) -> Resolution {
let mut candidates: Vec<&Held> = Vec::new();
for key in referent.keys() {
for record in self
.by_origin
.get(&(referent.level, key))
.into_iter()
.flatten()
{
// One destination record matching both keys is one record, not two.
if !candidates.iter().any(|held| held.0 == record.0) {
candidates.push(record);
}
}
}
match candidates.as_slice() {
[] => Resolution::NoCounterpart,
// A counterpart the destination reports no location for names nowhere a reader
// could go, so the source's own string is left standing rather than removed.
[(_, location)] => location
.clone()
.map_or(Resolution::NoCounterpart, Resolution::Rewrite),
_ => Resolution::Ambiguous,
}
}
}
impl Engine {
/// Copy every item a request names into one configured destination.
///
/// This is the whole of the verb, and the command line drives exactly this: a copy a
/// Rust caller makes and a copy typed at a shell are the same call, so the two cannot
/// answer the same copy differently.
///
/// # Errors
///
/// Returns [`EngineError`] when the destination is not configured, cannot be built,
/// cannot be written, or — for a document copy — declares it has no documents; when an
/// id names nothing; when an origin names an item the
/// destination no longer holds and `--recreate` was not given; and when the
/// destination refuses the write — including a field or a metadata key it cannot
/// carry, which it names rather than dropping.
pub async fn copy(&self, request: &CopyRequest) -> Result<CopyReport, EngineError> {
// Before anything is read: `--create` says there is nothing to look for, and the other
// two say how to look.
if request.create {
for (given, flag) in [
(request.match_by.is_some(), CopyLookup::MatchBy),
(request.recreate, CopyLookup::Recreate),
] {
if given {
return Err(EngineError::CreateWith { flag });
}
}
}
let destination = self.writable(&request.destination)?;
// Before anything is read, and from the declaration rather than from a failed
// write: a destination that says it has no documents has nowhere to put one.
if request.scope == CopyScope::Documents {
documentary(destination)?;
}
// The sources this command names, each read once before and once after, so what it
// spent is the difference between two of each one's own running totals.
let mut metered = vec![destination];
for routed in self.routes.reachable(&request.destination) {
if let Some(source) = self.ready().find(|source| source.name() == &routed)
&& !metered.iter().any(|held| held.name() == source.name())
{
metered.push(source);
}
}
for id in request.items.as_slice() {
if let Some(source) = self.ready().find(|source| source.name() == &id.source)
&& !metered.iter().any(|held| held.name() == source.name())
{
metered.push(source);
}
}
let before = readings(&metered).await;
let mut journal = Journal::default();
match self.copy_all(request, &mut journal).await {
Ok((mut report, deliverers)) => {
// After the whole copy is complete and outside the journal: what the relation
// rule writes is the delivered tasks' own, and a failure there is reported
// against that task rather than undoing a copy that landed.
for deliverer in &deliverers {
report.delivered.extend(
self.deliver(
&deliverer.destination,
deliverer.category,
&deliverer.now,
&deliverer.before,
)
.await,
);
}
report.spent = spent_between(&before, &readings(&metered).await);
Ok(report)
}
Err(error) => Err(self.undo(journal, error).await),
}
}
/// The copy itself, with everything it writes recorded so a failure can be undone.
///
/// The ids named together are **one** copied set, and that is what makes an edge
/// between any two of them a real edge at the destination: a copy of two projects at
/// once knows that a task in the first depends on a task in the second, and a task
/// knows that the project it belongs to is being created beside it. Copying them one
/// at a time could not, and wrote the far end as the id it had at its *source* — a
/// dangling reference to somewhere the destination has never heard of.
async fn copy_all(
&self,
request: &CopyRequest,
journal: &mut Journal,
) -> Result<(CopyReport, Vec<Deliverer>), EngineError> {
let mut running = Running::default();
// The whole copied set, established before anything is written. For a project
// copy that means reading every named project's membership first: the set is the
// whole request rather than one project of it.
let items = match &request.scope {
CopyScope::Tasks | CopyScope::Documents => {
let kind = if request.scope == CopyScope::Documents {
Level::Document
} else {
Level::Task
};
running
.resolvable
.extend(request.items.as_slice().iter().cloned());
let mut planned = Vec::new();
for id in request.items.as_slice() {
planned.push(self.plan(request, kind, id, &mut running).await?);
}
self.copy_items(request, planned, None, &mut running, journal)
.await?
}
CopyScope::Projects { tasks } => {
let mut projects = Vec::new();
for id in request.items.as_slice() {
let members = if *tasks {
self.project_members(id).await?
} else {
Vec::new()
};
running.resolvable.push(id.clone());
running.resolvable.extend(members.iter().cloned());
projects.push((id.clone(), members));
}
self.copy_projects(request, &projects, &[], &mut running, journal)
.await?
}
CopyScope::Members(named) => {
let (projects, unrecorded) =
self.named_members(request, named, &mut running).await?;
self.copy_projects(request, &projects, &unrecorded, &mut running, journal)
.await?
}
};
let references = running.references;
let delivers_rewritten = running.delivers_rewritten;
// Every member's destination id is known once the first pass has landed it, so the
// tasks each landed task delivers are settled here rather than after the repair.
let deliverers: Vec<Deliverer> = running
.landed
.iter()
.map(|task| Deliverer {
destination: task.destination.clone(),
category: task.category,
now: targets(
&resolved_entries(&mapped_delivers(
&task.delivers,
&task.origin,
&task.destination.source,
&running.resolvable,
&running.counterparts,
)),
&task.destination.source,
),
before: task.before.clone(),
})
.collect();
let linking = std::mem::take(&mut running.linking);
let homes = std::mem::take(&mut running.homes);
self.repair(request, running, journal).await?;
if !request.dry_run {
self.record_members(homes, journal).await?;
}
let mut items = items;
self.link(linking, &mut items, journal).await?;
Ok((
CopyReport {
items,
references_rewritten: references.rewritten,
references_unresolved: references.unresolved,
references_ambiguous: references.ambiguous,
delivers_rewritten,
// Filled in by `copy`, once the copy is complete.
delivered: Vec::new(),
// Filled in by `copy`, which is the one place both readings are taken.
spent: None,
},
deliverers,
))
}
/// Write every deferred item again, now that every destination id is known.
///
/// This is the second half of the two passes an edge between two items of one copy
/// needs: the far end's destination id does not exist until it has been created, so
/// the item that points at it lands first without that edge and is completed here.
/// It runs once for the whole request rather than once per project, because a far end
/// may be in a project this copy has not reached yet.
async fn repair(
&self,
request: &CopyRequest,
running: Running,
journal: &mut Journal,
) -> Result<(), EngineError> {
if request.dry_run {
return Ok(());
}
let Running {
resolvable,
counterparts,
deferred,
..
} = running;
for entry in deferred {
let destination = self.writable(&entry.item.to)?;
let edges = mapped_edges(
&entry.item.edges,
&entry.item.source.source,
&entry.item.to,
&resolvable,
&counterparts,
);
let delivers = delivers_of(&entry.item, &entry.item.to, &resolvable, &counterparts);
self.write(
destination,
&entry.item,
Some(entry.destination),
entry.filed,
&resolved(&edges),
&resolved_entries(&delivers),
entry.prior,
journal,
)
.await?;
}
Ok(())
}
/// Put the destination back the way this copy found it, then report why it failed.
///
/// Undone in reverse, and an item this copy created is removed rather than restored —
/// the entry recording what it looked like a moment after creation is not a state
/// anybody asked for. When the destination cannot take one of them back, the refusal
/// says so and names what is still there, because a user told "the copy failed" about
/// a destination that is not as they left it will copy again over a tree nobody
/// described.
async fn undo(&self, journal: Journal, error: EngineError) -> EngineError {
let created: Vec<(&SourceName, Level, &NativeId)> = journal
.entries
.iter()
.filter_map(|entry| match entry {
Undo::Created { at, kind, id } => Some((at, *kind, id)),
Undo::Updated { .. } => None,
})
.collect();
// The ids and the refusal are one value rather than two, because they are one
// fact: an item is only left behind because the destination refused to take it
// back, so the first refusal carries the first id and neither half can be
// recorded without the other.
let mut unrestored: Option<(LeftBehind, SourceError)> = None;
// The links first, because they were written last: once the whole copy had landed.
for entry in journal.links.iter().rev() {
if let Err(problem) = self.unlink(entry).await {
match &mut unrestored {
Some((left_behind, _)) => left_behind.push(entry.item.clone()),
None => unrestored = Some((LeftBehind::new(entry.item.clone()), problem)),
}
}
}
for entry in journal.entries.iter().rev() {
let outcome = match self.writable(entry.at()) {
Err(failed) => Err(SourceError::Refused {
message: failed.to_string(),
}),
Ok(destination) => match entry {
Undo::Created { kind, id, .. } => remove(destination, *kind, id).await,
Undo::Updated { at, id, prior }
if !created.contains(&(at, prior.item.level(), id)) =>
{
restore(destination, id, prior).await
}
Undo::Updated { .. } => Ok(()),
},
};
if let Err(problem) = outcome {
let id = GlobalId::new(entry.at().clone(), entry.id().clone());
match &mut unrestored {
Some((left_behind, _)) => left_behind.push(id),
None => unrestored = Some((LeftBehind::new(id), problem)),
}
}
}
match unrestored {
None => error,
Some((left_behind, refusal)) => EngineError::CopyNotUndone {
error: Box::new(error),
left_behind,
refusal,
},
}
}
/// Record, on every item this copy landed, where it landed — once the whole copy has
/// landed, so a copy that fails before this point has written nothing on any source.
///
/// Each outcome is told what happened to its item's link. The items landed are exactly
/// the outcomes that are not orphans, in the order they landed; a dry run records
/// nothing, so this is never asked about one.
async fn link(
&self,
linking: Vec<Linking>,
items: &mut [CopyOutcome],
journal: &mut Journal,
) -> Result<(), EngineError> {
let mut landed = items
.iter_mut()
.filter(|outcome| !matches!(outcome.action, CopyAction::Orphaned { .. }));
for entry in linking {
let slot = landed
.next()
.filter(|outcome| outcome.source == entry.item)
.and_then(|outcome| outcome.action.link_slot())
.expect("every item a copy landed is reported, in the order it landed");
*slot = Some(self.linked(entry, journal).await?);
}
Ok(())
}
/// Write one landed item's `onetaskgraph.copies` entry for the destination, when it
/// does not already say this, and say what was done.
///
/// An item found by its own origin records nothing: that correspondence is already
/// recorded, on the destination item. A source that cannot hold the entry — it has no
/// write side, or refuses the narrow metadata write — leaves the copy as it landed and
/// is reported so; any other failure fails the copy, which then undoes every write it
/// made, this one's journal entry included.
async fn linked(&self, entry: Linking, journal: &mut Journal) -> Result<CopyLink, EngineError> {
// Keyed by the source the item really landed in, which a routed copy decides per item.
let landed_in = entry.destination.source.clone();
if entry.found == Some(Found::Origin) {
return Ok(CopyLink::Unchanged);
}
let wanted = Value::String(entry.destination.to_string());
// Only the entries that are links are kept. Anything else under the key — a value
// that is not an object, or an entry a person spelled wrong — is not a link this or
// any later copy can read, so it is replaced by what is, exactly as an entry naming
// another item is rewritten.
let mut links = well_formed(entry.held.as_ref());
if links.get(landed_in.as_str()) == Some(&wanted) {
return Ok(CopyLink::Unchanged);
}
let source = self.readable(&entry.item.source)?;
if !source.source().writes().is_supported() {
return Ok(CopyLink::Unrecorded);
}
links.insert(landed_in.to_string(), wanted);
// Journalled before the write, for the reason a destination write is: a write that
// stopped part way is only put back by what this journal holds.
journal.links.push(Unlink {
item: entry.item.clone(),
level: entry.level,
before: entry.held,
});
match set_link(
source,
entry.level,
&entry.item.native,
&Value::Object(links),
)
.await
{
Ok(true) => Ok(CopyLink::Recorded),
// Nothing was written: the source no longer holds the item, or said it cannot
// hold the key.
Ok(false) | Err(SourceError::Refused { .. }) => {
journal.links.pop();
Ok(CopyLink::Unrecorded)
}
// §4.18 once let a plugin refuse any key in this namespace as malformed, and one
// written then still may. Its answer says nothing about whether it wrote, so the
// journal entry stays and a later failure puts the item back regardless.
Err(SourceError::Malformed { .. }) => Ok(CopyLink::Unrecorded),
Err(error) => Err(refused(source, error)),
}
}
/// Put one copied item's `onetaskgraph.copies` back to what it held before this copy.
///
/// A key it held is written back through the same narrow write. A key it did not hold
/// cannot be removed through that write, so the item is read as it now is and written
/// back whole, without the key, through the source's own write interface — the same
/// way an overwritten destination item is restored.
async fn unlink(&self, entry: &Unlink) -> Result<(), SourceError> {
let as_source_error = |error: EngineError| match error {
EngineError::SourceRefused { error, .. } => error,
other => SourceError::Refused {
message: other.to_string(),
},
};
let source = self.readable(&entry.item.source).map_err(as_source_error)?;
if let Some(before) = &entry.before {
return set_link(source, entry.level, &entry.item.native, before)
.await
.map(|_| ());
}
let Some(mut held) = self
.prior(source, entry.level, &entry.item.native)
.await
.map_err(as_source_error)?
else {
return Ok(());
};
metadata_of(&mut held.item).remove(MetadataKey::COPIES_KEY);
restore(source, &entry.item.native, &held).await
}
/// The member project `home` names in `to`, created and recorded on the home when it
/// names none — for a write that is not a copy, such as `task create`, routed away from
/// the source its project is in. `None` when the home itself is not there.
///
/// What it wrote is handed back, so a write that then fails takes it back through
/// [`Engine::unfile`] and leaves no member nobody asked for.
pub(super) async fn member_project(
&self,
home: &GlobalId,
to: &SourceName,
) -> Result<Filed, EngineError> {
let mut running = Running::default();
let mut journal = Journal::default();
let mut filed = self
.member_in(false, home, to, &mut running, &mut journal)
.await;
if filed.is_ok() {
let homes = std::mem::take(&mut running.homes);
if let Err(error) = self.record_members(homes, &mut journal).await {
filed = Err(error);
}
}
match filed {
Ok(project) => Ok(Filed { project, journal }),
Err(error) => Err(self.undo(journal, error).await),
}
}
/// Take back what [`Engine::member_project`] wrote, because the write it filed for failed
/// with `error`.
pub(super) async fn unfile(&self, filed: Filed, error: EngineError) -> EngineError {
self.undo(filed.journal, error).await
}
/// The destination source, once it is established it exists and can be written.
fn writable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
let name = self.known(name)?;
if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
return Err(EngineError::DestinationUnavailable {
name: name.to_string(),
error: unavailable.error().clone(),
});
}
let source = self
.ready()
.find(|source| source.name() == &name)
.ok_or(EngineError::NoSources)?;
if !source.source().writes().is_supported() {
return Err(EngineError::NotWritable {
name: name.to_string(),
kind: source.kind().to_owned(),
});
}
Ok(source)
}
/// Copy each project and what travels with it: every task in it, the members a member
/// copy names, or nothing at all.
///
/// Every item of every project is read and its target decided **before any of them is
/// written, and each exactly once.** That is what lets a member copy refuse an edge it
/// cannot resolve while the destination is still as it was found, and what stops a
/// whole copy resolving one target twice — once to learn the ids the project's own
/// edges name, and again to land it — which against a hosted destination is a read per
/// item spent on an answer the command already had.
async fn copy_projects(
&self,
request: &CopyRequest,
projects: &[(GlobalId, Vec<GlobalId>)],
unrecorded: &[GlobalId],
running: &mut Running,
journal: &mut Journal,
) -> Result<Vec<CopyOutcome>, EngineError> {
let mut plans = Vec::new();
for (id, members) in projects {
if self.routes.routes(&request.destination) {
// Every task is read before the project is aimed, because where its home goes
// is decided by where they do.
let project = self.read(Level::Project, id).await?;
let mut reads = Vec::new();
for member in members {
reads.push(self.read(Level::Task, member).await?);
}
let home = self.home_placement(request, &project, &reads).await?;
let project = self.aim(request, project, home).await?;
let mut tasks = Vec::new();
for read in reads {
let placement = match &read.item {
Item::Task(task) => {
self.routes.place(&request.destination, &task.repositories)
}
Item::Project(_) | Item::Document(_) => {
self.placed_at(request, &request.destination)
}
};
tasks.push(self.aim(request, read, placement).await?);
}
plans.push((project, tasks));
} else {
let project = self.plan(request, Level::Project, id, running).await?;
let mut tasks = Vec::new();
for member in members {
tasks.push(self.plan(request, Level::Task, member, running).await?);
}
plans.push((project, tasks));
}
}
for (project, tasks) in &plans {
for item in std::iter::once(project).chain(tasks) {
unrecorded_far_end(item, unrecorded)?;
}
}
let mut outcomes = Vec::new();
for (project, tasks) in plans {
outcomes.extend(
self.copy_project(request, project, tasks, running, journal)
.await?,
);
}
Ok(outcomes)
}
/// Where a routed project copy lands the project: its **home**.
///
/// A home that already exists, in any source the copy can reach, stays where it is —
/// found by the project's own link or origin. Otherwise the home follows its tasks only
/// when every one copied routes to one source, and so does the project's own
/// repositories when it names any; any other project's home is the copy's destination.
/// With no task copied, the project's own repositories alone decide.
async fn home_placement(
&self,
request: &CopyRequest,
project: &Read,
tasks: &[Read],
) -> Result<Placement, EngineError> {
let Item::Project(held) = &project.item else {
return Ok(self.placed_at(request, &request.destination));
};
let mut placements: Vec<Placement> = tasks
.iter()
.filter_map(|read| match &read.item {
Item::Task(task) => {
Some(self.routes.place(&request.destination, &task.repositories))
}
Item::Project(_) | Item::Document(_) => None,
})
.collect();
if !held.repositories.is_empty() {
placements.push(self.routes.place(&request.destination, &held.repositories));
}
let rule = match placements.first() {
Some(first)
if placements
.iter()
.all(|placement| placement.destination == first.destination) =>
{
first.clone()
}
_ => self.placed_at(request, &request.destination),
};
for elsewhere in self.routes.reachable(&request.destination) {
if elsewhere == rule.destination {
continue;
}
let there = self.writable(&elsewhere)?;
let named = link_of(&held.metadata, &elsewhere)
.or_else(|| origin_of(&held.metadata).filter(|origin| origin.source == elsewhere));
let Some(named) = named else {
continue;
};
let Some(found) = self.prior(there, Level::Project, &named.native).await? else {
continue;
};
let names_back = origin_of(described(&found.item).1).as_ref() == Some(&project.source);
if names_back || origin_of(&held.metadata).as_ref() == Some(&named) {
return Ok(self.placed_at(request, &elsewhere));
}
}
Ok(rule)
}
/// Land one planned project, then the tasks planned beside it.
async fn copy_project(
&self,
request: &CopyRequest,
project: Planned,
tasks: Vec<Planned>,
running: &mut Running,
journal: &mut Journal,
) -> Result<Vec<CopyOutcome>, EngineError> {
let (carries_tasks, walks_orphans) = match request.scope {
CopyScope::Projects { tasks } => (tasks, tasks),
CopyScope::Members(_) => (true, false),
CopyScope::Tasks | CopyScope::Documents => (false, false),
};
let id = project.source.clone();
let members: Vec<GlobalId> = tasks.iter().map(|task| task.source.clone()).collect();
// The project lands with every edge it can already resolve — its members' targets
// are known from their plans — so a project that has not changed is not written at
// all. What it *reports* is read the way it always has been: against the ids known
// before its own members' were, unless every edge resolves and nothing differs.
// Reading it any other way would move the word a repeat copy reports for a project
// whose only difference is an edge, which is not this change's to move.
let mut before_members = running.counterparts.clone();
if let Target::Update { id: target, .. } = &project.target {
before_members.insert(
id.to_string(),
GlobalId::new(project.to.clone(), target.clone()),
);
}
for item in std::iter::once(&project).chain(&tasks) {
if let Target::Update { id: target, .. } = &item.target {
running.counterparts.insert(
item.source.to_string(),
GlobalId::new(item.to.clone(), target.clone()),
);
}
}
let home_source = project.to.clone();
let unchanged = match &project.target {
Target::Update { id: target, .. } => {
let settled = mapped_edges(
&project.edges,
&id.source,
&home_source,
&running.resolvable,
&running.counterparts,
);
let first = mapped_edges(
&project.edges,
&id.source,
&home_source,
&running.resolvable,
&before_members,
);
let held = project.held.as_ref();
let unchanged_with = |edges: &[Option<DependencyEdge>]| {
!changes(
held,
&project,
target,
&None,
&resolved(edges),
&[],
&home_source,
)
};
Some(
(!settled.iter().any(Option::is_none) && unchanged_with(&settled))
|| unchanged_with(&first),
)
}
Target::Create => None,
};
// A project this copy creates holds nothing it did not file itself, so nothing filed
// under it can be an orphan and the destination is not walked for one.
let created = matches!(project.target, Target::Create);
let mut outcomes = self
.copy_items(request, vec![project], None, running, journal)
.await?;
if let (Some(unchanged), Some(landed), Some(via)) = (
unchanged,
outcomes[0].destination().cloned(),
outcomes[0].action.found_by(),
) {
let link = outcomes[0].action.link();
outcomes[0].action = if unchanged {
CopyAction::Unchanged {
destination: landed,
via,
link,
}
} else {
CopyAction::Updated {
destination: landed,
via,
link,
}
};
}
if !carries_tasks {
return Ok(outcomes);
}
// `None` when a dry run would have created the project: nothing was written, so
// there is no destination project id to file the tasks under. Every task is still
// read and still reported, because that is what a dry run is for.
let filed = outcomes.first().and_then(CopyOutcome::destination).cloned();
let filing = match &filed {
None => None,
Some(home) => {
let mut filing = Filing::from([(home.source.clone(), home.native.clone())]);
// A task routed to another source than its home's is filed under the home's
// member project there, found or created before any of those tasks lands.
let mut elsewhere: Vec<SourceName> = Vec::new();
for task in &tasks {
if task.to != home.source && !elsewhere.contains(&task.to) {
elsewhere.push(task.to.clone());
}
}
for to in elsewhere {
if let Some(member) = self
.member_in(request.dry_run, home, &to, running, journal)
.await?
{
filing.insert(to, member);
}
}
Some(filing)
}
};
outcomes.extend(
self.copy_items(request, tasks, filing.as_ref(), running, journal)
.await?,
);
// A member copy was told which members it carries, so it cannot tell a member it
// left out from one the source no longer holds, and does not walk for either.
if walks_orphans
&& !created
&& let Some(filed) = filed
{
let destination = self.writable(&filed.source)?;
outcomes.extend(
self.orphans(destination, &id, &filed.native, &members)
.await?,
);
// A routed plan's tasks are filed under its members too, so each member is walked
// in its own source for what the source plan no longer holds.
if self.routes.routes(&request.destination) {
for member in self.members_named(&filed, running).await? {
let there = self.writable(&member.source)?;
outcomes.extend(self.orphans(there, &id, &member.native, &members).await?);
}
}
}
Ok(outcomes)
}
/// The member projects `home` names: the list this copy already holds for it, grown
/// members included, or else the one the home itself records.
async fn members_named(
&self,
home: &GlobalId,
running: &Running,
) -> Result<Vec<GlobalId>, EngineError> {
if let Some(held) = running.homes.get(&home.to_string()) {
return Ok(held.members.clone());
}
let at = self.writable(&home.source)?;
let Some(held) = at
.source()
.get_project(&home.native)
.await
.map_err(|error| refused(at, error))?
else {
return Ok(Vec::new());
};
members_of(home, &held.metadata).map_err(|message| EngineError::SourceRefused {
name: home.source.to_string(),
error: SourceError::Malformed { message },
})
}
/// The members a member copy names, per project and in the order they were named, and
/// every member it does not name that records no destination id.
///
/// Settled before anything is read at the destination. A named id that is a member of
/// none of the projects is refused here, naming it. An unnamed member whose recorded
/// origin names the destination joins the copied set with that id already known, so an
/// edge to it resolves exactly as an edge to a copied item does, and costs no read. An
/// unnamed member recording none is returned, so an edge to it is refused before
/// anything is written — see [`unrecorded_far_end`].
async fn named_members(
&self,
request: &CopyRequest,
named: &CopyItems,
running: &mut Running,
) -> Result<(Vec<(GlobalId, Vec<GlobalId>)>, Vec<GlobalId>), EngineError> {
let projects = request.items.as_slice();
// A routed copy can land a member in any source it reaches, so an origin naming any
// of them is a destination id already known.
let reachable = self.routes.reachable(&request.destination);
let mut carried = Vec::new();
let mut unrecorded = Vec::new();
for project in projects {
let held = self.project_member_tasks(project).await?;
let mut members: Vec<GlobalId> = Vec::new();
for id in named.as_slice() {
if held.iter().any(|task| &task.id == id) && !members.contains(id) {
members.push(id.clone());
}
}
for task in held {
if members.contains(&task.id) {
continue;
}
// A routed copy also reads the member's own link, because a plan authored
// where it is copied from records where each task landed rather than an
// origin, and its tasks may have landed in any source the copy reaches.
let linked = || {
reachable
.iter()
.find_map(|there| link_of(&task.item.metadata, there))
.filter(|_| self.routes.routes(&request.destination))
};
match origin_of(&task.item.metadata)
.filter(|origin| reachable.contains(&origin.source))
.or_else(linked)
{
Some(origin) => {
running.counterparts.insert(task.id.to_string(), origin);
running.resolvable.push(task.id);
}
None => unrecorded.push(task.id),
}
}
running.resolvable.push(project.clone());
running.resolvable.extend(members.iter().cloned());
carried.push((project.clone(), members));
}
if let Some(stray) = named
.as_slice()
.iter()
.find(|id| !carried.iter().any(|(_, members)| members.contains(id)))
{
return Err(EngineError::NotAMember {
id: stray.clone(),
projects: projects.to_vec(),
});
}
Ok((carried, unrecorded))
}
/// Every task the source holds in `project`, by qualified id.
async fn project_members(&self, project: &GlobalId) -> Result<Vec<GlobalId>, EngineError> {
Ok(self
.project_member_tasks(project)
.await?
.into_iter()
.map(|task| task.id)
.collect())
}
/// Every task the source holds in `project`, as the source reported it.
///
/// The ids alone are what a copy files under a project; the whole task is what a
/// document's references need, because the location a reference names is a field of it.
async fn project_member_tasks(
&self,
project: &GlobalId,
) -> Result<Vec<Qualified<Task>>, EngineError> {
let mut request = TaskRequest {
sources: vec![project.source.clone()],
filters: Filters::default(),
project: ProjectSelector::Qualified(project.clone()),
priorities: Vec::new(),
commented_since: None,
metadata: Vec::new(),
origin: None,
include_members: false,
paging: Paging {
limit: PROJECT_PAGE,
token: None,
},
};
let mut members = Vec::new();
// Pages by this engine's own token rather than by a source cursor, and the
// asymmetry with the three walks below is deliberate. A source answering two
// cursors with each other advances on every page, so `unrepeated` under the list
// verb never fires; the cycle shows only as a token handed back unchanged, which
// is what this loop pages by. Point it at the source and a project copy spins.
//
// No page bound here for the same reason: `Engine::tasks` merges under the budget
// it was asked, so nothing longer than `PROJECT_PAGE` can arrive. `fits` in
// `fetch::walk` refuses the page a source can really overrun, its own, while
// these very members are read.
let misbehaved = |error| EngineError::SourceRefused {
name: project.source.to_string(),
error,
};
loop {
let asked = request.paging.token.clone();
let response = self.tasks(&request).await?;
if let Some(failure) = response.errors.first() {
return Err(EngineError::SourceRefused {
name: failure.source.to_string(),
error: failure.error.clone(),
});
}
unrepeated(
response.next.as_ref(),
asked.as_ref(),
"the tasks of a project were being read for a copy",
)
.map_err(misbehaved)?;
members.extend(response.items);
match response.next {
Some(token) => request.paging.token = Some(token),
None => return Ok(members),
}
}
}
/// Every document the source holds in `project`, as the source reported it.
///
/// Paged by this engine's own token for the reason the member walk above is, and the
/// note there says why.
async fn project_documents(
&self,
project: &GlobalId,
) -> Result<Vec<Qualified<Document>>, EngineError> {
let mut request = DocumentRequest {
sources: vec![project.source.clone()],
filters: DocumentFilters::default(),
project: ProjectSelector::Qualified(project.clone()),
paging: Paging {
limit: PROJECT_PAGE,
token: None,
},
};
let mut held = Vec::new();
let misbehaved = |error| EngineError::SourceRefused {
name: project.source.to_string(),
error,
};
loop {
let asked = request.paging.token.clone();
let response = self.documents(&request).await?;
if let Some(failure) = response.errors.first() {
return Err(EngineError::SourceRefused {
name: failure.source.to_string(),
error: failure.error.clone(),
});
}
unrepeated(
response.next.as_ref(),
asked.as_ref(),
"the documents of a project were being read for a copy",
)
.map_err(misbehaved)?;
held.extend(response.items);
match response.next {
Some(token) => request.paging.token = Some(token),
None => return Ok(held),
}
}
}
/// Point every reference the documents of this copy hold at the destination's own
/// records, and say how many it could not.
///
/// A document copied out of a local Markdown store used to arrive naming absolute paths
/// under one checkout on one machine, dead for the only reader the copy exists for,
/// while the destination held its own record for every one of them the whole time. This
/// is what closes that, and it is deliberately **not** a Markdown-link parser: the
/// artifact that motivated it holds bare absolute paths inside backticks in a table
/// cell, which `[text](target)` matching would have left exactly as it found them.
///
/// The correspondence is re-established from what the *destination* records at
/// [`GlobalId::ORIGIN_KEY`], not from the mapping this copy holds. That mapping is not
/// available when it is needed: `project copy` and `document copy` are separate verbs,
/// one invocation carries one [`CopyScope`], and a project copy carries no documents —
/// so by the time the document is copied, the tasks were written by a process that has
/// exited. Reading the destination is also what makes a document copied on its own
/// work, which a same-run mapping never could.
///
// 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-e2e/tests/e2e/rendered.rs`, and the entry's fields are `TemplateProvenance`'s own.
/// Tasks and projects are not touched. A document's content is rewritten, and so is
/// one thing beside it: when this substitutes at least one reference into a rendering
/// that still hashes to the `body_digest` its [`TemplateProvenance`] records, the copy
/// records as `body_digest` the digest of the rendering as this rewrote its references,
/// carrying the entry's other three fields; any other entry, or none, is carried
/// verbatim. The answers that rendering was made from, and the `answers_digest` naming
/// them, still hold the pre-copy locations, so the copy is not a fresh rendering of them
/// and a regenerate from them reproduces the pre-copy content.
// llmlint: ignore-end[contracts_have_one_source_or_a_drift_gate]
async fn rewrite_references(
&self,
planned: &mut [Planned],
counts: &mut Counted,
) -> Result<(), EngineError> {
// Read at the source, once per project rather than once per document: several
// documents of one project is the ordinary case.
let mut by_project: BTreeMap<String, Vec<Referent>> = BTreeMap::new();
let mut named: Vec<Vec<Referent>> = Vec::new();
for item in planned.iter() {
named.push(self.named_referents(item, &mut by_project).await?);
}
// The destination is walked only for a copy that really names something, and only
// for the interfaces those referents are read from — in each source a document of
// this call lands in, which a routed copy decides per document.
let mut levels: BTreeMap<SourceName, Vec<Level>> = BTreeMap::new();
for (item, referents) in planned.iter().zip(&named) {
for referent in referents {
let wanted = levels.entry(item.to.clone()).or_default();
if !wanted.contains(&referent.level) {
wanted.push(referent.level);
}
}
}
if levels.is_empty() {
return Ok(());
}
let mut walked: BTreeMap<SourceName, Counterparts> = BTreeMap::new();
for (to, levels) in levels {
let destination = self.writable(&to)?;
walked.insert(to, self.counterparts(destination, &levels).await?);
}
for (item, referents) in planned.iter_mut().zip(named) {
let Some(counterparts) = walked.get(&item.to) else {
continue;
};
let Item::Document(document) = &mut item.item else {
continue;
};
let Some(content) = &document.content else {
continue;
};
let (rewritten, made) = substitute(content, &table_for(&referents, counterparts));
if made.rewritten > 0
&& let Some(provenance) = restamped(&document.metadata, content, &rewritten)
{
document
.metadata
.insert(TemplateProvenance::KEY.to_owned(), provenance.to_value());
}
document.content = Some(rewritten);
counts.add(made);
}
Ok(())
}
/// The records of one document's own project whose location string its content really
/// holds, as a whole reference.
///
/// A document with no content, or with no project at the source, names nothing: the
/// referent set is the document's own project, its tasks and its other documents, and
/// there is no such set without a project.
async fn named_referents(
&self,
item: &Planned,
by_project: &mut BTreeMap<String, Vec<Referent>>,
) -> Result<Vec<Referent>, EngineError> {
let Item::Document(document) = &item.item else {
return Ok(Vec::new());
};
let (Some(content), Some(project)) =
(document.content.as_deref(), document.project.as_ref())
else {
return Ok(Vec::new());
};
if content.is_empty() {
return Ok(Vec::new());
}
let project = GlobalId::new(item.source.source.clone(), project.clone());
let key = project.to_string();
if !by_project.contains_key(&key) {
let read = self.referents(&project).await?;
by_project.insert(key.clone(), read);
}
Ok(by_project[&key]
.iter()
// A document does not name itself: the referent set is every *other* record
// filed under the project. Told by the interface as well as the id, because an
// id alone does not identify a record — a folder of Markdown filing `A.md`
// under both `tasks/` and `documents/` is the ordinary case rather than the
// contrived one, and excluding by id alone would drop that task from the set.
.filter(|referent| referent.level != Level::Document || referent.id != item.source)
.filter(|referent| holds(content, &referent.location))
.cloned()
.collect())
}
/// Every record filed under one project at its own source, with the location string
/// that source reports for it.
///
/// The project record itself, every task filed under it, and every document filed under
/// it. All three are reads this engine already knows how to make.
async fn referents(&self, project: &GlobalId) -> Result<Vec<Referent>, EngineError> {
let source = self.readable(&project.source)?;
let mut referents = Vec::new();
if let Some(held) = source
.source()
.get_project(&project.native)
.await
.map_err(|error| refused(source, error))?
{
note(
&mut referents,
project.clone(),
Level::Project,
held.location.as_ref(),
&held.metadata,
);
}
for task in self.project_member_tasks(project).await? {
note(
&mut referents,
task.id,
Level::Task,
task.item.location.as_ref(),
&task.item.metadata,
);
}
for document in self.project_documents(project).await? {
note(
&mut referents,
document.id,
Level::Document,
document.item.location.as_ref(),
&document.item.metadata,
);
}
Ok(referents)
}
/// Walk the destination once for every record it holds at the levels named, one page
/// at a time.
///
/// One page is *read* at a time, and what is kept from each is three fields of each
/// record — its id, the origin it records and the location the destination reports —
/// never the page. That is more than [`Engine::scan`] keeps, and deliberately: the whole
/// point of this walk is that one pass answers every document and every referent of the
/// invocation, so what it learns has to outlive the page it learned it from. Nothing is
/// written down and the index is dropped with the call.
///
/// The other difference from `scan` is the *answer*: this one keeps every match, so a
/// destination holding two records for one work item is reported as ambiguous rather
/// than resolved to the first.
async fn counterparts(
&self,
destination: &ResolvedSource,
levels: &[Level],
) -> Result<Counterparts, EngineError> {
let mut found = Counterparts::default();
for level in levels {
// Every cursor this level has already been sent. `unrepeated` below catches a
// source that hands back the cursor it was just given, and its own note says
// why it catches no more than that: a source cycling through two cursors
// advances on every page, and seeing it needs memory the walks that share that
// helper do not keep. This walk does keep memory — it is building an index that
// outlives each page — so here the memory exists and the cycle is caught. The
// walks one level up catch the same defect as a page token handed back
// unchanged; this one pages by the source's own cursor and has no level above
// it, so nothing else would.
let mut asked_before: BTreeSet<String> = BTreeSet::new();
let mut cursor: Option<Cursor> = None;
loop {
if let Some(next) = &cursor
&& !asked_before.insert(next.0.clone())
{
return Err(refused(
destination,
SourceError::Malformed {
message: "the source returned a cursor it had already been \
given while the destination was being walked for the \
records a document's references name, so the walk \
would never end"
.to_owned(),
},
));
}
let asked = cursor.clone();
let request = request_for(destination, cursor);
let next = match level {
Level::Task => {
let page = destination
.source()
.query_tasks(&TaskQuery::default(), &request)
.await
.map_err(|error| refused(destination, error))?;
fits(page.items.len(), request.limit)
.map_err(|error| refused(destination, error))?;
for task in &page.items {
found.note(*level, &task.id, task.location.as_ref(), &task.metadata);
}
page.next
}
Level::Project => {
let page = destination
.source()
.query_projects(&ProjectQuery::default(), &request)
.await
.map_err(|error| refused(destination, error))?;
fits(page.items.len(), request.limit)
.map_err(|error| refused(destination, error))?;
for project in &page.items {
found.note(
*level,
&project.id,
project.location.as_ref(),
&project.metadata,
);
}
page.next
}
Level::Document => {
let page = destination
.source()
.query_documents(&DocumentQuery::default(), &request)
.await
.map_err(|error| refused(destination, error))?;
fits(page.items.len(), request.limit)
.map_err(|error| refused(destination, error))?;
for document in &page.items {
found.note(
*level,
&document.id,
document.location.as_ref(),
&document.metadata,
);
}
page.next
}
};
unrepeated(
next.as_ref(),
asked.as_ref(),
"the destination was being walked for the records a document's \
references name",
)
.map_err(|error| refused(destination, error))?;
match next {
Some(next) => cursor = Some(next),
None => break,
}
}
}
Ok(found)
}
/// Destination tasks filed under the copied project whose origin the source no longer
/// holds.
///
/// A copy never deletes, so each is left exactly as it is and reported.
async fn orphans(
&self,
destination: &ResolvedSource,
project: &GlobalId,
at_destination: &NativeId,
copied: &[GlobalId],
) -> Result<Vec<CopyOutcome>, EngineError> {
let mut orphans = Vec::new();
let mut cursor: Option<Cursor> = None;
loop {
let asked = cursor.clone();
let request = request_for(destination, cursor);
let page: Page<Task> = destination
.source()
.query_tasks(&TaskQuery::default(), &request)
.await
.map_err(|error| refused(destination, error))?;
fits(page.items.len(), request.limit).map_err(|error| refused(destination, error))?;
for task in &page.items {
if task.project.as_ref() != Some(at_destination) {
continue;
}
let Some(origin) = origin_of(&task.metadata) else {
continue;
};
if origin.source != project.source || copied.contains(&origin) {
continue;
}
orphans.push(CopyOutcome {
source: origin,
action: CopyAction::Orphaned {
destination: GlobalId::new(destination.name().clone(), task.id.clone()),
},
placed: None,
});
}
unrepeated(
page.next.as_ref(),
asked.as_ref(),
"the destination was being read for items the copy left behind",
)
.map_err(|error| refused(destination, error))?;
match page.next {
Some(next) => cursor = Some(next),
None => return Ok(orphans),
}
}
}
/// Resolve and write every item planned, holding back the ones whose edges are not
/// resolvable yet.
///
/// An edge between two items of one copy can point at a member whose destination id
/// does not exist until it has been created, so the item that points at it lands
/// without that edge and is handed to `deferred`. [`Engine::repair`] finishes it once
/// the *whole* request has landed — not once this call has, because the far end may
/// be in another project of the same command.
async fn copy_items(
&self,
request: &CopyRequest,
mut planned: Vec<Planned>,
project: Option<&Filing>,
running: &mut Running,
journal: &mut Journal,
) -> Result<Vec<CopyOutcome>, EngineError> {
// Only a document's own content names other records, and only once every document
// of this call has been read: the destination is walked once for all of them, and
// the content the rest of this call lands is the rewritten one — which is what
// makes a repeat copy of an already-rewritten document report `unchanged`.
if planned
.iter()
.any(|item| item.item.level() == Level::Document)
{
self.rewrite_references(&mut planned, &mut running.references)
.await?;
}
for item in &planned {
if let Target::Update { id, .. } = &item.target {
running.counterparts.insert(
item.source.to_string(),
GlobalId::new(item.to.clone(), id.clone()),
);
}
if let Item::Task(task) = &item.item {
running.delivers_rewritten +=
members_named(&task.delivers, &item.source.source, &running.resolvable);
}
}
// Resolved once per item, and used by both passes: the repair pass writes the
// same item again, and re-deriving this there could file it somewhere else.
let mut filed = Vec::new();
for item in &planned {
filed.push(match (project, &item.item) {
(_, Item::Project(_)) => None,
(Some(filing), _) => filing.get(&item.to).cloned(),
(None, _) => self.filed(request, item, None, running, journal).await?,
});
}
let mut outcomes = Vec::new();
let mut unresolved = Vec::new();
let mut priors = Vec::new();
for (index, item) in planned.iter().enumerate() {
let destination = self.writable(&item.to)?;
let edges = mapped_edges(
&item.edges,
&item.source.source,
&item.to,
&running.resolvable,
&running.counterparts,
);
let delivers = delivers_of(item, &item.to, &running.resolvable, &running.counterparts);
if edges.iter().any(Option::is_none) || delivers.iter().any(Option::is_none) {
unresolved.push(index);
}
let resolvable_delivers = resolved_entries(&delivers);
let (outcome, prior) = self
.land(
destination,
request,
item,
filed[index].clone(),
Pointing {
edges: &edges,
delivers: &resolvable_delivers,
},
journal,
)
.await?;
if let Some(id) = outcome.destination() {
running
.counterparts
.insert(item.source.to_string(), id.clone());
if !request.dry_run {
running.linking.push(Linking {
item: item.source.clone(),
level: item.item.level(),
found: item.target.found(),
destination: id.clone(),
held: described(&item.item)
.1
.get(MetadataKey::COPIES_KEY)
.cloned(),
});
}
}
if !request.dry_run
&& let (Item::Task(task), Some(landed)) = (&item.item, outcome.destination())
{
running.landed.push(LandedTask {
destination: landed.clone(),
origin: item.source.source.clone(),
delivers: task.delivers.clone(),
before: match prior.as_ref().map(|prior| &prior.item) {
Some(Item::Task(held)) => targets(&held.delivers, destination.name()),
_ => Vec::new(),
},
category: task.status.category,
});
}
outcomes.push(outcome);
priors.push(prior);
}
if !request.dry_run {
for (index, item) in planned.into_iter().enumerate() {
if !unresolved.contains(&index) {
continue;
}
// Every item a copy that is not a dry run lands has a destination id: the
// one outcome without one is a dry run that would have created, and this
// block does not run for a dry run.
let id = outcomes[index]
.destination()
.expect("a copy that writes lands every item it planned")
.clone();
running.deferred.push(Deferred {
item,
filed: filed[index].clone(),
destination: id.native,
prior: priors[index].clone(),
});
}
}
Ok(outcomes)
}
/// Read one item, place it, read its forward edges, and decide where it is going.
///
/// A task is placed by its own repositories; a document goes with its project's home.
async fn plan(
&self,
request: &CopyRequest,
kind: Level,
id: &GlobalId,
running: &mut Running,
) -> Result<Planned, EngineError> {
let read = self.read(kind, id).await?;
let placement = match &read.item {
Item::Task(task) => self.routes.place(&request.destination, &task.repositories),
Item::Document(document) => {
self.document_placement(request, &read.source, document, running)
.await?
}
Item::Project(project) => self
.routes
.place(&request.destination, &project.repositories),
};
self.aim(request, read, placement).await
}
/// Read one item out of its own source.
async fn read(&self, kind: Level, id: &GlobalId) -> Result<Read, EngineError> {
let source = self.readable(&id.source)?;
if kind == Level::Document {
documentary(source)?;
}
let item = match kind {
Level::Task => source
.source()
.get_task(&id.native)
.await
.map_err(|error| refused(source, error))?
.map(|task| Item::Task(Box::new(task))),
Level::Project => source
.source()
.get_project(&id.native)
.await
.map_err(|error| refused(source, error))?
.map(|project| Item::Project(Box::new(project))),
Level::Document => source
.source()
.get_document(&id.native)
.await
.map_err(|error| refused(source, error))?
.map(|document| Item::Document(Box::new(document))),
}
.ok_or_else(|| EngineError::NoSuchItem { id: id.to_string() })?;
Ok(Read {
source: id.clone(),
item,
})
}
/// Decide where one read item is going, once it is known which source it lands in.
async fn aim(
&self,
request: &CopyRequest,
read: Read,
placement: Placement,
) -> Result<Planned, EngineError> {
let Read { source: id, item } = read;
let destination = self.writable(&placement.destination)?;
if item.level() == Level::Document {
documentary(destination)?;
}
// Before the destination is read, and so before it is written: a destination that
// holds no priority has nowhere to put one, and every item of a copy is planned before
// any of them lands. A task carrying `none` passes and writes exactly as it always did.
if let Item::Task(task) = &item {
holds_priority(destination, &id.to_string(), task.priority)?;
}
let routed = self.routes.routes(&request.destination);
// A task or a document whose counterpart already sits somewhere a route could have
// sent it, other than where it routes now, was moved by a repository changed after
// the copy. That is a person's decision rather than a silent move, so it is refused
// before anything is written. A project's home is exempt: it stays where it landed.
if routed && item.level() != Level::Project {
self.misrouted(request, &id, &item, &placement.destination)
.await?;
}
let source = self.readable(&id.source)?;
// Read, and refused, before the destination is read and before anything is written:
// an asset its source does not hold, and a destination that stores none.
let assets = carried_assets(source, &id, &item).await?;
if let Some(first) = assets.first() {
assets::stores(destination, &id.to_string(), &first.name)?;
}
let edges = forward_edges(source, &id.native, item.level()).await?;
let (target, held) = self.target(destination, request, &id, &item).await?;
Ok(Planned {
source: id,
item,
edges,
target,
held,
to: placement.destination.clone(),
placed: routed.then_some(placement),
assets,
})
}
/// Refuse an item whose existing counterpart, by its link or its origin, sits in a source
/// its copy could reach other than `to`, the one it routes to now.
async fn misrouted(
&self,
request: &CopyRequest,
id: &GlobalId,
item: &Item,
to: &SourceName,
) -> Result<(), EngineError> {
let metadata = described(item).1;
for elsewhere in self.routes.reachable(&request.destination) {
if &elsewhere == to {
continue;
}
let there = self.writable(&elsewhere)?;
let linked = match link_of(metadata, &elsewhere) {
Some(link) => self
.prior(there, item.level(), &link.native)
.await?
.filter(|held| origin_of(described(&held.item).1).as_ref() == Some(id))
.map(|_| link),
None => None,
};
let originated = match origin_of(metadata).filter(|origin| origin.source == elsewhere) {
Some(origin) => self
.prior(there, item.level(), &origin.native)
.await?
.map(|_| origin),
None => None,
};
if let Some(counterpart) = linked.or(originated) {
return Err(EngineError::Misrouted {
item: id.clone(),
counterpart,
route: to.clone(),
});
}
}
Ok(())
}
/// Where a document lands: with its project's home, wherever that is; by its own
/// repositories when its project has none yet, or it is in no project.
async fn document_placement(
&self,
request: &CopyRequest,
id: &GlobalId,
document: &Document,
running: &mut Running,
) -> Result<Placement, EngineError> {
if self.routes.routes(&request.destination)
&& let Some(project) = &document.project
{
let qualified = GlobalId::new(id.source.clone(), project.clone());
if let Some(home) = self.find_home(request, &qualified, running).await? {
return Ok(self.placed_at(request, &home.source));
}
}
Ok(self
.routes
.place(&request.destination, &document.repositories))
}
/// What the report says placed an item in `at`: the first route entry sending there, or
/// none for the copy's own destination.
fn placed_at(&self, request: &CopyRequest, at: &SourceName) -> Placement {
Placement {
destination: at.clone(),
route: self
.routes
.of(&request.destination)
.iter()
.position(|route| route.to() == at)
.map(|index| u32::try_from(index).unwrap_or(u32::MAX)),
}
}
/// Which destination item this one corresponds to, by the link the item records, the
/// two origin rules and the caller's escape, and what the destination holds there.
async fn target(
&self,
destination: &ResolvedSource,
request: &CopyRequest,
id: &GlobalId,
item: &Item,
) -> Result<(Target, Option<Prior>), EngineError> {
let (title, metadata) = described(item);
// A caller asserting there is nothing to find is believed, and nothing is looked for —
// unless the item itself says otherwise, by naming a counterpart at the destination.
if request.create {
let carrier = link_of(metadata, destination.name()).or_else(|| {
origin_of(metadata).filter(|origin| &origin.source == destination.name())
});
if let Some(carrier) = carrier {
return Err(EngineError::CreateCarried {
item: id.clone(),
carrier,
});
}
return Ok((Target::Create, None));
}
// The link first, and it is believed only when the item it names still names this
// one back: a person who re-pointed that item's origin has said it is not this
// one's counterpart any more, so the link is ignored and rewritten to whatever the
// rules below find.
if let Some(link) = link_of(metadata, destination.name()) {
match self.prior(destination, item.level(), &link.native).await? {
Some(held) if origin_of(described(&held.item).1).as_ref() == Some(id) => {
return Ok((
Target::Update {
id: link.native,
found: Found::Link,
},
Some(held),
));
}
Some(_) => {}
None if request.recreate => {}
None => {
return Err(EngineError::StaleLink {
item: id.to_string(),
link: link.to_string(),
});
}
}
}
if let Some(origin) = origin_of(metadata)
&& &origin.source == destination.name()
{
// The read that says the origin still names something is the read of what it
// holds, so rule 1 costs one round trip rather than two.
if let Some(held) = self
.prior(destination, item.level(), &origin.native)
.await?
{
return Ok((
Target::Update {
id: origin.native,
found: Found::Origin,
},
Some(held),
));
}
if !request.recreate {
return Err(EngineError::StaleOrigin {
item: id.to_string(),
origin: origin.to_string(),
});
}
}
if let Some(found) = self
.scan(destination, item.level(), &Wanted::Origin(id.to_string()))
.await?
{
let held = self.prior(destination, item.level(), &found).await?;
return Ok((
Target::Update {
id: found,
found: Found::Scan,
},
held,
));
}
let wanted = match &request.match_by {
Some(MatchBy::Title) => Some(Wanted::Title(title.to_owned())),
Some(MatchBy::Metadata(key)) => metadata
.get(key)
.map(|value| Wanted::Metadata(key.clone(), value.clone())),
None => None,
};
if let Some(wanted) = wanted
&& let Some(found) = self.scan(destination, item.level(), &wanted).await?
{
let held = self.prior(destination, item.level(), &found).await?;
return Ok((
Target::Update {
id: found,
found: Found::Match,
},
held,
));
}
Ok((Target::Create, None))
}
/// Walk the destination one page at a time, looking for `wanted`.
///
/// One page is held at a time and nothing is written down, which is the same bound
/// every other compensation in this engine works under.
async fn scan(
&self,
destination: &ResolvedSource,
kind: Level,
wanted: &Wanted,
) -> Result<Option<NativeId>, EngineError> {
let mut cursor: Option<Cursor> = None;
let tasks = wanted.task_query(&destination.source().capabilities());
loop {
let asked = cursor.clone();
let request = request_for(destination, cursor);
let next = match kind {
Level::Task => {
let page = destination
.source()
.query_tasks(&tasks, &request)
.await
.map_err(|error| refused(destination, error))?;
fits(page.items.len(), request.limit)
.map_err(|error| refused(destination, error))?;
for task in &page.items {
if wanted.found(&task.title, &task.metadata) {
return Ok(Some(task.id.clone()));
}
}
page.next
}
Level::Project => {
let page = destination
.source()
.query_projects(&ProjectQuery::default(), &request)
.await
.map_err(|error| refused(destination, error))?;
fits(page.items.len(), request.limit)
.map_err(|error| refused(destination, error))?;
for project in &page.items {
if wanted.found(&project.title, &project.metadata) {
return Ok(Some(project.id.clone()));
}
}
page.next
}
Level::Document => {
let page = destination
.source()
.query_documents(&DocumentQuery::default(), &request)
.await
.map_err(|error| refused(destination, error))?;
fits(page.items.len(), request.limit)
.map_err(|error| refused(destination, error))?;
for document in &page.items {
if wanted.found(&document.title, &document.metadata) {
return Ok(Some(document.id.clone()));
}
}
page.next
}
};
unrepeated(
next.as_ref(),
asked.as_ref(),
"the destination was being scanned for the item to update",
)
.map_err(|error| refused(destination, error))?;
match next {
Some(next) => cursor = Some(next),
None => return Ok(None),
}
}
}
/// Write one planned item, or say what a dry run would have done.
///
/// Answers with what the destination held there beforehand as well, which is what
/// makes an item written twice restorable to what it was rather than to what this
/// copy's first pass left.
async fn land(
&self,
destination: &ResolvedSource,
request: &CopyRequest,
item: &Planned,
project: Option<NativeId>,
pointing: Pointing<'_>,
journal: &mut Journal,
) -> Result<(CopyOutcome, Option<Prior>), EngineError> {
let Pointing { edges, delivers } = pointing;
let (target, found) = match &item.target {
Target::Update { id, found } => (Some(id.clone()), Some(*found)),
Target::Create => (None, None),
};
// Every link is settled once the whole copy has landed, by `Engine::link`, so none
// is known here.
let reached = |destination: Option<GlobalId>, changed: bool| match (found, destination) {
(Some(via), Some(destination)) if changed => CopyAction::Updated {
destination,
via: via.via(),
link: None,
},
(Some(via), Some(destination)) => CopyAction::Unchanged {
destination,
via: via.via(),
link: None,
},
(_, destination) => CopyAction::Created {
destination,
via: NoCounterpart::Created,
link: None,
},
};
// The one read of the destination item, made where its target was found, used to
// decide whether the write changes anything and — if the copy cannot finish — to
// put that item back.
let prior = item.held.clone();
let edges = resolved(edges);
let qualified = |native: NativeId| GlobalId::new(destination.name().clone(), native);
if let Some(id) = &target
&& !changes(
prior.as_ref(),
item,
id,
&project,
&edges,
delivers,
destination.name(),
)
{
return Ok((
CopyOutcome {
source: item.source.clone(),
action: reached(Some(qualified(id.clone())), false),
placed: item.placed.clone(),
},
prior,
));
}
if request.dry_run {
return Ok((
CopyOutcome {
source: item.source.clone(),
// Null only for a create here: nothing was created, so there is no id to
// report.
action: reached(target.map(qualified), true),
placed: item.placed.clone(),
},
prior,
));
}
let written = qualified(
self.write(
destination,
item,
target,
project,
&edges,
delivers,
prior.clone(),
journal,
)
.await?,
);
Ok((
CopyOutcome {
source: item.source.clone(),
action: reached(Some(written), true),
placed: item.placed.clone(),
},
prior,
))
}
/// Which destination project this item is filed under, when it is filed at all.
///
/// A task copied as part of a project copy is filed under that project's counterpart,
/// which the copy has just established. A task copied on its own has to find it.
async fn filed(
&self,
request: &CopyRequest,
item: &Planned,
project: Option<NativeId>,
running: &mut Running,
journal: &mut Journal,
) -> Result<Option<NativeId>, EngineError> {
match (&item.item, project) {
(Item::Task(task), None) => {
self.counterpart(request, item, task.project.as_ref(), running, journal)
.await
}
(Item::Document(document), None) => {
self.counterpart(request, item, document.project.as_ref(), running, journal)
.await
}
(Item::Task(_) | Item::Document(_), filed) => Ok(filed),
(Item::Project(_), _) => Ok(None),
}
}
/// The destination project this task's own project corresponds to, when there is one.
///
/// A task copied on its own keeps its source's project id when the destination holds
/// no counterpart: the field is opaque to this engine, and dropping it would lose
/// what the source said.
///
/// Looked for once per source project and landing source per command. A project this
/// command itself copied is already known, and one an earlier task of this command was
/// filed under was already looked for; walking the destination again for either would
/// spend a scan per task on an answer the command holds. A project whose own link names
/// its counterpart there is found by that link — see [`Engine::linked_project`] — and
/// the destination is walked only when it does not.
///
/// A routed item whose project's home landed in another source than it does is filed
/// under that home's member project in its own source — see [`Engine::member_in`].
async fn counterpart(
&self,
request: &CopyRequest,
item: &Planned,
project: Option<&NativeId>,
running: &mut Running,
journal: &mut Journal,
) -> Result<Option<NativeId>, EngineError> {
let Some(project) = project else {
return Ok(None);
};
let qualified = GlobalId::new(item.source.source.clone(), project.clone());
let key = (qualified.to_string(), item.to.clone());
if let Some(looked) = running.filings.get(&key) {
return Ok(looked.clone());
}
let filed = match self.find_home(request, &qualified, running).await? {
Some(home) if home.source == item.to => Some(home.native),
Some(home) => {
self.member_in(request.dry_run, &home, &item.to, running, journal)
.await?
}
None => Some(project.clone()),
};
running.filings.insert(key, filed.clone());
Ok(filed)
}
/// The counterpart a source project has in some source this copy can reach — its home —
/// when it has one: the one this command landed, the one its own link names, or the one a
/// walk of each reachable source finds recording it as its origin.
async fn find_home(
&self,
request: &CopyRequest,
project: &GlobalId,
running: &Running,
) -> Result<Option<GlobalId>, EngineError> {
if let Some(landed) = running.counterparts.get(&project.to_string()) {
return Ok(Some(landed.clone()));
}
if let Some(linked) = self.linked_project(request, project).await? {
return Ok(Some(linked));
}
for reachable in self.routes.reachable(&request.destination) {
let there = self.writable(&reachable)?;
if let Some(found) = self
.scan(there, Level::Project, &Wanted::Origin(project.to_string()))
.await?
{
return Ok(Some(GlobalId::new(reachable, found)));
}
}
Ok(None)
}
/// The member project `home` names in `to`, creating it when it names none there.
///
/// Found through the home's `onetaskgraph.members`, read once per command. A member it
/// names that `to` no longer holds is replaced rather than refused: the member exists
/// only to file the home's routed tasks, and nothing a person wrote lives on it. A dry
/// run creates nothing, so a member it would create files its tasks under nothing.
async fn member_in(
&self,
dry_run: bool,
home: &GlobalId,
to: &SourceName,
running: &mut Running,
journal: &mut Journal,
) -> Result<Option<NativeId>, EngineError> {
let key = home.to_string();
if !running.homes.contains_key(&key) {
let at = self.writable(&home.source)?;
let Some(held) = at
.source()
.get_project(&home.native)
.await
.map_err(|error| refused(at, error))?
else {
return Ok(None);
};
running
.homes
.insert(key.clone(), Home::of(home.clone(), &held)?);
}
let named = running.homes[&key]
.members
.iter()
.find(|member| &member.source == to)
.cloned();
let there = self.writable(to)?;
if let Some(member) = named
&& there
.source()
.get_project(&member.native)
.await
.map_err(|error| refused(there, error))?
.is_some()
{
return Ok(Some(member.native));
}
if dry_run {
return Ok(None);
}
let held = &running.homes[&key];
let member = Project {
id: home.native.clone(),
title: held.title.clone(),
content: None,
status: held.status.clone(),
labels: Vec::new(),
url: None,
location: None,
created_at: None,
updated_at: None,
metadata: BTreeMap::from([(
MetadataKey::MEMBER_OF_KEY.to_owned(),
Value::String(home.to_string()),
)]),
repositories: Vec::new(),
};
let created = there
.source()
.write_project(&ItemWrite {
target: None,
item: member,
depends_on: Vec::new(),
})
.await
.map_err(|error| refused(there, error))?;
journal.record(Undo::Created {
at: to.clone(),
kind: Level::Project,
id: created.clone(),
});
let entry = running
.homes
.get_mut(&key)
.expect("the home was recorded above");
entry.members.retain(|member| &member.source != to);
entry
.members
.push(GlobalId::new(to.clone(), created.clone()));
entry.grew = true;
Ok(Some(created))
}
/// Write each home's grown `onetaskgraph.members` through its own project write, once the
/// whole copy has landed.
///
/// The home is read as it now stands and written back with that one entry changed, and
/// the read is what the journal restores when the copy cannot finish — unless the copy
/// already holds an earlier entry for it, which is what it was before this copy began.
async fn record_members(
&self,
homes: BTreeMap<String, Home>,
journal: &mut Journal,
) -> Result<(), EngineError> {
for home in homes.into_values().filter(|home| home.grew) {
let at = self.writable(&home.id.source)?;
let Some(prior) = self.prior(at, Level::Project, &home.id.native).await? else {
continue;
};
let Item::Project(mut project) = prior.item.clone() else {
continue;
};
project.metadata.insert(
MetadataKey::MEMBERS_KEY.to_owned(),
Value::Array(
home.members
.iter()
.map(|member| Value::String(member.to_string()))
.collect(),
),
);
let edges = prior.edges.clone();
at.source()
.check_status_write(
ItemKind::Project,
project.status.category,
Some(&home.id.native),
)
.await
.map_err(|error| refused(at, error))?;
journal.record(Undo::Updated {
at: home.id.source.clone(),
id: home.id.native.clone(),
prior,
});
at.source()
.write_project(&ItemWrite {
target: Some(home.id.native.clone()),
item: *project,
depends_on: edges,
})
.await
.map_err(|error| refused(at, error))?;
}
Ok(())
}
/// The home project a source project's own `onetaskgraph.copies` names in a source this
/// copy can reach, when the project it names there still records that source project as
/// its origin.
///
/// The filing lookup's half of following a link: one read of the project at its source
/// and one of its counterpart by id, in place of a walk of the destination's projects.
/// A link that names nothing there, or names a project somebody re-pointed, is not a
/// refusal here as it is for a copy's own target — this only decides where an item is
/// filed — so the walk answers instead, exactly as it did before there were links.
async fn linked_project(
&self,
request: &CopyRequest,
project: &GlobalId,
) -> Result<Option<GlobalId>, EngineError> {
let source = self.readable(&project.source)?;
if !source.source().capabilities().projects.is_native() {
return Ok(None);
}
let Some(held) = source
.source()
.get_project(&project.native)
.await
.map_err(|error| refused(source, error))?
else {
return Ok(None);
};
for reachable in self.routes.reachable(&request.destination) {
let Some(link) = link_of(&held.metadata, &reachable) else {
continue;
};
let there = self.writable(&reachable)?;
let named = there
.source()
.get_project(&link.native)
.await
.map_err(|error| refused(there, error))?;
if named.is_some_and(|there| origin_of(&there.metadata).as_ref() == Some(project)) {
return Ok(Some(link));
}
}
Ok(None)
}
/// What the destination holds at one id, item and forward edges together.
///
/// One read for both purposes it serves — deciding whether a write changes anything,
/// and putting the item back if the copy cannot finish — because a second read of the
/// same item is a second round trip against a hosted destination for nothing.
async fn prior(
&self,
destination: &ResolvedSource,
kind: Level,
id: &NativeId,
) -> Result<Option<Prior>, EngineError> {
let held = match kind {
Level::Task => destination
.source()
.get_task(id)
.await
.map_err(|error| refused(destination, error))?
.map(|task| Item::Task(Box::new(task))),
Level::Project => destination
.source()
.get_project(id)
.await
.map_err(|error| refused(destination, error))?
.map(|project| Item::Project(Box::new(project))),
Level::Document => destination
.source()
.get_document(id)
.await
.map_err(|error| refused(destination, error))?
.map(|document| Item::Document(Box::new(document))),
};
let Some(item) = held else {
return Ok(None);
};
let edges = forward_edges(destination, id, kind).await?;
let recorded =
assets::recorded(described(&item).1).map_err(|error| refused(destination, error))?;
let assets = held_assets(destination, kind, id, recorded.as_ref()).await?;
Ok(Some(Prior {
item,
edges,
assets,
recorded,
}))
}
/// Hand one item to the destination's own write interface, recording how to take it
/// back.
// llmlint: ignore[suppressions_justified] A write is the item, where it is going, what
// it is filed under, its edges, what was there before and the journal that records how
// to put it back. Each is a distinct decision made by a different part of the copy, and
// grouping them would only move the argument list to a constructor.
#[allow(clippy::too_many_arguments)]
async fn write(
&self,
destination: &ResolvedSource,
item: &Planned,
target: Option<NativeId>,
project: Option<NativeId>,
edges: &[DependencyEdge],
delivers: &[TaskRef],
prior: Option<Prior>,
journal: &mut Journal,
) -> Result<NativeId, EngineError> {
let created_kind = item.item.level();
let suggested = target
.clone()
.unwrap_or_else(|| created_id(&item.item, project.as_ref()));
// Settled before the journal takes `prior`, and from that same read: what the
// destination holds at the origin key is what a copy-back leaves there, and what it
// holds as `delivered_by` is what the item keeps.
let origin = recorded(item, prior.as_ref());
let landing = outgoing(item, suggested, project, &origin, delivers, prior.as_ref());
// Asked before the journal records anything or the destination is sent anything: a
// status the destination has no name for refuses this write while there is nothing to
// put back, where refused inside the write it would leave the journal restoring an
// item the write never touched.
check_status(destination, &landing, target.as_ref()).await?;
// Recorded *before* the write rather than after it. A destination's own write is
// several calls — `docs/plugin-protocol.md` §4.9 — and one of them failing leaves
// the ones before it applied. No source can put those back, because only this
// journal holds what was there; recorded after a successful write, an update that
// stopped part way was the one way a copy could end and leave the destination
// altered. A restore of an item the write never reached rewrites what is already
// there, which costs one mutation and is what "either complete or it never
// happened" is worth.
// A write carries assets when the item's content references any, and when the item it
// overwrites holds any: what lands then holds exactly the assets this copy carries.
let held_assets = prior.as_ref().and_then(|prior| prior.assets.clone());
let carrying = !item.assets.is_empty() || held_assets.is_some();
let recorded = prior.as_ref().and_then(|prior| prior.recorded.clone());
if let (Some(id), Some(mut prior)) = (target.clone(), prior) {
if carrying && prior.assets.is_none() {
// Putting this item back then has to take away the assets this write adds.
prior.assets = Some(Vec::new());
}
journal.record(Undo::Updated {
at: destination.name().clone(),
id,
prior,
});
}
let landed = match landing {
Item::Task(task) if carrying => destination
.source()
.write_task_with_assets(
&ItemWrite {
target: target.clone(),
item: *task,
depends_on: edges.to_vec(),
},
None,
&assets::write_of(item.assets.clone(), recorded.clone()),
)
.await
.map(|written| written.id)
.map_err(|error| refused(destination, error))?,
Item::Document(document) if carrying => destination
.source()
.write_document_with_assets(
&ItemWrite {
target: target.clone(),
item: *document,
depends_on: Vec::new(),
},
None,
&assets::write_of(item.assets.clone(), recorded.clone()),
)
.await
.map(|written| written.id)
.map_err(|error| refused(destination, error))?,
Item::Task(task) => destination
.source()
.write_task(&ItemWrite {
target: target.clone(),
item: *task,
depends_on: edges.to_vec(),
})
.await
.map_err(|error| refused(destination, error))?,
Item::Project(project) => destination
.source()
.write_project(&ItemWrite {
target: target.clone(),
item: *project,
depends_on: edges.to_vec(),
})
.await
.map_err(|error| refused(destination, error))?,
// No edges, and that is the contract: a document takes part in no dependency
// graph, so there is nothing here for `depends_on` to carry.
Item::Document(document) => destination
.source()
.write_document(&ItemWrite {
target: target.clone(),
item: *document,
depends_on: Vec::new(),
})
.await
.map_err(|error| refused(destination, error))?,
};
// A created item can only be journalled here: its id is what the write answers
// with. A create that fails leaves nothing behind — §4.9 makes taking the item
// back the source's own duty, because a write that refused must not leave an item
// nobody asked for.
if target.is_none() {
journal.record(Undo::Created {
at: destination.name().clone(),
kind: created_kind,
id: landed.clone(),
});
}
Ok(landed)
}
/// A configured source that built, for reading an item out of.
fn readable(&self, name: &SourceName) -> Result<&ResolvedSource, EngineError> {
let name = self.known(name)?;
if let Some(unavailable) = self.unavailable().find(|source| source.name() == &name) {
return Err(EngineError::SourceRefused {
name: name.to_string(),
error: unavailable.error().clone(),
});
}
self.ready()
.find(|source| source.name() == &name)
.ok_or(EngineError::NoSources)
}
}
/// How many tasks of a project are read at once while walking it.
const PROJECT_PAGE: std::num::NonZeroU32 = std::num::NonZeroU32::new(50).expect("50 is not zero");
/// Each source's own running totals, in the order the sources were given.
///
/// A reading a source could not take is read as that source not metering: what a command
/// spent is a report about the work, and a failed reading must not become a failure of the
/// work itself. That holds when only one of a source's two readings failed as well, since
/// a difference needs both ends.
pub(super) async fn readings(sources: &[&ResolvedSource]) -> Vec<Option<Metering>> {
let mut read = Vec::with_capacity(sources.len());
for source in sources {
read.push(source.source().metering().await.ok().flatten());
}
read
}
/// What the sources spent between two readings of them, or `None` when none of them meters.
///
/// A source that does not meter contributes nothing and does not make the total zero: a
/// command whose sources all declined reports that it cannot say, not that it spent nothing.
/// A budget is keyed by its name and unit together, so two sources naming one budget add up
/// and two units of one budget stay apart.
///
/// A source's two readings are a plugin's word, so they are held to [`Metering`]'s contract
/// before either is believed, and a pair that breaks it is that source not metering — see
/// [`difference`].
pub(super) fn spent_between(
before: &[Option<Metering>],
after: &[Option<Metering>],
) -> Option<Spent> {
let mut metered = false;
let mut requests = 0_u64;
let mut budgets: BTreeMap<(String, String), (u64, u64)> = BTreeMap::new();
for (before, after) in before.iter().zip(after) {
let (Some(before), Some(after)) = (before, after) else {
continue;
};
let Some((sent, spent)) = difference(before, after) else {
continue;
};
metered = true;
requests = requests.saturating_add(sent);
for (key, measured, modelled) in spent {
let total = budgets.entry(key).or_default();
total.0 = total.0.saturating_add(measured).saturating_add(modelled);
total.1 = total.1.saturating_add(modelled);
}
}
metered.then(|| Spent {
requests,
budgets: budgets
.into_iter()
.map(|((budget, unit), (amount, modelled))| BudgetSpent {
budget,
unit,
amount,
lower_bound: modelled > 0,
})
.collect(),
})
}
/// One budget's name and unit, and the measured and modelled amounts spent against it.
type BudgetDifference = ((String, String), u64, u64);
/// What one source sent and spent between two of its readings, or `None` when the pair
/// cannot be a running total's.
///
/// A running total never falls, never drops a budget it has named, and names every budget
/// it keeps, once; a pair that breaks any of that is a source that reset its figures or
/// reported something else, and a difference taken over it would be a number that measures
/// nothing. Such a source is reported as not metering rather than as having spent a clamped
/// zero.
fn difference(before: &Metering, after: &Metering) -> Option<(u64, Vec<BudgetDifference>)> {
let sent = after.requests.checked_sub(before.requests)?;
for reading in [before, after] {
let mut named = BTreeSet::new();
for budget in &reading.budgets {
if budget.budget.is_empty()
|| budget.unit.is_empty()
|| !named.insert((&budget.budget, &budget.unit))
{
return None;
}
}
}
let held = |from: &Metering, budget: &onetaskgraph_plugin_api::Metered| {
from.budgets
.iter()
.find(|held| held.budget == budget.budget && held.unit == budget.unit)
.cloned()
};
if before
.budgets
.iter()
.any(|budget| held(after, budget).is_none())
{
return None;
}
let mut spent = Vec::with_capacity(after.budgets.len());
for budget in &after.budgets {
let earlier = held(before, budget);
let measured = budget
.measured
.checked_sub(earlier.as_ref().map_or(0, |held| held.measured))?;
let modelled = budget
.modelled
.checked_sub(earlier.as_ref().map_or(0, |held| held.modelled))?;
spent.push((
(budget.budget.clone(), budget.unit.clone()),
measured,
modelled,
));
}
Some((sent, spent))
}
/// One page request against `source`, at the largest page it will serve.
fn request_for(source: &ResolvedSource, cursor: Option<Cursor>) -> PageRequest {
PageRequest {
cursor,
limit: source.source().capabilities().max_page_size.max(1),
}
}
/// Whether writing this item would change what the destination already holds.
///
/// A free function over the state already read rather than a method that reads it again:
/// the same answer is wanted where the item is landed and where a repeat copy of a project
/// decides whether it settled, and a second read there is a second round trip for nothing.
fn changes(
held: Option<&Prior>,
item: &Planned,
target: &NativeId,
project: &Option<NativeId>,
edges: &[DependencyEdge],
delivers: &[TaskRef],
destination: &SourceName,
) -> bool {
let Some(held) = held else {
return true;
};
let mut outgoing = outgoing(
item,
target.clone(),
project.clone(),
&recorded(item, Some(held)),
delivers,
Some(held),
);
if (!item.assets.is_empty() || held.assets.is_some())
&& !lands_as_held(item, held, &mut outgoing)
{
return true;
}
!same(&held.item, &outgoing, destination) || !same_edges(&held.edges, edges)
}
/// Whether `held` already holds exactly the assets `item` carries, by name and digest — and,
/// when it does, `outgoing` made what the destination would store of it.
///
/// A destination that serves its assets at a URL says what it holds in its own record,
/// `onetaskgraph.assets`, and would rewrite the references to the URLs recorded there; one
/// that keeps them beside the record says so in the assets it lists, and stores the references
/// as written.
fn lands_as_held(item: &Planned, held: &Prior, outgoing: &mut Item) -> bool {
let Some(recorded) = &held.recorded else {
let holds: Vec<onetaskgraph_plugin_api::Asset> = held
.assets
.iter()
.flatten()
.map(|payload| onetaskgraph_plugin_api::Asset {
name: payload.name.clone(),
sha256: payload.sha256.clone(),
content_type: payload.content_type,
path: None,
})
.collect();
return assets::same_set(&holds, &item.assets);
};
if recorded.0.len() != item.assets.len() {
return false;
}
let mut served = AssetUploads::default();
for payload in &item.assets {
let Some(url) = recorded.reusable(&payload.name, &payload.sha256) else {
return false;
};
served.0.insert(
payload.name.clone(),
onetaskgraph_plugin_api::AssetUpload {
sha256: payload.sha256.clone(),
url: url.to_owned(),
},
);
}
let (content, metadata) = match outgoing {
Item::Task(task) => (&mut task.content, &mut task.metadata),
Item::Document(document) => (&mut document.content, &mut document.metadata),
Item::Project(_) => return true,
};
if let Some(text) = content {
*text = onetaskgraph_plugin_api::serve_asset_references(text, metadata, &served);
}
true
}
/// The image assets `item`'s content references, each with its bytes as `source` holds them,
/// in the order the content first references them — none for a project.
///
/// # Errors
///
/// [`EngineError::AssetNotHeld`] naming the record and the asset for a reference `source` holds
/// no asset for, and [`EngineError::SourceRefused`] when it cannot answer.
async fn carried_assets(
source: &ResolvedSource,
id: &GlobalId,
item: &Item,
) -> Result<Vec<AssetPayload>, EngineError> {
let (content, owner) = match item {
Item::Task(task) => (task.content.as_deref(), assets::Owner::Task),
Item::Document(document) => (document.content.as_deref(), assets::Owner::Document),
Item::Project(_) => return Ok(Vec::new()),
};
let referenced = onetaskgraph_plugin_api::asset_references(content.unwrap_or_default());
if referenced.is_empty() {
return Ok(Vec::new());
}
let listed = match owner {
assets::Owner::Document => source.source().document_assets(&id.native).await,
assets::Owner::Task => source.source().task_assets(&id.native).await,
}
.map_err(|error| refused(source, error))?;
let mut carried = Vec::with_capacity(referenced.len());
for name in referenced {
let not_held = || EngineError::AssetNotHeld {
record: id.to_string(),
asset: name.to_string(),
};
if !listed.iter().any(|asset| asset.name == name) {
return Err(not_held());
}
let bytes = match owner {
assets::Owner::Document => source.source().document_asset(&id.native, &name).await,
assets::Owner::Task => source.source().task_asset(&id.native, &name).await,
}
.map_err(|error| refused(source, error))?
.ok_or_else(not_held)?;
carried.push(AssetPayload::of(name, bytes));
}
Ok(carried)
}
/// The image assets the item `id` of kind `kind` holds at `destination`, as [`Prior::assets`]
/// keeps them: `None` for a destination that stores none, for a project, and for an item that
/// holds none.
async fn held_assets(
destination: &ResolvedSource,
kind: Level,
id: &NativeId,
recorded: Option<&AssetUploads>,
) -> Result<Option<Vec<AssetPayload>>, EngineError> {
let owner = match kind {
Level::Task => assets::Owner::Task,
Level::Document => assets::Owner::Document,
Level::Project => return Ok(None),
};
if !destination.source().capabilities().assets.is_native() {
return Ok(None);
}
let listed = match owner {
assets::Owner::Document => destination.source().document_assets(id).await,
assets::Owner::Task => destination.source().task_assets(id).await,
}
.map_err(|error| refused(destination, error))?;
if listed.is_empty() {
// A destination that lists none may still serve some — one reached over the stdio
// protocol lists nothing — and its own record of what it uploaded is then what it holds,
// every one of them reusable by its digest without reading a byte back.
return Ok(recorded
.filter(|recorded| !recorded.0.is_empty())
.map(|recorded| {
recorded
.0
.iter()
.map(|(name, upload)| AssetPayload {
name: name.clone(),
sha256: upload.sha256.clone(),
content_type: name.content_type(),
bytes: None,
})
.collect()
}));
}
let mut held = Vec::with_capacity(listed.len());
for asset in listed {
// Bytes the destination's own record says it serves are not read back: putting them
// back reuses what it serves.
let bytes = if recorded
.is_some_and(|recorded| recorded.reusable(&asset.name, &asset.sha256).is_some())
{
None
} else {
match owner {
assets::Owner::Document => {
destination.source().document_asset(id, &asset.name).await
}
assets::Owner::Task => destination.source().task_asset(id, &asset.name).await,
}
.map_err(|error| refused(destination, error))?
};
held.push(AssetPayload {
name: asset.name,
sha256: asset.sha256,
content_type: asset.content_type,
bytes,
});
}
Ok(Some(held))
}
/// Remove one item this copy created, through the destination's own write interface.
async fn remove(
destination: &ResolvedSource,
kind: Level,
id: &NativeId,
) -> Result<(), SourceError> {
match kind {
Level::Task => destination.source().delete_task(id).await,
Level::Project => destination.source().delete_project(id).await,
Level::Document => destination.source().delete_document(id).await,
}
}
/// Write one item back exactly as the destination held it before this copy.
async fn restore(
destination: &ResolvedSource,
id: &NativeId,
prior: &Prior,
) -> Result<(), SourceError> {
if let Some(held) = &prior.assets {
let assets = AssetWrite {
assets: held.clone(),
recorded_assets: prior.recorded.clone(),
};
match &prior.item {
Item::Task(task) => {
return destination
.source()
.write_task_with_assets(
&ItemWrite {
target: Some(id.clone()),
item: (**task).clone(),
depends_on: prior.edges.clone(),
},
None,
&assets,
)
.await
.map(|_| ());
}
Item::Document(document) => {
return destination
.source()
.write_document_with_assets(
&ItemWrite {
target: Some(id.clone()),
item: (**document).clone(),
depends_on: Vec::new(),
},
None,
&assets,
)
.await
.map(|_| ());
}
Item::Project(_) => {}
}
}
match &prior.item {
Item::Task(task) => destination
.source()
.write_task(&ItemWrite {
target: Some(id.clone()),
item: (**task).clone(),
depends_on: prior.edges.clone(),
})
.await
.map(|_| ()),
Item::Project(project) => destination
.source()
.write_project(&ItemWrite {
target: Some(id.clone()),
item: (**project).clone(),
depends_on: prior.edges.clone(),
})
.await
.map(|_| ()),
Item::Document(document) => destination
.source()
.write_document(&ItemWrite {
target: Some(id.clone()),
item: (**document).clone(),
depends_on: Vec::new(),
})
.await
.map(|_| ()),
}
}
/// Hold `value` at `onetaskgraph.copies` on one item, through its source's narrow metadata
/// write — `false` when the source holds no such item.
async fn set_link(
source: &ResolvedSource,
level: Level,
id: &NativeId,
value: &Value,
) -> Result<bool, SourceError> {
let key = MetadataKey::copies();
Ok(match level {
Level::Task => source
.source()
.set_task_metadata(id, &key, value)
.await?
.is_some(),
Level::Project => source
.source()
.set_project_metadata(id, &key, value)
.await?
.is_some(),
Level::Document => source
.source()
.set_document_metadata(id, &key, value)
.await?
.is_some(),
})
}
/// The metadata of either kind of item, to edit.
fn metadata_of(item: &mut Item) -> &mut BTreeMap<String, Value> {
match item {
Item::Task(task) => &mut task.metadata,
Item::Project(project) => &mut project.metadata,
Item::Document(document) => &mut document.metadata,
}
}
/// Refuse a document copy addressed to a source that declares it has none.
///
/// Read off the declaration rather than by asking, which is what "not asked" means: the
/// engine learned at the handshake that this source holds no documents, so it refuses
/// naming the source and its plugin instead of sending a read that would be refused there.
/// Applied at both ends of a copy — a source with no documents holds nothing to copy out,
/// and a destination with none has nowhere to put one.
fn documentary(source: &ResolvedSource) -> Result<(), EngineError> {
if source.source().capabilities().documents.is_native() {
return Ok(());
}
Err(EngineError::NoDocuments {
name: source.name().to_string(),
kind: source.kind().to_owned(),
})
}
/// One source failing while a copy was mid-flight.
fn refused(source: &ResolvedSource, error: SourceError) -> EngineError {
EngineError::SourceRefused {
name: source.name().to_string(),
error,
}
}
/// Every forward edge at one item, walked to exhaustion one page at a time.
pub(super) async fn forward_edges(
source: &ResolvedSource,
id: &NativeId,
kind: Level,
) -> Result<Vec<DependencyEdge>, EngineError> {
// A document has no edges to walk, and asking for them would mean asking a source for
// a graph the contract says nothing may point into.
if kind == Level::Document {
return Ok(Vec::new());
}
let mut edges = Vec::new();
let mut cursor: Option<Cursor> = None;
loop {
let asked = cursor.clone();
let request = request_for(source, cursor);
let page = match kind {
Level::Task | Level::Document => {
source
.source()
.task_dependencies(id, Direction::DependsOn, &request)
.await
}
Level::Project => {
source
.source()
.project_dependencies(id, Direction::DependsOn, &request)
.await
}
}
.map_err(|error| refused(source, error))?;
fits(page.items.len(), request.limit).map_err(|error| refused(source, error))?;
edges.extend(page.items);
unrepeated(
page.next.as_ref(),
asked.as_ref(),
"an item's dependencies were being read for a copy",
)
.map_err(|error| refused(source, error))?;
match page.next {
Some(next) => cursor = Some(next),
None => return Ok(edges),
}
}
}
/// The location string one record reports, when it reports a usable one.
///
/// Either variant's own `String`, and `None` for a record the source gave no location for
/// or gave an empty string for: there is nothing to look for in a document's content and
/// nothing to point a reader at.
fn located(location: Option<&Location>) -> Option<String> {
let (Location::Path(held) | Location::Url(held)) = location?;
(!held.is_empty()).then(|| held.clone())
}
/// Record one candidate referent, when its source said where it is.
fn note(
into: &mut Vec<Referent>,
id: GlobalId,
level: Level,
location: Option<&Location>,
metadata: &BTreeMap<String, Value>,
) {
if let Some(location) = located(location) {
into.push(Referent {
id,
origin: origin_of(metadata),
level,
location,
});
}
}
/// What every location string this document names becomes, longest first.
///
/// Longest first because a shorter location may start where a longer one does — a
/// project's directory and a task's file under it — and the longer of the two is the
/// record that occurrence names.
fn table_for(referents: &[Referent], counterparts: &Counterparts) -> Vec<(String, Resolution)> {
let mut table: Vec<(String, Resolution)> = Vec::new();
for referent in referents {
// Two referents reporting one location string: an occurrence of it cannot be
// attributed to either, and a rewrite would be *confidently wrong* rather than
// merely unhelpful. So neither is chosen and both occurrences are counted.
if let Some(held) = table
.iter_mut()
.find(|(location, _)| location == &referent.location)
{
held.1 = Resolution::Ambiguous;
continue;
}
table.push((referent.location.clone(), counterparts.resolve(referent)));
}
table.sort_by_key(|(location, _)| std::cmp::Reverse(location.len()));
table
}
/// Whether `content` holds `location` at least once, stopped on both sides.
fn holds(content: &str, location: &str) -> bool {
(0..content.len()).any(|at| delimited_at(content, at, location))
}
/// Whether `location` occurs at `at` **stopped on both sides** — by
/// [`stops_a_location`], or by the end of the content — rather than as part of a longer
/// location-like string.
///
/// A location string occurring inside a longer one is a different string naming a
/// different record: `/…/tasks/p/t.md` must not be rewritten inside `/…/tasks/p/t.md.bak`,
/// `https://example.invalid/1` must not be rewritten inside `https://example.invalid/12`,
/// and a project's location that is a directory prefix of a task's must not be rewritten
/// inside that task's. What deciding it this way costs is stated on [`stops_a_location`].
///
/// A `#` straight after the location also stops it, and only on that side: it opens a
/// fragment — `[budgets](/…/t.md#budgets)`, `https://…/issues/1#budgets` — which names a
/// place *inside* the record rather than another record. The location before it is
/// rewritten like any other, and the fragment, never part of the match, is carried through
/// [`substitute`]'s walk byte-for-byte. Read this way, a file whose own name holds a `#`
/// beside a referent's — `t.md#2` beside `t.md` — would be taken for that referent with a
/// fragment; a *referent* so named is still matched whole, because the table is tried
/// longest first.
fn delimited_at(content: &str, at: usize, location: &str) -> bool {
if !content.is_char_boundary(at) || !content[at..].starts_with(location) {
return false;
}
let before = content[..at].chars().next_back();
let after = content[at + location.len()..].chars().next();
stops_a_location(before) && (stops_a_location(after) || after == Some(FRAGMENT))
}
/// What opens a fragment after a location: see [`delimited_at`].
const FRAGMENT: char = '#';
/// Whether a character cannot continue a path or a link, so a location string beside one
/// ends there.
///
/// Stated as what *stops* a location rather than as what one may contain, because the
/// second list is unbounded — a path may hold very nearly any byte, and a URL more. Every
/// character not named here continues, which is what leaves the three cases above alone;
/// the end of the content counts as a stop. The set is what the artifact this exists for
/// really wraps a bare path in — a backtick in a table cell — plus the delimiters prose
/// and Markdown put next to one.
///
/// **Sentence punctuation is deliberately absent, and that is a stated cost rather than an
/// oversight.** `.`, `!`, `?` and `:` each equally *continue* a real location — `/…/t.md`
/// and `/…/t.md.bak` are two files, `…/1` and `…/1?q=2` two pages — so admitting them as
/// stops would rewrite one record's location into another's. The price is that a location
/// written bare at the end of a sentence is not recognised at all: its text is left
/// byte-for-byte and it is counted in neither figure, exactly as a reference to another
/// project's record is. That is the direction to be wrong in, because this edits the
/// content of somebody's document, where a confidently wrong rewrite is worse than one
/// that never happens.
fn stops_a_location(character: Option<char>) -> bool {
match character {
None => true,
Some(character) => character.is_whitespace() || "`\"'()[]{}<>|,;".contains(character),
}
}
/// One document's content with every whole reference rewritten, and what that took.
///
/// A location string with no confident counterpart is left **byte-for-byte** as it was
/// rather than removed or guessed at, and so is every character of the content that is not
/// a rewritten reference. Nothing is added and nothing is reformatted.
fn substitute(content: &str, table: &[(String, Resolution)]) -> (String, Counted) {
let mut written = String::with_capacity(content.len());
let mut counts = Counted::default();
let mut at = 0;
while at < content.len() {
if let Some((location, resolution)) = table
.iter()
.find(|(location, _)| delimited_at(content, at, location))
{
match resolution {
Resolution::Rewrite(there) => {
written.push_str(there);
counts.rewritten += 1;
}
// Both left-alone outcomes count as unresolved on the branch that counts
// them, which is what really holds `ambiguous` at or below `unresolved`.
Resolution::NoCounterpart => {
written.push_str(location);
counts.unresolved += 1;
}
Resolution::Ambiguous => {
written.push_str(location);
counts.unresolved += 1;
counts.ambiguous += 1;
}
}
at += location.len();
continue;
}
let character = content[at..]
.chars()
.next()
.expect("a character at a boundary this walk only ever lands on");
written.push(character);
at += character.len_utf8();
}
(written, counts)
}
/// The entry [`Engine::rewrite_references`] records for a document it rewrote from `read`
/// to `written`, or `None` to carry what the document records verbatim.
///
/// `None` unless `read` is still the rendering its entry vouches for, because only then is
/// the one change between it and `written` this copy's own: a hand edit is never laundered.
fn restamped(
metadata: &BTreeMap<String, Value>,
read: &str,
written: &str,
) -> Option<TemplateProvenance> {
let mut provenance = TemplateProvenance::read(metadata).ok().flatten()?;
if provenance.body_digest.as_str() != body_digest(read) {
return None;
}
provenance.body_digest =
Sha256Digest::parse(body_digest(written)).expect("`body_digest` spells a digest");
Some(provenance)
}
/// The origin one item records, when it records a usable one.
fn origin_of(metadata: &BTreeMap<String, Value>) -> Option<GlobalId> {
metadata
.get(GlobalId::ORIGIN_KEY)?
.as_str()?
.parse::<GlobalId>()
.ok()
}
/// The destination item one item's `onetaskgraph.copies` entry names at `destination`, when
/// it holds a usable one.
///
/// An entry that is not a qualified id, or is one naming another source than the key it is
/// filed under, is no link at all: the copy goes on by the rules that follow and rewrites
/// it to what they find.
fn link_of(metadata: &BTreeMap<String, Value>, destination: &SourceName) -> Option<GlobalId> {
let linked = metadata
.get(MetadataKey::COPIES_KEY)?
.as_object()?
.get(destination.as_str())?
.as_str()?
.parse::<GlobalId>()
.ok()?;
(&linked.source == destination).then_some(linked)
}
/// The entries of a `onetaskgraph.copies` value that are links: a source name, and a
/// qualified id of that same source.
fn well_formed(value: Option<&Value>) -> serde_json::Map<String, Value> {
let Some(Value::Object(held)) = value else {
return serde_json::Map::new();
};
held.iter()
.filter(|(destination, linked)| {
linked
.as_str()
.and_then(|linked| linked.parse::<GlobalId>().ok())
.is_some_and(|linked| linked.source.as_str() == destination.as_str())
})
.map(|(destination, linked)| (destination.clone(), linked.clone()))
.collect()
}
/// Why a value is not one a `onetaskgraph.copies` entry may hold, or nothing when it is: a
/// JSON object mapping each destination source name to a qualified id of that source.
///
/// What the stdio boundary holds an incoming value of that key to, so a plugin is never
/// handed a link it could not answer a later copy with.
pub(crate) fn malformed_links(value: &Value) -> Option<String> {
let Value::Object(held) = value else {
return Some(format!(
"{} holds an object of destination source names to qualified ids, not {value}",
MetadataKey::COPIES_KEY
));
};
let wrong = held.len() - well_formed(Some(value)).len();
(wrong > 0).then(|| {
format!(
"{} holds an object of destination source names to qualified ids of that source, \
and {wrong} of its entries is not one: {value}",
MetadataKey::COPIES_KEY
)
})
}
/// The title and metadata of either kind of item.
fn described(item: &Item) -> (&str, &BTreeMap<String, Value>) {
match item {
Item::Task(task) => (&task.title, &task.metadata),
Item::Project(project) => (&project.title, &project.metadata),
Item::Document(document) => (&document.title, &document.metadata),
}
}
/// The item as the destination should hold it.
///
/// `url`, `location`, `created_at` and `updated_at` are the destination's own and are
/// never written — where the *source* holds an item says nothing about where the
/// destination does, which is why a copied document does not arrive claiming the path or
/// the link its source reported. The two reserved keys this product encodes typed fields
/// under are removed, because those fields travel as themselves — leaving the encoding
/// beside them would have the destination hold one thing twice, and disagree with itself
/// the moment one changed.
///
/// A task's `delivers` is `delivers`, already resolved against the destination, and its
/// `delivered_by` is the one the destination holds — never the source's, and empty for an
/// item this copy creates: that list is the store's to keep, at the destination. The
/// `onetaskgraph.copies` link is kept on the same terms: it says where the *destination*
/// item was copied to, which is the destination's to hold, never the source's to send.
fn outgoing(
item: &Planned,
id: NativeId,
project: Option<NativeId>,
origin: &Origin,
delivers: &[TaskRef],
held: Option<&Prior>,
) -> Item {
let own = held.map(|held| described(&held.item).1);
let carried = |metadata: &BTreeMap<String, Value>| carried(metadata, origin, own);
match &item.item {
Item::Task(task) => Item::Task(Box::new(Task {
id,
url: None,
location: None,
created_at: None,
updated_at: None,
project,
metadata: carried(&task.metadata),
delivers: delivers.to_vec(),
delivered_by: match held.map(|held| &held.item) {
Some(Item::Task(held)) => held.delivered_by.clone(),
_ => Vec::new(),
},
..(**task).clone()
})),
Item::Project(project) => Item::Project(Box::new(Project {
id,
url: None,
location: None,
created_at: None,
updated_at: None,
metadata: carried(&project.metadata),
..(**project).clone()
})),
Item::Document(document) => Item::Document(Box::new(Document {
id,
url: None,
location: None,
created_at: None,
updated_at: None,
project,
metadata: carried(&document.metadata),
..(**document).clone()
})),
}
}
/// The id a created item is offered to the destination under.
///
/// A task whose id is `<its project's id>/<rest>` is offered as `<rest>` under the
/// destination project it is filed in, so a destination whose ids are paths files it in that
/// project rather than the source's. Any other id, and every update, is left as it is. The
/// `a_project_copied_*` journeys in `crates/onetaskgraph-e2e/tests/e2e/copy.rs` hold the Markdown
/// plugin's id shape to this rule.
fn created_id(item: &Item, filed: Option<&NativeId>) -> NativeId {
if let (Item::Task(task), Some(filed)) = (item, filed)
&& let Some(own) = &task.project
&& let Some(rest) = task
.id
.as_str()
.strip_prefix(own.as_str())
.and_then(|rest| rest.strip_prefix('/'))
&& !rest.is_empty()
{
return NativeId(format!("{}/{rest}", filed.as_str()));
}
item.id().clone()
}
/// The metadata a copy carries: the caller's own keys untouched, the origin settled, and the
/// destination's own store-kept keys — its `onetaskgraph.copies` link, and the
/// `onetaskgraph.members` or `onetaskgraph.member_of` that ties it to its member projects or
/// its home — kept as `own`, the destination item, holds them.
///
/// The key is removed before it is settled rather than overwritten, because the item being
/// copied carries an origin of its own and [`Origin::Keeps`] must not let it through. Its
/// link is removed for the same reason: it names where *that* item was copied to, and on
/// the destination it would name the destination item's own counterpart as itself. Its
/// member keys name projects tied to *it*, which say nothing about the destination's.
fn carried(
metadata: &BTreeMap<String, Value>,
origin: &Origin,
own: Option<&BTreeMap<String, Value>>,
) -> BTreeMap<String, Value> {
let mut carried = metadata.clone();
carried.remove(Repository::METADATA_KEY);
carried.remove(DependencyEdge::RECORDED_KEY);
carried.remove(TaskRef::DELIVERS_KEY);
carried.remove(TaskRef::DELIVERED_BY_KEY);
carried.remove(GlobalId::ORIGIN_KEY);
for kept in [
MetadataKey::COPIES_KEY,
MetadataKey::MEMBERS_KEY,
MetadataKey::MEMBER_OF_KEY,
MetadataKey::ASSETS_KEY,
] {
carried.remove(kept);
if let Some(held) = own.and_then(|own| own.get(kept)) {
carried.insert(kept.to_owned(), held.clone());
}
}
let held = match origin {
Origin::Records(id) => Some(Value::String(id.to_string())),
Origin::Keeps(held) => held.clone(),
};
if let Some(held) = held {
carried.insert(GlobalId::ORIGIN_KEY.to_owned(), held);
}
carried
}
/// What one landed item records at [`GlobalId::ORIGIN_KEY`].
enum Origin {
/// The qualified id this item was copied from, as the id type rather than as its
/// spelling: the key holds a [`GlobalId`] and nothing else may be recorded there.
Records(GlobalId),
/// Whatever the destination already holds there — `None` when it holds nothing, which
/// is written as the key being absent rather than as a null.
///
/// A [`Value`] and not a [`GlobalId`], because this variant does not interpret what it
/// carries: it is the destination's own metadata entry, held for the length of one
/// write and put back exactly as it was read. Parsing it would turn a value a
/// destination holds and this engine cannot read into a value this engine deletes,
/// which is the opposite of what keeping it means.
Keeps(Option<Value>),
}
/// Which of the two a copy of this item does.
///
/// A copy that reached its target by rule 1 is a copy-back: the item being copied names
/// the destination item, so the destination is the *original* and the id being copied
/// belongs to the copy that came out of it. Recording that id there would overwrite the
/// original's own provenance — and with it the correspondence every later copy from the
/// source it was authored in depends on. That copy would then match nothing and create a
/// second item beside the one it meant to update, which is the whole failure: nothing is
/// reported, and whoever reads that board now has two. So a copy-back leaves the
/// destination's origin exactly as the destination holds it, absent included, and every
/// other copy records the id it was copied from.
fn recorded(item: &Planned, held: Option<&Prior>) -> Origin {
if let Target::Update {
found: Found::Origin,
..
} = &item.target
{
return Origin::Keeps(
held.and_then(|held| described(&held.item).1.get(GlobalId::ORIGIN_KEY).cloned()),
);
}
Origin::Records(item.source.clone())
}
/// Whether the destination already reads exactly as this copy would leave it.
///
/// The destination's own `url` and timestamps are excluded because a copy never writes
/// them, so a difference there is not one this copy would close.
fn same(held: &Item, outgoing: &Item, destination: &SourceName) -> bool {
match (held, outgoing) {
(Item::Task(held), Item::Task(outgoing)) => {
// Qualified before they are compared, so `T-1` and `folder:T-1` at the destination
// `folder` are the one entry they are.
targets(&held.delivers, destination) == targets(&outgoing.delivers, destination)
&& targets(&held.delivered_by, destination)
== targets(&outgoing.delivered_by, destination)
&& held.title == outgoing.title
&& held.content == outgoing.content
&& held.status == outgoing.status
&& held.priority == outgoing.priority
&& held.labels == outgoing.labels
&& held.project == outgoing.project
&& held.metadata == outgoing.metadata
&& held.repositories == outgoing.repositories
}
(Item::Project(held), Item::Project(outgoing)) => {
held.title == outgoing.title
&& held.content == outgoing.content
&& held.status == outgoing.status
&& held.labels == outgoing.labels
&& held.metadata == outgoing.metadata
&& held.repositories == outgoing.repositories
}
// No status, because a document has none; no edges, because it is in no graph.
(Item::Document(held), Item::Document(outgoing)) => {
held.title == outgoing.title
&& held.content == outgoing.content
&& held.labels == outgoing.labels
&& held.project == outgoing.project
&& held.metadata == outgoing.metadata
&& held.repositories == outgoing.repositories
}
_ => false,
}
}
/// Whether the destination's forward edges already say what this copy would write.
fn same_edges(held: &[DependencyEdge], outgoing: &[DependencyEdge]) -> bool {
let ends = |edges: &[DependencyEdge]| {
let mut ends: Vec<(String, ItemKind, DependencyKind)> = edges
.iter()
.map(|edge| (edge.to.id().to_owned(), edge.to.kind, edge.kind))
.collect();
ends.sort_by(|left, right| left.0.cmp(&right.0));
ends
};
ends(held) == ends(outgoing)
}
/// Each read edge as the destination should record it, or `None` when its far end is a
/// member of this copy whose destination id is not known yet.
///
/// `destination` is the source the near end lands in. A far end this copy landed in another
/// source — a routed copy's other half — is written qualified, which every source records
/// as a cross-source edge on the near item.
fn mapped_edges(
edges: &[DependencyEdge],
origin: &SourceName,
destination: &SourceName,
copied: &[GlobalId],
written: &BTreeMap<String, GlobalId>,
) -> Vec<Option<DependencyEdge>> {
edges
.iter()
.map(|edge| {
let far = GlobalId::new(origin.clone(), NativeId(edge.to.id().to_owned()));
let id = if let Some(native) = names(&edge.to, destination) {
// A far end already qualified to the destination's own source is that
// source's own item, so it is written the way that source names its own:
// unqualified. Leaving it qualified would have the destination hold an
// edge into itself written as if it left, which is the one spelling the
// reserved key exists to keep for edges that really do.
Some(native)
} else if !edge.to.is_qualified() && copied.contains(&far) {
// A member of this copy, inside one source included: a copy there lands
// beside the item it was read from, so the edge names the copy rather than
// the original the copy's own project does not hold.
written
.get(&far.to_string())
.map(|landed| spelled_from(landed, destination))
} else if edge.to.is_qualified() || origin == destination {
// Already naming a source of its own, or a copy inside one source where
// the far end's own id is the destination's id.
Some(edge.to.id().to_owned())
} else {
Some(far.to_string())
}?;
DependencyEndpoint::new(id, edge.to.kind)
.ok()
.map(|to| DependencyEdge {
from: edge.from.clone(),
to,
kind: edge.kind,
})
})
.collect()
}
/// Each `delivers` entry of one planned task as the destination should hold it, or `None`
/// where it names a member of this copy whose destination id is not known yet. Empty for
/// anything but a task.
fn delivers_of(
item: &Planned,
destination: &SourceName,
copied: &[GlobalId],
written: &BTreeMap<String, GlobalId>,
) -> Vec<Option<TaskRef>> {
match &item.item {
Item::Task(task) => mapped_delivers(
&task.delivers,
&item.source.source,
destination,
copied,
written,
),
Item::Project(_) | Item::Document(_) => Vec::new(),
}
}
/// Each `delivers` entry as the destination should hold it, or `None` when it names a member
/// of this copy whose destination id is not known yet.
///
/// An entry naming a member of the copied set becomes that member's own id at the
/// destination — bare, because the member is the destination's, unless the id holds a colon
/// a bare entry would be misread by. Every other entry is carried through qualified, so a
/// bare one read at `origin` goes on naming a task of `origin`.
fn mapped_delivers(
entries: &[TaskRef],
origin: &SourceName,
destination: &SourceName,
copied: &[GlobalId],
written: &BTreeMap<String, GlobalId>,
) -> Vec<Option<TaskRef>> {
entries
.iter()
.map(|entry| {
let qualified = entry.in_source(origin);
let Ok(far) = qualified.as_str().parse::<GlobalId>() else {
return Some(qualified);
};
if !copied.contains(&far) {
return Some(qualified);
}
let landed = written.get(&far.to_string())?;
let native = &landed.native;
Some(
if &landed.source != destination || native.as_str().contains(':') {
TaskRef::qualified(&landed.source, native)
} else {
TaskRef::new(native.as_str())
.unwrap_or_else(|_| TaskRef::qualified(destination, native))
},
)
})
.collect()
}
/// How many of one task's `delivers` entries name a member of the copied set.
fn members_named(entries: &[TaskRef], origin: &SourceName, copied: &[GlobalId]) -> u64 {
let named = entries
.iter()
.filter(|entry| {
entry
.in_source(origin)
.as_str()
.parse::<GlobalId>()
.is_ok_and(|far| copied.contains(&far))
})
.count();
u64::try_from(named).unwrap_or(u64::MAX)
}
/// The entries that could be resolved, which is every one of them on the second pass.
fn resolved_entries(entries: &[Option<TaskRef>]) -> Vec<TaskRef> {
entries.iter().flatten().cloned().collect()
}
/// How an item landed at `landed` is named from one landed in `near`: by its native id in
/// the same source, qualified from any other.
fn spelled_from(landed: &GlobalId, near: &SourceName) -> String {
if &landed.source == near {
landed.native.0.clone()
} else {
landed.to_string()
}
}
/// The native id a qualified endpoint names at `destination`, when it names one there.
fn names(endpoint: &DependencyEndpoint, destination: &SourceName) -> Option<String> {
if !endpoint.is_qualified() {
return None;
}
let id: GlobalId = endpoint.id().parse().ok()?;
(&id.source == destination).then_some(id.native.0)
}
/// Where a project copy files the tasks it carries: in each source a task lands in, the
/// home there or the home's member project.
type Filing = BTreeMap<SourceName, NativeId>;
/// The edges that could be resolved, which is every one of them on the second pass.
fn resolved(edges: &[Option<DependencyEdge>]) -> Vec<DependencyEdge> {
edges.iter().flatten().cloned().collect()
}
/// Refuse an item of a member copy whose edge names a member that copy does not carry
/// and whose destination id nothing records.
///
/// `unrecorded` is empty for every other copy, which is what makes this a no-op there. An
/// edge already naming a source of its own, and every edge of a copy inside one source,
/// is written as it was read by [`mapped_edges`], so neither can need a recorded origin.
fn unrecorded_far_end(item: &Planned, unrecorded: &[GlobalId]) -> Result<(), EngineError> {
if item.source.source == item.to {
return Ok(());
}
for edge in &item.edges {
if edge.to.is_qualified() {
continue;
}
let far = GlobalId::new(
item.source.source.clone(),
NativeId(edge.to.id().to_owned()),
);
if unrecorded.contains(&far) {
return Err(EngineError::UnrecordedMember {
item: item.source.clone(),
member: far,
destination: item.to.clone(),
});
}
}
Ok(())
}