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

onetaskgraph_github_projects/
expected.rs
