onetaskgraph_github_projects/
expected.rs

1//! What this source does for a write the store's local cache expects.
2//!
3//! The store keeps each task it read beside what this source held about it — the
4//! [`Resolved`] record a write takes its node ids, its board item, its field options and its
5//! current values from — and the board's field definitions beside every task. A write the
6//! caller says it knows the task at a given version of hands both back through
7//! [`TaskSource::prime_write`](onetaskgraph_plugin_api::TaskSource::prime_write), and every
8//! write path here already takes a record held in this process instead of reading the issue
9//! again: so the write sends its mutations and nothing else.
10//!
11//! While such a write runs a journal is open. Every request is recorded with what it came to —
12//! landed, refused, or lost — and every mutation that has a twin selecting the whole issue
13//! (`graphql::whole_twin`) is sent as that twin, so its response vouches for what the issue
14//! holds afterwards, at the version its change token spells, and the store can advance its
15//! entry without reading. Nothing here outlives the process: the store, not this source, is
16//! what keeps a record between commands.
17
18use std::collections::BTreeMap;
19use std::sync::{Mutex, MutexGuard};
20
21use onetaskgraph_plugin_api::{
22    AssetUploads, JournalEntry, NativeId, PartOutcome, Repository, RequestKind, Shown, SourceError,
23    WholeTask, WriteJournal,
24};
25use serde::{Deserialize, Serialize};
26use serde_json::{Value, json};
27
28use super::{
29    BOARD_ITEMS_PAGE_SIZE, BoardFields, BoardKind, GitHubProjectsSource, ItemKind, MAX_PAGE_SIZE,
30    NESTED_PAGE_SIZE, RepositoryTarget, Resolved, SERVED_AT, change_token, comment_page, facets_of,
31    graphql, is_mutation, operation_description, optional_time,
32};
33
34/// The version of the context this source hands the store, so a context an earlier or a later
35/// build wrote is one this build declines rather than misreads.
36///
37/// `tests/context.rs` pins it beside the shape [`context_shapes`] answers, so a change to what a
38/// context holds that does not move this number fails there.
39pub const CONTEXT_VERSION: u64 = 3;
40
41/// The JSON Schema of the task context and the board context this source hands the store.
42#[must_use]
43pub fn context_shapes() -> Value {
44    json!({
45        "task": schemars::schema_for!(Context),
46        "board": schemars::schema_for!(Board),
47    })
48}
49
50/// One task's context: its record, and the comments on it this process read, by id, with the
51/// body GitHub holds — which an edit keeps a copy's origin marker from.
52#[derive(Serialize, Deserialize, schemars::JsonSchema)]
53struct Context {
54    version: u64,
55    item: Resolved,
56    /// Every comment on the task, when a read carried them all, and `None` when none did — so
57    /// an edit of a comment the context cannot vouch for reads which issue it is on, rather than
58    /// taking an unread set for an empty one.
59    comments: Option<BTreeMap<NativeId, String>>,
60}
61
62/// The board's context: its own id and field definitions.
63#[derive(Serialize, Deserialize, schemars::JsonSchema)]
64struct Board {
65    version: u64,
66    fields: BoardFields,
67}
68
69/// What this source holds for expected writes, for the length of one process.
70#[derive(Default)]
71pub(super) struct Held {
72    /// The journal, while one is open.
73    journal: Mutex<Option<WriteJournal>>,
74    /// The comments this process read on each issue, with the body each holds.
75    learnt: Mutex<BTreeMap<NativeId, BTreeMap<NativeId, String>>>,
76    /// The comments a primed context says each issue holds, `None` where it read none.
77    primed: Mutex<BTreeMap<NativeId, Option<BTreeMap<NativeId, String>>>>,
78    /// The design documents this process read by id, as it resolved them: what a write of one
79    /// the engine has just held to that read's version takes instead of reading it again, handed
80    /// back through [`GitHubProjectsSource::context_of`]. Dropped with the command.
81    documents: Mutex<BTreeMap<NativeId, Resolved>>,
82    /// Issues an expected write was primed for and then let go of — to read afresh after a
83    /// stale id — which still send only what differs from what they read, so a part that
84    /// already landed is not sent again. Dropped with the command.
85    narrowed: Mutex<std::collections::BTreeSet<NativeId>>,
86}
87
88/// A lock an earlier panic poisoned is read as it stands: what these hold is an
89/// optimisation, and a write that cannot use it reads instead.
90fn held<T>(lock: &Mutex<T>) -> MutexGuard<'_, T> {
91    lock.lock()
92        .unwrap_or_else(std::sync::PoisonError::into_inner)
93}
94
95impl Held {
96    /// Whether a journal is open.
97    pub(super) fn journaling(&self) -> bool {
98        held(&self.journal).is_some()
99    }
100
101    /// Open a journal.
102    pub(super) fn begin(&self) {
103        *held(&self.journal) = Some(WriteJournal::default());
104    }
105
106    /// Close the journal, answering with what it recorded.
107    pub(super) fn end(&self) -> WriteJournal {
108        held(&self.journal).take().unwrap_or_default()
109    }
110
111    fn record(&self, entry: JournalEntry) {
112        if let Some(journal) = held(&self.journal).as_mut() {
113            journal.requests.push(entry);
114        }
115    }
116
117    /// Learn the comments one detail read of `issue` carried, when it carried them all.
118    pub(super) fn learn_comments(&self, issue: &NativeId, node: &Value) {
119        let Some(connection) = node.get("comments") else {
120            return;
121        };
122        if connection.pointer("/pageInfo/hasNextPage") != Some(&Value::Bool(false)) {
123            return;
124        }
125        let Some(nodes) = connection.get("nodes").and_then(Value::as_array) else {
126            return;
127        };
128        // Every comment read whole, or nothing learnt: a set missing one is not the set the issue
129        // holds, and taking it for that would answer an edit of the missing one as no comment
130        // of this issue.
131        let Some(comments) = nodes
132            .iter()
133            .map(|comment| {
134                Some((
135                    NativeId(comment.get("id")?.as_str()?.to_owned()),
136                    comment.get("body")?.as_str()?.to_owned(),
137                ))
138            })
139            .collect::<Option<BTreeMap<_, _>>>()
140        else {
141            held(&self.learnt).remove(issue);
142            return;
143        };
144        held(&self.learnt).insert(issue.clone(), comments);
145    }
146
147    /// Forget every issue an expected write let go of, and every document read: a new command
148    /// writes as it always did.
149    pub(super) fn end_command(&self) {
150        held(&self.narrowed).clear();
151        held(&self.documents).clear();
152    }
153
154    /// Keep `item`, a design document a read by id resolved, as what an expected write of it
155    /// may take.
156    pub(super) fn read_document(&self, item: &Resolved) {
157        if item.kind == BoardKind::Document {
158            held(&self.documents).insert(item.id.clone(), item.clone());
159        }
160    }
161
162    /// The body comment `comment` holds on `issue`, by a primed context: `Some(Some(body))`
163    /// when it is one of the issue's, `Some(None)` when it is not, and `None` when no primed
164    /// context says.
165    pub(super) fn primed_comment(
166        &self,
167        issue: &NativeId,
168        comment: &NativeId,
169    ) -> Option<Option<String>> {
170        held(&self.primed)
171            .get(issue)
172            .and_then(Option::as_ref)
173            .map(|comments| comments.get(comment).cloned())
174    }
175}
176
177/// `variables`, with the page sizes every whole-issue twin declares.
178pub(super) fn whole_variables(mut variables: Value) -> Value {
179    if let Some(members) = variables.as_object_mut() {
180        members.insert("first".to_owned(), json!(MAX_PAGE_SIZE));
181        members.insert("nestedFirst".to_owned(), json!(NESTED_PAGE_SIZE));
182        members.insert("boardItems".to_owned(), json!(BOARD_ITEMS_PAGE_SIZE));
183        members.insert("duplicates".to_owned(), json!(true));
184    }
185    variables
186}
187
188/// Which part of a write `document` is, in the words a write's `result` names it by.
189fn part_of(document: &str) -> String {
190    let part = if [
191        graphql::ADD_COMMENT,
192        graphql::UPDATE_COMMENT,
193        graphql::DELETE_COMMENT,
194        graphql::ADD_COMMENT_WHOLE,
195        graphql::UPDATE_COMMENT_WHOLE,
196    ]
197    .contains(&document)
198    {
199        "comment"
200    } else if [graphql::UPDATE_ISSUE, graphql::UPDATE_ISSUE_WHOLE].contains(&document) {
201        "issue"
202    } else if [
203        graphql::UPDATE_FIELD,
204        graphql::UPDATE_FIELDS,
205        graphql::CLEAR_FIELD,
206        graphql::UPDATE_FIELD_WHOLE,
207        graphql::CLEAR_FIELD_WHOLE,
208    ]
209    .contains(&document)
210    {
211        "fields"
212    } else if [graphql::ADD_BLOCKED_BY, graphql::REMOVE_BLOCKED_BY].contains(&document) {
213        "dependencies"
214    } else if [graphql::ADD_SUB_ISSUE, graphql::REMOVE_SUB_ISSUE].contains(&document) {
215        "project"
216    } else if document == graphql::UPDATE_DRAFT {
217        "draft"
218    } else {
219        return operation_description(document).to_owned();
220    };
221    part.to_owned()
222}
223
224/// Whether `error` is a refusal of an id a primed context supplied that no longer resolves:
225/// a node deleted since it was cached, or a single-select option renamed or deleted.
226fn stale(error: &SourceError) -> bool {
227    let SourceError::Refused { message } = error else {
228        return false;
229    };
230    let message = message.to_ascii_lowercase();
231    message.contains(super::UNRESOLVABLE_NODE) || message.contains("single select option")
232}
233
234/// What one request came to, as a journal records it.
235fn outcome(sent: &Result<Value, SourceError>) -> PartOutcome {
236    match sent {
237        Ok(_) => PartOutcome::Landed,
238        // The request left, and its answer did not come back whole: whether GitHub applied it
239        // cannot be known, and sending it again could apply it twice.
240        Err(SourceError::Unavailable { message })
241            if message.starts_with("GitHub GraphQL request failed")
242                || message.starts_with("GitHub GraphQL response could not be read") =>
243        {
244            PartOutcome::Unknown {
245                error: message.clone(),
246            }
247        }
248        Err(error) => PartOutcome::Rejected {
249            error: error.to_string(),
250            stale: stale(error),
251        },
252    }
253}
254
255impl GitHubProjectsSource {
256    /// The context of `id`: the record this process last resolved for it, and the comments it
257    /// read on it — or, for a design document, the record its last read by id resolved, which
258    /// carries no comments because a document's write never touches one.
259    pub(super) fn context_of(&self, id: &NativeId) -> Option<Value> {
260        let task = self
261            .resolved_cache()
262            .ok()?
263            .get(id)
264            .filter(|item| item.kind == BoardKind::Work(ItemKind::Task))
265            .cloned();
266        let Some(item) = task else {
267            let item = held(&self.expected.documents).get(id).cloned()?;
268            return serde_json::to_value(Context {
269                version: CONTEXT_VERSION,
270                item,
271                comments: None,
272            })
273            .ok();
274        };
275        let comments = held(&self.expected.learnt).get(id).cloned();
276        serde_json::to_value(Context {
277            version: CONTEXT_VERSION,
278            item,
279            comments,
280        })
281        .ok()
282    }
283
284    /// The board's context, when this process read the board's fields — on their own, or
285    /// carried by a read of an issue by its own id, which selects every field the board
286    /// defines.
287    pub(super) fn board_context_held(&self) -> Option<Value> {
288        let held = self.fields_cache().ok()?.clone();
289        let fields = match held {
290            Some(fields) => fields,
291            None => self
292                .resolved_cache()
293                .ok()?
294                .values()
295                .find_map(Resolved::carried_board)?,
296        };
297        serde_json::to_value(Board {
298            version: CONTEXT_VERSION,
299            fields,
300        })
301        .ok()
302    }
303
304    /// Take `context` as the record of `id`, and `board` as the board's fields.
305    pub(super) fn prime(&self, id: &NativeId, context: &Value, board: Option<&Value>) -> bool {
306        let Ok(context) = serde_json::from_value::<Context>(context.clone()) else {
307            return false;
308        };
309        // Only a task or a design document of this board primes a write: a context naming a
310        // project is not one any write may take. A write of the other kind than the context's
311        // finds no item of its own kind in it, and answers exactly as it would after a read.
312        if context.version != CONTEXT_VERSION
313            || context.item.id != *id
314            || !matches!(
315                context.item.kind,
316                BoardKind::Work(ItemKind::Task) | BoardKind::Document
317            )
318        {
319            return false;
320        }
321        let mut item = context.item;
322        let named = item.named_board().map(|named| named.as_str().to_owned());
323        let board = board
324            .and_then(|board| serde_json::from_value::<Board>(board.clone()).ok())
325            .filter(|board| board.version == CONTEXT_VERSION);
326        if let Some(board) = &board {
327            // The board's own fields are what every task of it writes against, and the board
328            // context is refreshed whenever a write finds an option re-minted — so it is
329            // preferred to the definitions the task's own record carried when it was read.
330            if item
331                .named_board()
332                .is_some_and(|named| named.as_str() == board.fields.id.as_str())
333            {
334                item.board_fields = Some(board.fields.fields.clone());
335            }
336        }
337        let Ok(mut cache) = self.resolved_cache() else {
338            return false;
339        };
340        cache.insert(id.clone(), item);
341        drop(cache);
342        held(&self.expected.primed).insert(id.clone(), context.comments);
343        // Held as this board's own fields only when they are this task's board's.
344        if let Some(board) = board
345            && named.as_deref() == Some(board.fields.id.as_str())
346            && let Ok(mut fields) = self.fields_cache()
347            && fields.is_none()
348        {
349            *fields = Some(board.fields);
350        }
351        true
352    }
353
354    /// Drop what [`Self::prime`] took for `id`, and the board's fields, so a write reads them.
355    pub(super) fn unprime_held(&self, id: &NativeId) {
356        if let Ok(mut cache) = self.resolved_cache() {
357            cache.remove(id);
358        }
359        if held(&self.expected.primed).remove(id).is_some() {
360            held(&self.expected.narrowed).insert(id.clone());
361            if let Ok(mut fields) = self.fields_cache() {
362                *fields = None;
363            }
364        }
365    }
366
367    /// Record one request sent while a journal is open.
368    pub(super) async fn journal(
369        &self,
370        document: &str,
371        variables: &Value,
372        sent: &Result<Value, SourceError>,
373    ) {
374        let mutation = is_mutation(document);
375        let mut entry = JournalEntry {
376            part: part_of(document),
377            kind: if mutation {
378                RequestKind::Mutation
379            } else {
380                RequestKind::Read
381            },
382            outcome: outcome(sent),
383            fields: BTreeMap::new(),
384            shown: Shown::Nothing,
385        };
386        if mutation && let Ok(data) = sent {
387            let (issue, fields) = read_back(document, variables, data);
388            entry.fields = fields;
389            if let Some(issue) = issue {
390                let version = self.token_of(issue);
391                entry.shown = match (self.whole_of(issue, version.as_deref()), version) {
392                    (Some(task), Some(_)) => Shown::Whole {
393                        task: Box::new(task),
394                    },
395                    (None, Some(version)) => Shown::Version { version },
396                    (_, None) => Shown::Nothing,
397                };
398                // What a response vouches for is the issue's state now, so a comment learnt
399                // on it is learnt afresh rather than added to what was read before.
400                if let Some(id) = issue.get("id").and_then(Value::as_str) {
401                    self.expected
402                        .learn_comments(&NativeId(id.to_owned()), issue);
403                }
404            }
405        }
406        self.expected.record(entry);
407    }
408
409    /// The change token `issue` spells, as a response shows it: every part present — its
410    /// facets read by exactly the rules a read takes them by — and the issue settled by the
411    /// response's own clock.
412    fn token_of(&self, issue: &Value) -> Option<String> {
413        let updated = optional_time(issue, "updatedAt").ok()??;
414        let nodes = issue.pointer("/projectItems/nodes")?.as_array()?;
415        let entry = self.board_entry(nodes)?;
416        let item = optional_time(entry, "updatedAt").ok()??;
417        let served = optional_time(issue, SERVED_AT).ok()??;
418        let parts = change_token::Parts {
419            issue: updated,
420            item,
421            comments: issue.pointer("/commentCount/totalCount")?.as_u64()?,
422            relationships: issue.pointer("/relationshipEvents/totalCount")?.as_u64()?,
423            facets: facets_of(issue, entry.get("fieldValues")?)?,
424            repository: Repository::try_from(format!(
425                "{}/{}",
426                RepositoryTarget::HOST,
427                issue.pointer("/repository/nameWithOwner")?.as_str()?
428            ))
429            .ok()?,
430        };
431        change_token::settled(parts.issue, parts.item, served).then(|| change_token::spell(&parts))
432    }
433
434    /// The task `issue` is, whole, at `version` — when the response carried every part the
435    /// store's entry holds: the issue on this board, its every comment, and a token.
436    ///
437    /// Every member the entry keeps has to be in the response, each connection read to its
438    /// end: the reader a read goes through takes a member left out for an empty one, which is
439    /// the right reading of a read GitHub answered and the wrong one of a response that simply
440    /// did not select it.
441    fn whole_of(&self, issue: &Value, version: Option<&str>) -> Option<WholeTask> {
442        version?;
443        let carried = |pointer: &str| issue.pointer(pointer).is_some();
444        let ended = |pointer: &str| {
445            issue.pointer(&format!("{pointer}/pageInfo/hasNextPage")) == Some(&Value::Bool(false))
446        };
447        let shown = WHOLE_MEMBERS.iter().all(|member| carried(member))
448            && ended("/labels")
449            && ended("/comments");
450        if !shown {
451            return None;
452        }
453        let nodes = issue.pointer("/projectItems/nodes")?.as_array()?;
454        let held = self.board_entry(nodes)?;
455        if held.pointer("/fieldValues/pageInfo/hasNextPage") != Some(&Value::Bool(false)) {
456            return None;
457        }
458        let item = json!({
459            "id": held.get("id")?,
460            "updatedAt": held.get("updatedAt"),
461            "project": held.get("project"),
462            "fieldValues": held.get("fieldValues"),
463            "content": issue,
464        });
465        let resolved = self.resolve(&item).ok()??;
466        if resolved.kind != BoardKind::Work(ItemKind::Task) {
467            return None;
468        }
469        let task = resolved.task().ok()?;
470        if task.version.as_deref() != version {
471            return None;
472        }
473        let page = comment_page(issue, &resolved.id.0, None).ok()?;
474        if page.next.is_some() {
475            return None;
476        }
477        let assets = AssetUploads::read(&resolved.slot)
478            .ok()?
479            .unwrap_or_default()
480            .0
481            .into_iter()
482            .map(|(name, upload)| onetaskgraph_plugin_api::Asset {
483                content_type: name.content_type(),
484                name,
485                sha256: upload.sha256,
486                path: None,
487            })
488            .collect();
489        WholeTask::new(task, page.items, assets, None)
490    }
491}
492
493/// Every member of an issue a response has to carry for the store's entry to take it whole.
494///
495/// `tests/context.rs` reconciles it against the issue fragment every read selects, so a member
496/// the reader comes to need is one this list has to name.
497pub const WHOLE_MEMBERS: [&str; 15] = [
498    "/id",
499    "/number",
500    "/title",
501    "/body",
502    "/url",
503    "/createdAt",
504    "/state",
505    "/stateReason",
506    "/repository",
507    "/parent",
508    "/subIssuesSummary",
509    "/labels/nodes",
510    "/projectItems/nodes",
511    "/comments/nodes",
512    "/updatedAt",
513];
514
515/// The issue a mutation's response shows, and each field the mutation set with the value that
516/// response shows it holding — `null` where it shows none.
517fn read_back<'a>(
518    document: &str,
519    variables: &Value,
520    data: &'a Value,
521) -> (Option<&'a Value>, BTreeMap<String, Value>) {
522    let mut fields = BTreeMap::new();
523    let issue = if [graphql::ADD_COMMENT, graphql::ADD_COMMENT_WHOLE].contains(&document) {
524        if let Some(added) = data.pointer("/addComment/commentEdge/node") {
525            fields.insert("comment".to_owned(), comment_fields(added));
526        }
527        data.pointer("/addComment/subject")
528    } else if [graphql::UPDATE_COMMENT, graphql::UPDATE_COMMENT_WHOLE].contains(&document) {
529        let edited = data.pointer("/updateIssueComment/issueComment");
530        if let Some(edited) = edited {
531            fields.insert("comment".to_owned(), comment_fields(edited));
532        }
533        edited.and_then(|edited| edited.get("issue"))
534    } else if document == graphql::DELETE_COMMENT {
535        fields.insert(
536            "comment".to_owned(),
537            json!({"deleted": variables.pointer("/input/id")}),
538        );
539        None
540    } else if [graphql::UPDATE_ISSUE, graphql::UPDATE_ISSUE_WHOLE].contains(&document) {
541        let issue = data.pointer("/updateIssue/issue");
542        let input = variables.get("input");
543        for (asked, shown) in [
544            ("title", "title"),
545            ("body", "body"),
546            ("stateInput", "state"),
547        ] {
548            if input.and_then(|input| input.get(asked)).is_some() {
549                fields.insert(
550                    shown.to_owned(),
551                    issue
552                        .and_then(|issue| issue.get(shown))
553                        .cloned()
554                        .unwrap_or(Value::Null),
555                );
556            }
557        }
558        issue
559    } else if [graphql::UPDATE_FIELD, graphql::UPDATE_FIELD_WHOLE].contains(&document) {
560        let item = data.pointer("/updateProjectV2ItemFieldValue/projectV2Item");
561        field_values(variables, item, &mut fields);
562        item.and_then(|item| item.get("content"))
563    } else if [graphql::CLEAR_FIELD, graphql::CLEAR_FIELD_WHOLE].contains(&document) {
564        let item = data.pointer("/clearProjectV2ItemFieldValue/projectV2Item");
565        field_values(variables, item, &mut fields);
566        item.and_then(|item| item.get("content"))
567    } else if document == graphql::UPDATE_FIELDS {
568        // Each aliased field write answers with its board item's id alone, so the values it
569        // set are not shown: each field it set is named, and holds what no response showed.
570        for argument in ["input", "second", "third", "fourth", "fifth", "sixth"] {
571            if let Some(field) = variables
572                .pointer(&format!("/{argument}/fieldId"))
573                .and_then(Value::as_str)
574            {
575                fields.insert(field.to_owned(), Value::Null);
576            }
577        }
578        None
579    } else {
580        None
581    };
582    (issue.filter(|issue| !issue.is_null()), fields)
583}
584
585/// What a write's result says of one comment it wrote: its id, its address and its body.
586fn comment_fields(comment: &Value) -> Value {
587    json!({
588        "id": comment.get("id"),
589        "url": comment.get("url"),
590        "body": comment.get("body"),
591    })
592}
593
594/// The field one field write set, by its name, with the value its response shows the board
595/// item holding there — read off the issue's own memberships, when the response carried them.
596fn field_values(variables: &Value, item: Option<&Value>, fields: &mut BTreeMap<String, Value>) {
597    let Some(field) = variables.pointer("/input/fieldId").and_then(Value::as_str) else {
598        return;
599    };
600    let memberships = item
601        .and_then(|item| item.pointer("/content/projectItems/nodes"))
602        .and_then(Value::as_array);
603    let item_id = item.and_then(|item| item.get("id")).and_then(Value::as_str);
604    let shown = memberships.and_then(|nodes| {
605        let entry = nodes
606            .iter()
607            .find(|node| node.get("id").and_then(Value::as_str) == item_id)?;
608        entry
609            .pointer("/fieldValues/nodes")?
610            .as_array()?
611            .iter()
612            .find(|value| value.pointer("/field/id").and_then(Value::as_str) == Some(field))
613    });
614    match shown {
615        Some(value) => {
616            let name = value
617                .pointer("/field/name")
618                .and_then(Value::as_str)
619                .unwrap_or(field)
620                .to_owned();
621            let held = value
622                .get("name")
623                .or_else(|| value.get("text"))
624                .cloned()
625                .unwrap_or(Value::Null);
626            fields.insert(name, held);
627        }
628        None => {
629            // A cleared field holds no value, and a response without the item's memberships
630            // shows none: either way the field is named and holds nothing shown.
631            fields.insert(field.to_owned(), Value::Null);
632        }
633    }
634}
onetaskgraph_github_projects/expected.rs

onetaskgraph_github_projects/
expected.rs
