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

onetaskgraph_github_projects/
expected.rs
