1#![deny(missing_docs)]
117
118use std::collections::BTreeMap;
119use std::sync::Mutex;
120
121use chrono::{DateTime, Utc};
122use onetaskgraph_plugin_api::{
123 Capabilities, Cursor, DependencyEdge, DependencyEndpoint, DependencyKind, DependencySupport,
124 Direction, Health, ItemKind, ItemWrite, Label, LabelFilter, NativeId, Page, PageRequest,
125 Project, ProjectFilter, ProjectQuery, Repository, SecretResolver, SourceError, SourceName,
126 SourcePlugin, Status, StatusCategory, Support, Task, TaskQuery, TaskSource, TextFields,
127 TextQuery, WriteSupport,
128};
129use reqwest::{Client, StatusCode, Url};
130use schemars::{Schema, schema_for};
131use secrecy::{ExposeSecret, SecretString};
132use serde::Deserialize;
133use serde_json::{Value, json};
134
135pub const KIND: &str = "github-projects";
137pub const MAX_PAGE_SIZE: u32 = 100;
139const NESTED_PAGE_SIZE: u32 = 50;
141
142pub mod graphql {
149 pub const BOARD: &str = r#"query($owner:String!,$number:Int!,$first:Int!,$after:String,$nestedFirst:Int!,$duplicates:Boolean!){
151 owner:repositoryOwner(login:$owner){
152 ... on ProjectV2Owner{projectV2(number:$number){...Board}}
153 }
154 } fragment Board on ProjectV2 { id title
155 fields(first:$nestedFirst){nodes{
156 ... on ProjectV2SingleSelectField{__typename id name options{id name}}
157 ... on ProjectV2Field{__typename id name}
158 }pageInfo{hasNextPage}}
159 items(first:$first,after:$after){nodes{id fieldValues(first:$nestedFirst){nodes{
160 ... on ProjectV2ItemFieldSingleSelectValue{name field{
161 ... on ProjectV2SingleSelectField{id name options{id name}}
162 }}
163 ... on ProjectV2ItemFieldTextValue{text field{... on ProjectV2Field{id name}}}
164 ... on ProjectV2ItemFieldLabelValue{labels(first:$nestedFirst){nodes{id name color}pageInfo{hasNextPage}}}
165 }pageInfo{hasNextPage}} content{
166 ... on Issue{__typename id title body url createdAt updatedAt state stateReason(enableDuplicate:$duplicates) repository{nameWithOwner} parent{id} subIssuesSummary{total} labels(first:$nestedFirst){nodes{id name color}pageInfo{hasNextPage}}}
167 ... on PullRequest{__typename id}
168 ... on DraftIssue{__typename id title body createdAt updatedAt}
169 }} pageInfo{hasNextPage endCursor}}
170 }"#;
171 pub const REPOSITORY: &str = r#"query($owner:String!,$name:String!){repository(owner:$owner,name:$name){id nameWithOwner}}"#;
173 pub const ISSUE_DEPENDENCIES: &str = r#"query($id:ID!,$first:Int!,$after:String){node(id:$id){__typename
175 ... on Issue{
176 blockedBy(first:$first,after:$after){nodes{...Related}pageInfo{hasNextPage endCursor}}
177 blocking(first:$first,after:$after){nodes{...Related}pageInfo{hasNextPage endCursor}}
178 }}} fragment Related on Issue{id body parent{id} subIssuesSummary{total}}"#;
179 pub const CREATE_ISSUE: &str =
181 r#"mutation($input:CreateIssueInput!){createIssue(input:$input){issue{id}}}"#;
182 pub const ADD_TO_BOARD: &str = r#"mutation($input:AddProjectV2ItemByIdInput!){addProjectV2ItemById(input:$input){item{id}}}"#;
184 pub const UPDATE_ISSUE: &str =
186 r#"mutation($input:UpdateIssueInput!){updateIssue(input:$input){issue{id}}}"#;
187 pub const UPDATE_DRAFT: &str = r#"mutation($input:UpdateProjectV2DraftIssueInput!){updateProjectV2DraftIssue(input:$input){draftIssue{id}}}"#;
189 pub const UPDATE_FIELD: &str = r#"mutation($input:UpdateProjectV2ItemFieldValueInput!){updateProjectV2ItemFieldValue(input:$input){projectV2Item{id}}}"#;
191 pub const ADD_SUB_ISSUE: &str =
193 r#"mutation($input:AddSubIssueInput!){addSubIssue(input:$input){issue{id} subIssue{id}}}"#;
194 pub const REMOVE_SUB_ISSUE: &str = r#"mutation($input:RemoveSubIssueInput!){removeSubIssue(input:$input){issue{id} subIssue{id}}}"#;
196 pub const ADD_BLOCKED_BY: &str = r#"mutation($input:AddBlockedByInput!){addBlockedBy(input:$input){issue{id} blockingIssue{id}}}"#;
198 pub const REMOVE_BLOCKED_BY: &str = r#"mutation($input:RemoveBlockedByInput!){removeBlockedBy(input:$input){issue{id} blockingIssue{id}}}"#;
200 pub const DELETE_ISSUE: &str =
206 r#"mutation($input:DeleteIssueInput!){deleteIssue(input:$input){repository{id}}}"#;
207}
208
209fn default_token_env() -> String {
210 "GH_PROJECTS_TOKEN".to_owned()
211}
212fn default_endpoint() -> String {
213 "https://api.github.com/graphql".to_owned()
214}
215
216#[derive(Debug, Clone, Deserialize, schemars::JsonSchema)]
221#[serde(untagged)]
222pub enum StatusTargetConfig {
223 Column(ColumnName),
225 Closed {
227 closed: ClosedState,
229 },
230}
231
232#[derive(Debug, Clone, PartialEq, Eq, Deserialize, schemars::JsonSchema)]
237#[serde(try_from = "String")]
238pub struct ColumnName(String);
239
240impl ColumnName {
241 fn as_str(&self) -> &str {
243 &self.0
244 }
245}
246
247impl TryFrom<String> for ColumnName {
248 type Error = String;
249
250 fn try_from(name: String) -> Result<Self, Self::Error> {
251 if name.trim().is_empty() {
252 return Err("a status_mapping option name cannot be blank".to_owned());
253 }
254 Ok(Self(name))
255 }
256}
257
258#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, schemars::JsonSchema)]
263#[serde(rename_all = "kebab-case")]
264pub enum ClosedState {
265 Completed,
267 NotPlanned,
269}
270
271impl ClosedState {
272 const fn reason(self) -> &'static str {
273 match self {
274 Self::Completed => "COMPLETED",
275 Self::NotPlanned => "NOT_PLANNED",
276 }
277 }
278}
279
280#[derive(Debug, Clone, Default, Deserialize, schemars::JsonSchema)]
282#[serde(default, deny_unknown_fields)]
283pub struct GitHubProjectsConfig {
284 pub owner: String, pub project_number: u32, pub repository: Option<String>, #[serde(default = "default_token_env")]
297 pub token_env: String, #[serde(default = "default_endpoint")]
300 pub endpoint: String, #[serde(default)]
308 pub status_mapping: BTreeMap<String, Option<StatusTargetConfig>>, }
310
311#[derive(Debug, Clone, Copy, Default)]
313pub struct Plugin;
314
315impl SourcePlugin for Plugin {
316 fn kind(&self) -> &'static str {
317 KIND
318 }
319 fn config_schema(&self) -> Schema {
320 schema_for!(GitHubProjectsConfig)
321 }
322 fn build(
323 &self,
324 name: &SourceName,
325 config: &Value,
326 secrets: &dyn SecretResolver,
327 ) -> Result<Box<dyn TaskSource>, SourceError> {
328 let config: GitHubProjectsConfig =
329 serde_json::from_value(config.clone()).map_err(|e| SourceError::Config {
330 message: format!("source {name}: {e}"),
331 })?;
332 let source =
333 GitHubProjectsSource::new(name, config, secrets).map_err(|error| match error {
334 SourceError::Config { message } => SourceError::Config {
335 message: format!("source {name}: {message}"),
336 },
337 SourceError::Auth { message } => SourceError::Auth {
338 message: format!("source {name}: {message}"),
339 },
340 other => other,
341 })?;
342 Ok(Box::new(source))
343 }
344}
345
346#[derive(Debug, Clone, PartialEq, Eq)]
348enum StatusTarget {
349 Disabled,
351 Column(ColumnName),
353 Closed(ClosedState),
355}
356
357pub const CATEGORIES: [StatusCategory; 7] = [
367 StatusCategory::Draft,
368 StatusCategory::Backlog,
369 StatusCategory::Todo,
370 StatusCategory::InProgress,
371 StatusCategory::Done,
372 StatusCategory::Cancelled,
373 StatusCategory::Unknown,
374];
375
376#[must_use]
378pub const fn category_position(category: StatusCategory) -> usize {
379 match category {
380 StatusCategory::Draft => 0,
381 StatusCategory::Backlog => 1,
382 StatusCategory::Todo => 2,
383 StatusCategory::InProgress => 3,
384 StatusCategory::Done => 4,
385 StatusCategory::Cancelled => 5,
386 StatusCategory::Unknown => 6,
387 }
388}
389
390fn category_name(category: StatusCategory) -> &'static str {
392 match category {
393 StatusCategory::Draft => "draft",
394 StatusCategory::Backlog => "backlog",
395 StatusCategory::Todo => "todo",
396 StatusCategory::InProgress => "in-progress",
397 StatusCategory::Done => "done",
398 StatusCategory::Cancelled => "cancelled",
399 StatusCategory::Unknown => "unknown",
400 }
401}
402
403fn shipped_column(name: &'static str) -> ColumnName {
408 ColumnName::try_from(name.to_owned()).expect("a shipped default names a board option")
409}
410
411fn shipped_default(category: StatusCategory) -> StatusTarget {
413 match category {
414 StatusCategory::Backlog => StatusTarget::Column(shipped_column("Backlog")),
415 StatusCategory::Todo => StatusTarget::Column(shipped_column("Todo")),
416 StatusCategory::InProgress => StatusTarget::Column(shipped_column("In Progress")),
417 StatusCategory::Done => StatusTarget::Closed(ClosedState::Completed),
418 StatusCategory::Cancelled => StatusTarget::Closed(ClosedState::NotPlanned),
419 StatusCategory::Draft | StatusCategory::Unknown => StatusTarget::Disabled,
420 }
421}
422
423#[derive(Debug, Clone)]
429struct StatusMapping {
430 targets: [StatusTarget; CATEGORIES.len()],
431}
432
433impl StatusMapping {
434 fn resolve(
435 configured: BTreeMap<String, Option<StatusTargetConfig>>,
436 instance: &SourceName,
437 ) -> Result<Self, SourceError> {
438 let mut overrides: BTreeMap<&'static str, Option<StatusTargetConfig>> = BTreeMap::new();
439 for (key, value) in configured {
440 let category = CATEGORIES
441 .iter()
442 .find(|category| category_name(**category) == key)
443 .ok_or_else(|| SourceError::Config {
444 message: format!(
445 "status_mapping names {key:?}, which is not a status category of source \
446 {instance}; the categories are {}",
447 CATEGORIES
448 .iter()
449 .map(|category| category_name(*category))
450 .collect::<Vec<_>>()
451 .join(", ")
452 ),
453 })?;
454 overrides.insert(category_name(*category), value);
455 }
456 let targets = CATEGORIES.map(|category| match overrides.remove(category_name(category)) {
459 None => shipped_default(category),
460 Some(None) => StatusTarget::Disabled,
461 Some(Some(StatusTargetConfig::Column(option))) => StatusTarget::Column(option),
462 Some(Some(StatusTargetConfig::Closed { closed })) => StatusTarget::Closed(closed),
463 });
464 let mapping = Self { targets };
465 for (index, category) in CATEGORIES.into_iter().enumerate() {
466 let StatusTarget::Column(option) = mapping.target(category) else {
467 continue;
468 };
469 if let Some(other) = CATEGORIES[..index].iter().find(|earlier| {
470 matches!(mapping.target(**earlier), StatusTarget::Column(name)
471 if name.as_str().eq_ignore_ascii_case(option.as_str()))
472 }) {
473 return Err(SourceError::Config {
474 message: format!(
475 "status_mapping of source {instance} sends both {} and {} to the board \
476 option {:?}; one option cannot read back as two categories",
477 category_name(*other),
478 category_name(category),
479 option.as_str()
480 ),
481 });
482 }
483 }
484 Ok(mapping)
485 }
486
487 fn target(&self, category: StatusCategory) -> &StatusTarget {
488 &self.targets[category_position(category)]
489 }
490
491 fn category_of(&self, option: &str) -> Option<StatusCategory> {
493 CATEGORIES.into_iter().find(|category| {
494 matches!(self.target(*category), StatusTarget::Column(name)
495 if name.as_str().eq_ignore_ascii_case(option))
496 })
497 }
498}
499
500#[derive(Debug, Clone)]
502struct RepositoryTarget {
503 owner: String, name: String, }
506
507impl RepositoryTarget {
508 fn parse(value: &str) -> Result<Self, SourceError> {
509 let (owner, name) = value.split_once('/').ok_or_else(|| SourceError::Config {
510 message: format!(
511 "repository must be spelled owner/name; {value:?} names no repository"
512 ),
513 })?;
514 if !valid_github_owner(owner) || !valid_github_repository_name(name) {
515 return Err(SourceError::Config {
516 message: format!(
517 "repository must be spelled owner/name with a GitHub login and one \
518 repository name; {value:?} is not"
519 ),
520 });
521 }
522 Ok(Self {
523 owner: owner.to_owned(),
524 name: name.to_owned(),
525 })
526 }
527
528 fn origin(&self) -> String {
529 format!("github.com/{}/{}", self.owner, self.name)
530 }
531}
532
533pub struct GitHubProjectsSource {
535 name: SourceName,
539 owner: String, project_number: u32, repository: Option<RepositoryTarget>,
542 endpoint: Url,
543 token: SecretString,
544 credential_name: String, statuses: StatusMapping,
546 client: Client,
547 created: Mutex<Vec<Resolved>>,
561}
562
563impl GitHubProjectsSource {
564 pub fn new(
571 name: &SourceName,
572 config: GitHubProjectsConfig,
573 secrets: &dyn SecretResolver,
574 ) -> Result<Self, SourceError> {
575 if !valid_github_owner(&config.owner) {
576 return Err(SourceError::Config {
577 message: "owner must be 1-39 ASCII letters, digits, or single hyphens, and cannot start or end with a hyphen".into(),
578 });
579 }
580 if config.project_number == 0 || config.project_number > i32::MAX as u32 {
581 return Err(SourceError::Config {
582 message: format!("project_number must be between 1 and {}", i32::MAX),
583 });
584 }
585 if !valid_environment_name(&config.token_env) {
586 return Err(SourceError::Config {
587 message: "token_env must be a valid environment-variable name".into(),
588 });
589 }
590 let repository = config
591 .repository
592 .as_deref()
593 .map(RepositoryTarget::parse)
594 .transpose()?;
595 let endpoint = Url::parse(&config.endpoint).map_err(|e| SourceError::Config {
596 message: format!("endpoint is not a valid URL: {e}"),
597 })?;
598 if endpoint.scheme() != "https"
599 && !(endpoint.scheme() == "http"
600 && endpoint
601 .host_str()
602 .is_some_and(|h| h == "127.0.0.1" || h == "localhost" || h == "::1"))
603 {
604 return Err(SourceError::Config {
605 message:
606 "endpoint must use HTTPS (HTTP is accepted only for a loopback test server)"
607 .into(),
608 });
609 }
610 let token = secrets.get(&config.token_env).filter(|token| !token.expose_secret().trim().is_empty()).ok_or_else(|| SourceError::Auth {
611 message: format!("environment variable {} is missing or empty; set it to a fine-grained GitHub token granting Projects and Issues read/write plus Pull requests read-only access for every repository represented on the board", config.token_env),
612 })?;
613 Ok(Self {
614 name: name.clone(),
615 owner: config.owner,
616 project_number: config.project_number,
617 repository,
618 endpoint,
619 token,
620 credential_name: config.token_env,
621 statuses: StatusMapping::resolve(config.status_mapping, name)?,
622 client: Client::builder()
623 .user_agent("onetaskgraph")
624 .build()
625 .map_err(|e| SourceError::Config {
626 message: format!("cannot build HTTP client: {e}"),
627 })?,
628 created: Mutex::new(Vec::new()),
629 })
630 }
631
632 async fn graphql(&self, query: &str, variables: Value) -> Result<Value, SourceError> {
633 let response = self
634 .client
635 .post(self.endpoint.clone())
636 .bearer_auth(self.token.expose_secret())
637 .json(&json!({"query": query, "variables": variables}))
638 .send()
639 .await
640 .map_err(|e| SourceError::Unavailable {
641 message: format!("GitHub GraphQL request failed: {e}"),
642 })?;
643 let status = response.status();
644 let retry_after = response
645 .headers()
646 .get("retry-after")
647 .and_then(|v| v.to_str().ok())
648 .and_then(|v| v.parse().ok());
649 let exhausted = response
650 .headers()
651 .get("x-ratelimit-remaining")
652 .and_then(|v| v.to_str().ok())
653 == Some("0");
654 if status == StatusCode::TOO_MANY_REQUESTS || exhausted {
655 return Err(SourceError::RateLimited {
656 retry_after_seconds: retry_after,
657 });
658 }
659 if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN {
660 return Err(SourceError::Auth {
661 message: format!(
662 "GitHub rejected the configured credential with HTTP {status}; grant it Projects and Issues read/write plus Pull requests read-only access for every repository represented on the board"
663 ),
664 });
665 }
666 if !status.is_success() {
667 return Err(SourceError::Unavailable {
668 message: format!("GitHub GraphQL returned HTTP {status}"),
669 });
670 }
671 let body: Value = response.json().await.map_err(|e| SourceError::Malformed {
672 message: format!("GitHub returned invalid JSON: {e}"),
673 })?;
674 let errors = body
675 .get("errors")
676 .map(|value| {
677 value.as_array().ok_or_else(|| SourceError::Malformed {
678 message: "GitHub response errors is not an array".into(),
679 })
680 })
681 .transpose()?;
682 if let Some(errors) = errors.filter(|errors| !errors.is_empty()) {
683 let messages = errors
684 .iter()
685 .filter_map(|e| e.get("message").and_then(Value::as_str))
686 .collect::<Vec<_>>()
687 .join("; ");
688 let message = if messages.is_empty() {
689 "GitHub returned GraphQL errors".into()
690 } else {
691 messages
692 };
693 let normalized = message.to_ascii_lowercase();
694 if normalized.contains("resource not accessible") || normalized.contains("scope") {
695 return Err(SourceError::Auth {
696 message: format!(
697 "{message}; grant {} Projects and Issues read/write plus Pull requests read-only access for every repository represented on the board",
698 self.credential_name
699 ),
700 });
701 }
702 return Err(SourceError::Refused { message });
703 }
704 body.get("data")
705 .filter(|data| data.is_object())
706 .cloned()
707 .ok_or_else(|| SourceError::Malformed {
708 message: "GitHub response has no data object".into(),
709 })
710 }
711
712 async fn board_page(
716 &self,
717 items_after: Option<&str>,
718 items_first: u32,
719 ) -> Result<Value, SourceError> {
720 let data = self
721 .graphql(
722 graphql::BOARD,
723 json!({"owner":self.owner,"number":self.project_number,
724 "first":items_first.min(MAX_PAGE_SIZE),"after":items_after,
725 "nestedFirst":NESTED_PAGE_SIZE,"duplicates":true}),
726 )
727 .await?;
728 data.pointer("/owner/projectV2")
729 .filter(|v| !v.is_null())
730 .cloned()
731 .ok_or_else(|| SourceError::Refused {
732 message: format!(
733 "GitHub project {}/{} was not found or is not visible to the token",
734 self.owner, self.project_number
735 ),
736 })
737 }
738
739 async fn board(&self) -> Result<Board, SourceError> {
741 let mut after: Option<String> = None;
742 let mut items = Vec::new();
743 let mut board;
744 loop {
745 let page = self.board_page(after.as_deref(), MAX_PAGE_SIZE).await?;
746 for item in page
747 .pointer("/items/nodes")
748 .and_then(Value::as_array)
749 .ok_or_else(|| SourceError::Malformed {
750 message: "GitHub project items.nodes is not an array".into(),
751 })?
752 {
753 if let Some(resolved) = self.resolve(item)? {
754 items.push(resolved);
755 }
756 }
757 let info = page
758 .pointer("/items/pageInfo")
759 .ok_or_else(|| SourceError::Malformed {
760 message: "GitHub project items have no pageInfo".into(),
761 })?;
762 let has_next = required_bool(info, "hasNextPage")?;
763 let next = has_next
764 .then(|| required_str(info, "endCursor"))
765 .transpose()?;
766 board = page.clone();
767 match next {
768 Some(next) => {
769 validate_cursor_progress(after.as_deref(), next)?;
770 after = Some(next.to_owned());
771 }
772 None => break,
773 }
774 }
775 for own in self.created()?.iter() {
776 if !items.iter().any(|item| item.id == own.id) {
777 items.push(own.clone());
778 }
779 }
780 Ok(Board {
781 id: required_str(&board, "id")?.to_owned(),
782 fields: board.get("fields").cloned().unwrap_or(Value::Null),
783 items,
784 })
785 }
786
787 fn created(&self) -> Result<std::sync::MutexGuard<'_, Vec<Resolved>>, SourceError> {
789 self.created.lock().map_err(|_| SourceError::Unavailable {
790 message: "this source's record of what it created in this run was left \
791 inconsistent by an earlier failure; next: run the command again"
792 .into(),
793 })
794 }
795
796 fn resolve(&self, item: &Value) -> Result<Option<Resolved>, SourceError> {
802 let content = item.get("content").ok_or_else(|| SourceError::Malformed {
803 message: "GitHub project item is missing content".into(),
804 })?;
805 if content.is_null() {
806 return Ok(None);
807 }
808 let content_kind = match required_str(content, "__typename")? {
809 "Issue" => ContentKind::Issue,
810 "DraftIssue" => ContentKind::DraftIssue,
811 _ => return Ok(None),
812 };
813 let field_values = item
814 .get("fieldValues")
815 .ok_or_else(|| SourceError::Malformed {
816 message: "GitHub project item is missing fieldValues".into(),
817 })?;
818 complete_connection(field_values, "project item field values")?;
819 let nodes = field_values
820 .get("nodes")
821 .and_then(Value::as_array)
822 .ok_or_else(|| SourceError::Malformed {
823 message: "GitHub project item fieldValues.nodes is not an array".into(),
824 })?;
825 if let Some(labels) = content.get("labels") {
826 complete_connection(labels, "content labels")?;
827 }
828 for field_value in nodes {
829 if let Some(labels) = field_value.get("labels") {
830 complete_connection(labels, "project item field labels")?;
831 }
832 }
833 let (body, slot) = metadata_body(optional_str(content, "body")?.map(str::to_owned))?;
834 let parent = optional_str(content.get("parent").unwrap_or(&Value::Null), "id")?
835 .map(|id| NativeId(id.to_owned()));
836 let sub_issues = match content_kind {
839 ContentKind::Issue => sub_issue_total(content)?,
840 ContentKind::DraftIssue => 0,
841 };
842 let content_id = required_str(content, "id")?;
843 let marked = ItemKind::from_metadata(&slot).map_err(|message| SourceError::Malformed {
844 message: format!("GitHub issue {content_id}: {message}"),
845 })?;
846 let kind = if parent.is_some() {
849 ItemKind::Task
850 } else if sub_issues > 0 || marked == Some(ItemKind::Project) {
851 ItemKind::Project
852 } else {
853 ItemKind::Task
854 };
855 let own_repository = content
856 .pointer("/repository/nameWithOwner")
857 .and_then(Value::as_str)
858 .map(|origin| Repository::try_from(format!("github.com/{origin}")))
859 .transpose()
860 .map_err(|message| SourceError::Malformed { message })?;
861 let repositories = if slot.contains_key(Repository::METADATA_KEY) {
862 Repository::from_metadata(&slot)
863 .map_err(|message| SourceError::Malformed { message })?
864 } else {
865 own_repository.clone().into_iter().collect()
866 };
867 Ok(Some(Resolved {
868 item_id: required_str(item, "id")?.to_owned(),
869 id: NativeId(content_id.to_owned()),
870 content_kind,
871 kind,
872 title: required_str(content, "title")?.to_owned(),
873 body: body.filter(|value| !value.is_empty()),
874 status: self.status(item, content)?,
875 labels: labels(content, nodes)?,
876 parent,
877 origin: text_field(nodes, ORIGIN_FIELD)?.filter(|value| !value.is_empty()),
878 url: optional_str(content, "url")?.map(str::to_owned),
879 created_at: optional_time(content, "createdAt")?,
880 updated_at: optional_time(content, "updatedAt")?,
881 own_repository,
882 repositories,
883 slot,
884 }))
885 }
886
887 fn status(&self, item: &Value, content: &Value) -> Result<Status, SourceError> {
897 let nodes = item
898 .pointer("/fieldValues/nodes")
899 .and_then(Value::as_array)
900 .expect("resolve validates fieldValues.nodes before mapping status");
901 let option = nodes
902 .iter()
903 .find(|value| value.pointer("/field/name").and_then(Value::as_str) == Some("Status"))
904 .map(|value| required_str(value, "name"))
905 .transpose()?;
906 let state = optional_str(content, "state")?;
907 if state == Some("CLOSED") {
908 let category = match optional_str(content, "stateReason")? {
909 None | Some("COMPLETED") => StatusCategory::Done,
910 Some("NOT_PLANNED") => StatusCategory::Cancelled,
911 Some(_) => StatusCategory::Unknown,
912 };
913 let fallback = match category {
914 StatusCategory::Done => "Done",
915 StatusCategory::Cancelled => "Cancelled",
916 _ => "Closed",
917 };
918 return Ok(Status {
919 category,
920 name: option.unwrap_or(fallback).to_owned(),
921 });
922 }
923 let name = option.unwrap_or("Open").to_owned();
924 Ok(Status {
925 category: self
926 .statuses
927 .category_of(&name)
928 .unwrap_or(StatusCategory::Unknown),
929 name,
930 })
931 }
932
933 fn column_for(
941 &self,
942 board: &Board,
943 status: &Status,
944 target: &StatusTarget,
945 ) -> Result<Option<(String, String)>, SourceError> {
946 let (wanted, required) = match target {
947 StatusTarget::Column(wanted) => (wanted.as_str(), true),
948 StatusTarget::Closed(_) => (status.name.as_str(), false),
949 StatusTarget::Disabled => return Ok(None),
950 };
951 let missing = |detail: &str| SourceError::Refused {
952 message: format!(
953 "status {} of source {} needs the board Status option {wanted:?}, and {detail}; add that option to the board, or point status_mapping.{} of this source at one it has",
954 category_name(status.category),
955 self.name,
956 category_name(status.category)
957 ),
958 };
959 let Some(field) = Board::field(&board.fields, "Status")? else {
960 return if required {
961 Err(missing("this board has no Status field"))
962 } else {
963 Ok(None)
964 };
965 };
966 if required_str(field, "__typename")? != "ProjectV2SingleSelectField" {
967 return if required {
968 Err(missing(
969 "this board's Status field is not a single-select field",
970 ))
971 } else {
972 Ok(None)
973 };
974 }
975 let option = field
976 .get("options")
977 .and_then(Value::as_array)
978 .and_then(|options| {
979 options.iter().find(|option| {
980 option
981 .get("name")
982 .and_then(Value::as_str)
983 .is_some_and(|name| name.eq_ignore_ascii_case(wanted))
984 })
985 });
986 match option {
987 None if required => Err(missing("this board does not have it")),
988 None => Ok(None),
989 Some(option) => Ok(Some((
990 required_str(field, "id")?.to_owned(),
991 required_str(option, "id")?.to_owned(),
992 ))),
993 }
994 }
995
996 fn resolved_target(&self, category: StatusCategory) -> Result<StatusTarget, SourceError> {
1003 let target = self.statuses.target(category).clone();
1004 if target != StatusTarget::Disabled {
1005 return Ok(target);
1006 }
1007 Err(SourceError::Refused {
1008 message: if category == StatusCategory::Draft {
1009 format!(
1010 "status draft is disabled for source {}: draft is incompatible with this \
1011 integration because GitHub draft issues cannot have sub-issues, and this \
1012 source stores a project's tasks as its issue's sub-issues",
1013 self.name
1014 )
1015 } else {
1016 format!(
1017 "status {} is disabled for source {}; set status_mapping.{} of this source \
1018 to a board Status option name or to a closed state",
1019 category_name(category),
1020 self.name,
1021 category_name(category)
1022 )
1023 },
1024 })
1025 }
1026
1027 async fn set_item_field(
1028 &self,
1029 board_id: &str,
1030 item_id: &str,
1031 field_id: &str,
1032 value: Value,
1033 ) -> Result<(), SourceError> {
1034 let data = self
1035 .graphql(
1036 graphql::UPDATE_FIELD,
1037 json!({"input":{
1038 "projectId":board_id,"itemId":item_id,"fieldId":field_id,"value":value
1039 }}),
1040 )
1041 .await?;
1042 let returned = data
1043 .pointer("/updateProjectV2ItemFieldValue/projectV2Item")
1044 .ok_or_else(|| SourceError::Malformed {
1045 message: "GitHub field update returned no project item".into(),
1046 })?;
1047 if required_str(returned, "id")? != item_id {
1048 return Err(SourceError::Malformed {
1049 message: "GitHub field update returned the wrong project item".into(),
1050 });
1051 }
1052 Ok(())
1053 }
1054
1055 async fn native_dependency_ids(&self, id: &NativeId) -> Result<Vec<String>, SourceError> {
1056 let mut after: Option<String> = None;
1057 let mut ids = Vec::new();
1058 loop {
1059 let data = self
1060 .graphql(
1061 graphql::ISSUE_DEPENDENCIES,
1062 json!({"id":id.0,"first":MAX_PAGE_SIZE,"after":after}),
1063 )
1064 .await?;
1065 let connection =
1066 data.pointer("/node/blockedBy")
1067 .ok_or_else(|| SourceError::Malformed {
1068 message: "GitHub dependency response has no blockedBy connection".into(),
1069 })?;
1070 ids.extend(
1071 connection
1072 .get("nodes")
1073 .and_then(Value::as_array)
1074 .ok_or_else(|| SourceError::Malformed {
1075 message: "GitHub dependency response nodes is not an array".into(),
1076 })?
1077 .iter()
1078 .map(|value| required_str(value, "id").map(str::to_owned))
1079 .collect::<Result<Vec<_>, _>>()?,
1080 );
1081 let next = next_cursor(connection)?;
1082 if let Some(next) = &next {
1083 validate_cursor_progress(after.as_deref(), &next.0)?;
1084 }
1085 after = next.map(|cursor| cursor.0);
1086 if after.is_none() {
1087 return Ok(ids);
1088 }
1089 }
1090 }
1091
1092 async fn dependencies(
1093 &self,
1094 id: &NativeId,
1095 near_kind: ItemKind,
1096 direction: Direction,
1097 page: &PageRequest,
1098 ) -> Result<Page<DependencyEdge>, SourceError> {
1099 validate_page(page)?;
1100 let limit = page.limit.min(MAX_PAGE_SIZE) as usize;
1101 let cursor = page.cursor.as_ref().map(|c| c.0.as_str());
1102 let recorded = recorded_offset(cursor, direction)?;
1103 let data = self
1108 .graphql(
1109 graphql::ISSUE_DEPENDENCIES,
1110 json!({"id":id.0,"first":page.limit.min(MAX_PAGE_SIZE),
1111 "after":if recorded.is_some() {None} else {cursor}}),
1112 )
1113 .await?;
1114 let node =
1115 data.get("node")
1116 .filter(|v| !v.is_null())
1117 .ok_or_else(|| SourceError::Refused {
1118 message: format!(
1119 "GitHub item {} was not found or does not support dependencies",
1120 id.0
1121 ),
1122 })?;
1123 let connection_name = match direction {
1124 Direction::DependsOn => "blockedBy",
1125 Direction::DependedOnBy => "blocking",
1126 };
1127 let natively_names = (required_str(node, "__typename")? == "Issue").then_some(near_kind);
1131 if let Some(offset) = recorded {
1132 return Ok(recorded_page(
1133 self.recorded_edges(id, near_kind, direction, natively_names)
1134 .await?,
1135 offset,
1136 limit,
1137 ));
1138 }
1139 if natively_names.is_none() {
1140 return Ok(recorded_page(
1141 self.recorded_edges(id, near_kind, direction, natively_names)
1142 .await?,
1143 0,
1144 limit,
1145 ));
1146 }
1147 let connection = node
1148 .get(connection_name)
1149 .ok_or_else(|| SourceError::Malformed {
1150 message: "GitHub dependency response is missing its connection".into(),
1151 })?;
1152 let nodes = connection
1153 .get("nodes")
1154 .and_then(Value::as_array)
1155 .ok_or_else(|| SourceError::Malformed {
1156 message: "GitHub dependency response nodes is not an array".into(),
1157 })?;
1158 let items = nodes
1162 .iter()
1163 .map(|value| {
1164 let related = NativeId(required_str(value, "id")?.into());
1165 let related_kind = related_kind(value)?;
1166 let (from, to) = match direction {
1167 Direction::DependsOn => (
1168 DependencyEndpoint::from_native(id.clone(), near_kind),
1169 DependencyEndpoint::from_native(related, related_kind),
1170 ),
1171 Direction::DependedOnBy => (
1172 DependencyEndpoint::from_native(related, related_kind),
1173 DependencyEndpoint::from_native(id.clone(), near_kind),
1174 ),
1175 };
1176 Ok(DependencyEdge {
1177 from,
1178 to,
1179 kind: DependencyKind::Blocks,
1180 })
1181 })
1182 .collect::<Result<Vec<_>, SourceError>>()?;
1183 let mut next = next_cursor(connection)?;
1184 if let Some(next) = &next {
1185 validate_cursor_progress(cursor, &next.0)?;
1186 }
1187 if next.is_none()
1188 && !self
1189 .recorded_edges(id, near_kind, direction, natively_names)
1190 .await?
1191 .is_empty()
1192 {
1193 next = Some(Cursor(format!("{RECORDED_CURSOR}0")));
1194 }
1195 Ok(Page { items, next })
1196 }
1197
1198 async fn recorded_edges(
1208 &self,
1209 id: &NativeId,
1210 near_kind: ItemKind,
1211 direction: Direction,
1212 natively_names: Option<ItemKind>,
1213 ) -> Result<Vec<DependencyEdge>, SourceError> {
1214 if direction != Direction::DependsOn {
1215 return Ok(Vec::new());
1216 }
1217 let Some(item) = self
1218 .board()
1219 .await?
1220 .items
1221 .into_iter()
1222 .find(|item| item.id == *id)
1223 else {
1224 return Ok(Vec::new());
1225 };
1226 DependencyEdge::recorded(&item.slot, id, near_kind, &self.name, natively_names)
1227 .map_err(|message| SourceError::Malformed { message })
1228 }
1229
1230 async fn repository_id(&self) -> Result<String, SourceError> {
1232 let repository = self
1233 .repository
1234 .as_ref()
1235 .ok_or_else(|| SourceError::Refused {
1236 message: format!(
1237 "source {} has no repository configured, and a GitHub Projects board has no \
1238 repository of its own to create an issue in; set repository: owner/name on \
1239 this source",
1240 self.name
1241 ),
1242 })?;
1243 let data = self
1244 .graphql(
1245 graphql::REPOSITORY,
1246 json!({"owner":repository.owner,"name":repository.name}),
1247 )
1248 .await?;
1249 let node = data
1250 .get("repository")
1251 .filter(|value| !value.is_null())
1252 .ok_or_else(|| SourceError::Refused {
1253 message: format!(
1254 "GitHub repository {}/{} was not found or is not visible to the token",
1255 repository.owner, repository.name
1256 ),
1257 })?;
1258 Ok(required_str(node, "id")?.to_owned())
1259 }
1260
1261 async fn write_item(
1263 &self,
1264 incoming: &Incoming<'_>,
1265 target: Option<&NativeId>,
1266 depends_on: &[DependencyEdge],
1267 ) -> Result<NativeId, SourceError> {
1268 let board = self.board().await?;
1269 let status_target = self.resolved_target(incoming.status.category)?;
1270 let column = self.column_for(&board, incoming.status, &status_target)?;
1271 let existing = target
1272 .map(|target| {
1273 board
1274 .items
1275 .iter()
1276 .find(|item| item.id == *target)
1277 .ok_or_else(|| SourceError::Refused {
1278 message: format!("GitHub destination item {} was not found", target.0),
1279 })
1280 })
1281 .transpose()?;
1282 let content_kind = existing.map_or(ContentKind::Issue, |item| item.content_kind);
1283 if content_kind == ContentKind::DraftIssue {
1284 if let StatusTarget::Closed(_) = status_target {
1285 return Err(SourceError::Refused {
1286 message: format!(
1287 "status {} of source {} closes the item's issue, and GitHub draft items \
1288 have no open or closed state",
1289 category_name(incoming.status.category),
1290 self.name
1291 ),
1292 });
1293 }
1294 if incoming.parent.is_some() {
1295 return Err(SourceError::Refused {
1296 message: "GitHub draft items cannot be a project's sub-issue".into(),
1297 });
1298 }
1299 }
1300 match existing {
1301 Some(item) if content_kind == ContentKind::Issue => {
1302 if item.labels != incoming.labels {
1303 return Err(SourceError::Refused {
1304 message: "GitHub issue labels differ from the labels being written".into(),
1305 });
1306 }
1307 }
1308 _ => {
1309 if !incoming.labels.is_empty() {
1310 return Err(SourceError::Refused {
1311 message: "GitHub items created by this destination carry no labels".into(),
1312 });
1313 }
1314 }
1315 }
1316
1317 let own_repository = match existing {
1318 Some(item) => item.own_repository.clone(),
1319 None => self
1320 .repository
1321 .as_ref()
1322 .map(|repository| Repository::try_from(repository.origin()))
1323 .transpose()
1324 .map_err(|message| SourceError::Config { message })?,
1325 };
1326 let (native, fallback) = self
1327 .partition_edges(&board, incoming.kind, content_kind, depends_on)
1328 .await?;
1329 let slot = slot_metadata(incoming, own_repository.as_ref(), &fallback);
1330 let body = compose_body(incoming.content, &slot)?;
1331 let origin = match incoming.metadata.get(ORIGIN_KEY) {
1338 None => "",
1339 Some(Value::String(origin)) => origin.as_str(),
1340 Some(other) => {
1341 return Err(SourceError::Refused {
1342 message: format!(
1343 "{ORIGIN_KEY} holds a qualified id spelled as a string, and this item's \
1344 is {other}"
1345 ),
1346 });
1347 }
1348 };
1349 let origin_field = match Board::field(&board.fields, ORIGIN_FIELD)? {
1353 Some(field) => {
1354 if required_str(field, "__typename")? != "ProjectV2Field" {
1355 return Err(SourceError::Refused {
1356 message: format!(
1357 "GitHub board source-owned {ORIGIN_FIELD} field is not a text field"
1358 ),
1359 });
1360 }
1361 Some(required_str(field, "id")?.to_owned())
1362 }
1363 None if incoming.metadata.contains_key(ORIGIN_KEY) => {
1364 return Err(SourceError::Refused {
1365 message: format!(
1366 "GitHub board has no source-owned {ORIGIN_FIELD} text field, and the \
1367 item carries {ORIGIN_KEY}; add a text field named {ORIGIN_FIELD} to \
1368 the board"
1369 ),
1370 });
1371 }
1372 None => None,
1373 };
1374
1375 let (content_id, item_id) = match existing {
1376 Some(item) => {
1377 self.update_existing(item, incoming, &body, &status_target)
1378 .await?;
1379 (item.id.clone(), item.item_id.clone())
1380 }
1381 None => {
1382 self.create_and_file_issue(&board, incoming, &body, &status_target)
1383 .await?
1384 }
1385 };
1386
1387 let landed = self
1395 .finish_write(
1396 &board,
1397 incoming,
1398 &content_id,
1399 &item_id,
1400 content_kind,
1401 existing,
1402 origin_field.as_deref(),
1403 origin,
1404 column,
1405 &native,
1406 )
1407 .await;
1408 if let Err(error) = landed {
1409 if existing.is_none() {
1410 let _ = self.delete_issue(&content_id).await;
1413 }
1414 return Err(error);
1415 }
1416
1417 if existing.is_none() {
1418 let remembered = Resolved {
1421 item_id,
1422 id: content_id.clone(),
1423 content_kind,
1424 kind: incoming.kind,
1425 title: incoming.title.to_owned(),
1426 body: body.clone(),
1427 status: incoming.status.clone(),
1428 labels: incoming.labels.to_vec(),
1429 parent: incoming.parent.cloned(),
1430 origin: (!origin.is_empty()).then(|| origin.to_owned()),
1431 url: None,
1432 created_at: None,
1433 updated_at: None,
1434 own_repository,
1435 repositories: incoming.repositories.to_vec(),
1436 slot,
1437 };
1438 self.created()?.push(remembered);
1439 }
1440 Ok(content_id)
1441 }
1442
1443 #[allow(clippy::too_many_arguments)]
1454 async fn finish_write(
1455 &self,
1456 board: &Board,
1457 incoming: &Incoming<'_>,
1458 content_id: &NativeId,
1459 item_id: &str,
1460 content_kind: ContentKind,
1461 existing: Option<&Resolved>,
1462 origin_field: Option<&str>,
1463 origin: &str,
1464 column: Option<(String, String)>,
1465 native: &[String],
1466 ) -> Result<(), SourceError> {
1467 if let Some(field_id) = origin_field {
1468 self.set_item_field(&board.id, item_id, field_id, json!({"text":origin}))
1469 .await?;
1470 }
1471
1472 if let Some((field_id, option_id)) = column {
1473 self.set_item_field(
1474 &board.id,
1475 item_id,
1476 &field_id,
1477 json!({"singleSelectOptionId":option_id}),
1478 )
1479 .await?;
1480 }
1481
1482 if content_kind == ContentKind::Issue {
1483 self.reparent(
1484 existing.and_then(|item| item.parent.clone()),
1485 content_id,
1486 incoming.parent,
1487 )
1488 .await?;
1489 self.reconcile_blocked_by(content_id, native).await?;
1490 }
1491 Ok(())
1492 }
1493
1494 async fn delete_issue(&self, id: &NativeId) -> Result<(), SourceError> {
1496 let data = self
1497 .graphql(graphql::DELETE_ISSUE, json!({"input":{"issueId":id.0}}))
1498 .await?;
1499 data.pointer("/deleteIssue/repository")
1500 .filter(|value| !value.is_null())
1501 .ok_or_else(|| SourceError::Malformed {
1502 message: "GitHub issue deletion returned no repository".into(),
1503 })?;
1504 self.created()?.retain(|own| own.id != *id);
1505 Ok(())
1506 }
1507
1508 async fn delete_item(&self, id: &NativeId) -> Result<(), SourceError> {
1515 let board = self.board().await?;
1516 let Some(item) = board.items.iter().find(|item| item.id == *id) else {
1517 return Ok(());
1518 };
1519 if item.content_kind == ContentKind::DraftIssue {
1520 return Err(SourceError::Refused {
1521 message: format!(
1522 "GitHub item {} is a draft, and this source removes an item by deleting \
1523 its issue; next: remove it from the board by hand",
1524 id.0
1525 ),
1526 });
1527 }
1528 let data = self
1529 .graphql(graphql::DELETE_ISSUE, json!({"input":{"issueId":id.0}}))
1530 .await?;
1531 data.pointer("/deleteIssue/repository")
1532 .filter(|value| !value.is_null())
1533 .ok_or_else(|| SourceError::Malformed {
1534 message: "GitHub issue deletion returned no repository".into(),
1535 })?;
1536 self.created()?.retain(|own| own.id != *id);
1537 Ok(())
1538 }
1539
1540 async fn partition_edges(
1542 &self,
1543 board: &Board,
1544 near_kind: ItemKind,
1545 near_content: ContentKind,
1546 depends_on: &[DependencyEdge],
1547 ) -> Result<(Vec<String>, Vec<DependencyEdge>), SourceError> {
1548 let mut native = Vec::new();
1549 let mut fallback = Vec::new();
1550 for edge in depends_on {
1551 let same_source = edge
1552 .to
1553 .source()
1554 .is_none_or(|source| source == self.name.as_str());
1555 let far_id = if edge.to.is_qualified() {
1560 edge.to
1561 .id()
1562 .split_once(':')
1563 .map_or(edge.to.id(), |(_, native)| native)
1564 } else {
1565 edge.to.id()
1566 };
1567 let far = if same_source {
1568 Some(
1569 board
1570 .items
1571 .iter()
1572 .find(|item| item.id.0 == far_id)
1573 .ok_or_else(|| SourceError::Refused {
1574 message: format!("GitHub dependency item {far_id} was not found"),
1575 })?,
1576 )
1577 } else {
1578 None
1579 };
1580 if let Some(disagreeing) = far.filter(|far| far.kind != edge.to.kind) {
1586 return Err(SourceError::Refused {
1587 message: format!(
1588 "GitHub dependency item {far_id} is a {} of this board, and this item \
1589 names it as a {}; record the kind it is",
1590 disagreeing.kind.marker(),
1591 edge.to.kind.marker()
1592 ),
1593 });
1594 }
1595 let native_here = near_content == ContentKind::Issue
1599 && far.is_some_and(|far| {
1600 far.content_kind == ContentKind::Issue && edge.to.kind == near_kind
1601 });
1602 if native_here {
1603 native.push(far_id.to_owned());
1604 } else {
1605 fallback.push(edge.clone());
1606 }
1607 }
1608 Ok((native, fallback))
1609 }
1610
1611 async fn update_existing(
1612 &self,
1613 item: &Resolved,
1614 incoming: &Incoming<'_>,
1615 body: &Option<String>,
1616 status_target: &StatusTarget,
1617 ) -> Result<(), SourceError> {
1618 let (operation, input, pointer) = match item.content_kind {
1619 ContentKind::DraftIssue => (
1620 graphql::UPDATE_DRAFT,
1621 json!({"draftIssueId":item.id.0,"title":incoming.title,"body":body}),
1622 "/updateProjectV2DraftIssue/draftIssue",
1623 ),
1624 ContentKind::Issue => (
1625 graphql::UPDATE_ISSUE,
1626 json!({"id":item.id.0,"title":incoming.title,"body":body,
1627 "stateInput":state_input(status_target)}),
1628 "/updateIssue/issue",
1629 ),
1630 };
1631 let data = self.graphql(operation, json!({"input":input})).await?;
1632 let returned = data
1633 .pointer(pointer)
1634 .ok_or_else(|| SourceError::Malformed {
1635 message: "GitHub item update returned no item".into(),
1636 })?;
1637 if required_str(returned, "id")? != item.id.0 {
1638 return Err(SourceError::Malformed {
1639 message: "GitHub item update returned the wrong item".into(),
1640 });
1641 }
1642 Ok(())
1643 }
1644
1645 async fn create_and_file_issue(
1651 &self,
1652 board: &Board,
1653 incoming: &Incoming<'_>,
1654 body: &Option<String>,
1655 status_target: &StatusTarget,
1656 ) -> Result<(NativeId, String), SourceError> {
1657 let repository_id = self.repository_id().await?;
1658 let data = self
1659 .graphql(
1660 graphql::CREATE_ISSUE,
1661 json!({"input":{
1662 "repositoryId":repository_id,"title":incoming.title,"body":body
1663 }}),
1664 )
1665 .await?;
1666 let created = data
1667 .pointer("/createIssue/issue")
1668 .filter(|value| !value.is_null())
1669 .ok_or_else(|| SourceError::Malformed {
1670 message: "GitHub issue creation returned no issue".into(),
1671 })?;
1672 let content_id = NativeId(required_str(created, "id")?.to_owned());
1673 let added = match self
1677 .graphql(
1678 graphql::ADD_TO_BOARD,
1679 json!({"input":{"projectId":board.id,"contentId":content_id.0}}),
1680 )
1681 .await
1682 {
1683 Ok(added) => added,
1684 Err(error) => {
1685 let _ = self.delete_issue(&content_id).await;
1686 return Err(error);
1687 }
1688 };
1689 let item = added
1690 .pointer("/addProjectV2ItemById/item")
1691 .filter(|value| !value.is_null())
1692 .ok_or_else(|| SourceError::Malformed {
1693 message: "GitHub board addition returned no project item".into(),
1694 })?;
1695 if let StatusTarget::Closed(_) = status_target {
1696 let closed = self
1697 .graphql(
1698 graphql::UPDATE_ISSUE,
1699 json!({"input":{"id":content_id.0,"stateInput":state_input(status_target)}}),
1700 )
1701 .await?;
1702 let returned =
1703 closed
1704 .pointer("/updateIssue/issue")
1705 .ok_or_else(|| SourceError::Malformed {
1706 message: "GitHub item update returned no item".into(),
1707 })?;
1708 if required_str(returned, "id")? != content_id.0 {
1709 return Err(SourceError::Malformed {
1710 message: "GitHub item update returned the wrong item".into(),
1711 });
1712 }
1713 }
1714 Ok((content_id, required_str(item, "id")?.to_owned()))
1715 }
1716
1717 async fn reparent(
1719 &self,
1720 held: Option<NativeId>,
1721 child: &NativeId,
1722 wanted: Option<&NativeId>,
1723 ) -> Result<(), SourceError> {
1724 if held.as_ref() == wanted {
1725 return Ok(());
1726 }
1727 if let Some(held) = &held {
1728 self.sub_issue(graphql::REMOVE_SUB_ISSUE, held, child, "removeSubIssue")
1729 .await?;
1730 }
1731 if let Some(wanted) = wanted {
1732 self.sub_issue(graphql::ADD_SUB_ISSUE, wanted, child, "addSubIssue")
1733 .await?;
1734 }
1735 Ok(())
1736 }
1737
1738 async fn sub_issue(
1739 &self,
1740 operation: &str,
1741 parent: &NativeId,
1742 child: &NativeId,
1743 root: &str,
1744 ) -> Result<(), SourceError> {
1745 let data = self
1746 .graphql(
1747 operation,
1748 json!({"input":{"issueId":parent.0,"subIssueId":child.0}}),
1749 )
1750 .await?;
1751 let issue =
1752 data.pointer(&format!("/{root}/issue"))
1753 .ok_or_else(|| SourceError::Malformed {
1754 message: "GitHub sub-issue update returned no issue".into(),
1755 })?;
1756 let sub =
1757 data.pointer(&format!("/{root}/subIssue"))
1758 .ok_or_else(|| SourceError::Malformed {
1759 message: "GitHub sub-issue update returned no sub-issue".into(),
1760 })?;
1761 if required_str(issue, "id")? != parent.0 || required_str(sub, "id")? != child.0 {
1762 return Err(SourceError::Malformed {
1763 message: "GitHub sub-issue update returned the wrong issues".into(),
1764 });
1765 }
1766 Ok(())
1767 }
1768
1769 async fn reconcile_blocked_by(
1770 &self,
1771 content_id: &NativeId,
1772 native: &[String],
1773 ) -> Result<(), SourceError> {
1774 let current = self.native_dependency_ids(content_id).await?;
1775 for (operation, far_id) in current
1776 .iter()
1777 .filter(|id| !native.contains(id))
1778 .map(|id| (graphql::REMOVE_BLOCKED_BY, id))
1779 .chain(
1780 native
1781 .iter()
1782 .filter(|id| !current.contains(id))
1783 .map(|id| (graphql::ADD_BLOCKED_BY, id)),
1784 )
1785 {
1786 let data = self
1787 .graphql(
1788 operation,
1789 json!({"input":{"issueId":content_id.0,"blockingIssueId":far_id}}),
1790 )
1791 .await?;
1792 let root = if operation == graphql::ADD_BLOCKED_BY {
1793 "addBlockedBy"
1794 } else {
1795 "removeBlockedBy"
1796 };
1797 let issue =
1798 data.pointer(&format!("/{root}/issue"))
1799 .ok_or_else(|| SourceError::Malformed {
1800 message: "GitHub dependency update returned no issue".into(),
1801 })?;
1802 let blocker = data
1803 .pointer(&format!("/{root}/blockingIssue"))
1804 .ok_or_else(|| SourceError::Malformed {
1805 message: "GitHub dependency update returned no blocking issue".into(),
1806 })?;
1807 if required_str(issue, "id")? != content_id.0 || required_str(blocker, "id")? != far_id
1808 {
1809 return Err(SourceError::Malformed {
1810 message: "GitHub dependency update returned the wrong issues".into(),
1811 });
1812 }
1813 }
1814 Ok(())
1815 }
1816}
1817
1818struct Board {
1820 id: String,
1821 fields: Value,
1822 items: Vec<Resolved>,
1823}
1824
1825impl Board {
1826 fn field<'a>(fields: &'a Value, name: &str) -> Result<Option<&'a Value>, SourceError> {
1827 complete_connection(fields, "project fields")?;
1828 let nodes = fields
1829 .get("nodes")
1830 .and_then(Value::as_array)
1831 .ok_or_else(|| SourceError::Malformed {
1832 message: "GitHub project fields.nodes is not an array".into(),
1833 })?;
1834 Ok(nodes
1835 .iter()
1836 .find(|field| field.get("name").and_then(Value::as_str) == Some(name)))
1837 }
1838}
1839
1840#[derive(Clone)]
1842struct Resolved {
1843 item_id: String,
1844 id: NativeId,
1845 content_kind: ContentKind,
1846 kind: ItemKind,
1847 title: String,
1848 body: Option<String>,
1849 status: Status,
1850 labels: Vec<Label>,
1851 parent: Option<NativeId>,
1852 origin: Option<String>,
1854 url: Option<String>,
1855 created_at: Option<DateTime<Utc>>,
1856 updated_at: Option<DateTime<Utc>>,
1857 own_repository: Option<Repository>,
1858 repositories: Vec<Repository>,
1859 slot: BTreeMap<String, Value>,
1860}
1861
1862impl Resolved {
1863 fn metadata(&self) -> BTreeMap<String, Value> {
1866 let mut metadata = self.slot.clone();
1867 metadata.remove(Repository::METADATA_KEY);
1868 metadata.remove(DependencyEdge::RECORDED_KEY);
1869 metadata.remove(ItemKind::METADATA_KEY);
1870 if let Some(origin) = &self.origin {
1871 metadata.insert(ORIGIN_KEY.to_owned(), Value::String(origin.clone()));
1872 }
1873 metadata
1874 }
1875
1876 fn task(&self) -> Task {
1877 Task {
1878 id: self.id.clone(),
1879 title: self.title.clone(),
1880 content: self.body.clone(),
1881 status: self.status.clone(),
1882 labels: self.labels.clone(),
1883 project: self.parent.clone(),
1884 url: self.url.clone(),
1885 created_at: self.created_at,
1886 updated_at: self.updated_at,
1887 metadata: self.metadata(),
1888 repositories: self.repositories.clone(),
1889 }
1890 }
1891
1892 fn project(&self) -> Project {
1893 Project {
1894 id: self.id.clone(),
1895 title: self.title.clone(),
1896 content: self.body.clone(),
1897 status: self.status.clone(),
1898 labels: self.labels.clone(),
1899 url: self.url.clone(),
1900 created_at: self.created_at,
1901 updated_at: self.updated_at,
1902 metadata: self.metadata(),
1903 repositories: self.repositories.clone(),
1904 }
1905 }
1906}
1907
1908struct Incoming<'a> {
1910 kind: ItemKind,
1911 title: &'a str,
1912 content: Option<&'a str>,
1913 status: &'a Status,
1914 labels: &'a [Label],
1915 metadata: &'a BTreeMap<String, Value>,
1916 repositories: &'a [Repository],
1917 parent: Option<&'a NativeId>,
1918}
1919
1920#[derive(Clone, Copy, PartialEq, Eq)]
1921enum ContentKind {
1922 DraftIssue,
1923 Issue,
1924}
1925
1926fn labels_match(labels: &[Label], filter: &LabelFilter) -> bool {
1932 let holds = |name: &String| {
1933 labels
1934 .iter()
1935 .any(|label| label.name.eq_ignore_ascii_case(name))
1936 };
1937 (filter.any_of.is_empty() || filter.any_of.iter().any(holds))
1938 && filter.all_of.iter().all(holds)
1939 && !filter.none_of.iter().any(holds)
1940}
1941
1942fn status_matches(category: StatusCategory, statuses: &[StatusCategory]) -> bool {
1945 statuses.is_empty() || statuses.contains(&category)
1946}
1947
1948fn text_matches(title: &str, content: Option<&str>, query: &TextQuery) -> bool {
1954 let terms = query.terms.to_lowercase();
1955 let in_title = title.to_lowercase().contains(&terms);
1956 let in_content = content.is_some_and(|body| body.to_lowercase().contains(&terms));
1957 match query.fields {
1958 TextFields::Title => in_title,
1959 TextFields::Content => in_content,
1960 TextFields::TitleOrContent => in_title || in_content,
1961 }
1962}
1963
1964fn task_matches(task: &Task, query: &TaskQuery) -> bool {
1965 labels_match(&task.labels, &query.labels)
1966 && status_matches(task.status.category, &query.statuses)
1967 && match &query.project {
1968 ProjectFilter::Any => true,
1969 ProjectFilter::Orphans => task.project.is_none(),
1970 ProjectFilter::Is(id) => task.project.as_ref() == Some(id),
1971 }
1972 && query
1973 .text
1974 .as_ref()
1975 .is_none_or(|text| text_matches(&task.title, task.content.as_deref(), text))
1976}
1977
1978fn project_matches(project: &Project, query: &ProjectQuery) -> bool {
1979 labels_match(&project.labels, &query.labels)
1980 && status_matches(project.status.category, &query.statuses)
1981 && query
1982 .text
1983 .as_ref()
1984 .is_none_or(|text| text_matches(&project.title, project.content.as_deref(), text))
1985}
1986
1987#[async_trait::async_trait]
1988impl TaskSource for GitHubProjectsSource {
1989 fn kind(&self) -> &'static str {
1990 KIND
1991 }
1992 fn capabilities(&self) -> Capabilities {
1993 Capabilities {
1994 projects: Support::Native,
1995 orphan_tasks: Support::Native,
1996 filter_by_label: Support::Native,
1997 filter_by_status: Support::Native,
1998 search_title: Support::Native,
1999 search_content: Support::Native,
2000 task_dependencies: DependencySupport::BothDirections,
2001 project_dependencies: DependencySupport::BothDirections,
2002 max_page_size: MAX_PAGE_SIZE,
2003 }
2004 }
2005 async fn health(&self) -> Result<Health, SourceError> {
2006 let board = self.board_page(None, 1).await?;
2007 Ok(Health {
2008 reachable: true,
2009 detail: Some(format!(
2010 "reading GitHub project {}/{} ({})",
2011 self.owner,
2012 self.project_number,
2013 required_str(&board, "title")?
2014 )),
2015 })
2016 }
2017 async fn get_task(&self, id: &NativeId) -> Result<Option<Task>, SourceError> {
2018 Ok(self
2019 .board()
2020 .await?
2021 .items
2022 .iter()
2023 .find(|item| item.id == *id && item.kind == ItemKind::Task)
2024 .map(Resolved::task))
2025 }
2026 async fn get_project(&self, id: &NativeId) -> Result<Option<Project>, SourceError> {
2027 Ok(self
2028 .board()
2029 .await?
2030 .items
2031 .iter()
2032 .find(|item| item.id == *id && item.kind == ItemKind::Project)
2033 .map(Resolved::project))
2034 }
2035 async fn query_tasks(
2036 &self,
2037 query: &TaskQuery,
2038 page: &PageRequest,
2039 ) -> Result<Page<Task>, SourceError> {
2040 validate_page(page)?;
2041 let tasks = self
2044 .board()
2045 .await?
2046 .items
2047 .iter()
2048 .filter(|item| item.kind == ItemKind::Task)
2049 .map(Resolved::task)
2050 .filter(|task| task_matches(task, query))
2051 .collect();
2052 Ok(offset_page(
2053 tasks,
2054 numeric_cursor(page.cursor.as_ref())?,
2055 page.limit.min(MAX_PAGE_SIZE) as usize,
2056 ))
2057 }
2058 async fn query_projects(
2059 &self,
2060 query: &ProjectQuery,
2061 page: &PageRequest,
2062 ) -> Result<Page<Project>, SourceError> {
2063 validate_page(page)?;
2064 let projects = self
2065 .board()
2066 .await?
2067 .items
2068 .iter()
2069 .filter(|item| item.kind == ItemKind::Project)
2070 .map(Resolved::project)
2071 .filter(|project| project_matches(project, query))
2072 .collect();
2073 Ok(offset_page(
2074 projects,
2075 numeric_cursor(page.cursor.as_ref())?,
2076 page.limit.min(MAX_PAGE_SIZE) as usize,
2077 ))
2078 }
2079 async fn labels(&self, page: &PageRequest) -> Result<Page<Label>, SourceError> {
2080 validate_page(page)?;
2081 let offset = numeric_cursor(page.cursor.as_ref())?;
2082 let mut labels = self
2083 .board()
2084 .await?
2085 .items
2086 .into_iter()
2087 .flat_map(|item| item.labels)
2088 .fold(Vec::new(), |mut all, label| {
2089 if !all.iter().any(|x: &Label| x.id == label.id) {
2090 all.push(label);
2091 }
2092 all
2093 });
2094 labels.sort_by(|a, b| a.name.cmp(&b.name).then(a.id.0.cmp(&b.id.0)));
2095 Ok(offset_page(
2096 labels,
2097 offset,
2098 page.limit.min(MAX_PAGE_SIZE) as usize,
2099 ))
2100 }
2101 async fn task_dependencies(
2102 &self,
2103 id: &NativeId,
2104 direction: Direction,
2105 page: &PageRequest,
2106 ) -> Result<Page<DependencyEdge>, SourceError> {
2107 self.dependencies(id, ItemKind::Task, direction, page).await
2108 }
2109 async fn project_dependencies(
2110 &self,
2111 id: &NativeId,
2112 direction: Direction,
2113 page: &PageRequest,
2114 ) -> Result<Page<DependencyEdge>, SourceError> {
2115 self.dependencies(id, ItemKind::Project, direction, page)
2116 .await
2117 }
2118
2119 fn writes(&self) -> WriteSupport {
2120 WriteSupport::Supported
2121 }
2122
2123 async fn write_task(&self, write: &ItemWrite<Task>) -> Result<NativeId, SourceError> {
2124 self.write_item(
2125 &Incoming {
2126 kind: ItemKind::Task,
2127 title: &write.item.title,
2128 content: write.item.content.as_deref(),
2129 status: &write.item.status,
2130 labels: &write.item.labels,
2131 metadata: &write.item.metadata,
2132 repositories: &write.item.repositories,
2133 parent: write.item.project.as_ref(),
2134 },
2135 write.target.as_ref(),
2136 &write.depends_on,
2137 )
2138 .await
2139 }
2140
2141 async fn write_project(&self, write: &ItemWrite<Project>) -> Result<NativeId, SourceError> {
2142 self.write_item(
2143 &Incoming {
2144 kind: ItemKind::Project,
2145 title: &write.item.title,
2146 content: write.item.content.as_deref(),
2147 status: &write.item.status,
2148 labels: &write.item.labels,
2149 metadata: &write.item.metadata,
2150 repositories: &write.item.repositories,
2151 parent: None,
2152 },
2153 write.target.as_ref(),
2154 &write.depends_on,
2155 )
2156 .await
2157 }
2158
2159 async fn delete_task(&self, id: &NativeId) -> Result<(), SourceError> {
2160 self.delete_item(id).await
2161 }
2162
2163 async fn delete_project(&self, id: &NativeId) -> Result<(), SourceError> {
2164 self.delete_item(id).await
2165 }
2166}
2167
2168const RECORDED_CURSOR: &str = "onetaskgraph.depends_on:";
2171
2172const ORIGIN_FIELD: &str = "onetaskgraph.origin";
2177
2178const ORIGIN_KEY: &str = "onetaskgraph.origin";
2192
2193fn recorded_offset(
2201 cursor: Option<&str>,
2202 direction: Direction,
2203) -> Result<Option<usize>, SourceError> {
2204 cursor
2205 .and_then(|cursor| cursor.strip_prefix(RECORDED_CURSOR))
2206 .map(|offset| {
2207 if direction != Direction::DependsOn {
2208 return Err(SourceError::Config {
2209 message: format!(
2210 "{RECORDED_CURSOR}{offset} resumes recorded forward edges, which a \
2211 reverse dependency read never issues; resume it in the direction \
2212 that reported it"
2213 ),
2214 });
2215 }
2216 offset.parse().map_err(|_| SourceError::Config {
2217 message: format!("{RECORDED_CURSOR}{offset} is not a recorded-edge cursor"),
2218 })
2219 })
2220 .transpose()
2221}
2222
2223fn recorded_page(edges: Vec<DependencyEdge>, offset: usize, limit: usize) -> Page<DependencyEdge> {
2224 let mut page = offset_page(edges, offset, limit.max(1));
2225 page.next = page
2226 .next
2227 .map(|cursor| Cursor(format!("{RECORDED_CURSOR}{}", cursor.0)));
2228 page
2229}
2230
2231fn related_kind(value: &Value) -> Result<ItemKind, SourceError> {
2237 let parent = optional_str(value.get("parent").unwrap_or(&Value::Null), "id")?;
2238 if parent.is_some() {
2239 return Ok(ItemKind::Task);
2240 }
2241 let (_, slot) = metadata_body(optional_str(value, "body")?.map(str::to_owned))?;
2242 let id = required_str(value, "id")?;
2243 let marked = ItemKind::from_metadata(&slot).map_err(|message| SourceError::Malformed {
2244 message: format!("GitHub issue {id}: {message}"),
2245 })?;
2246 let sub_issues = sub_issue_total(value)?;
2247 Ok(if sub_issues > 0 || marked == Some(ItemKind::Project) {
2248 ItemKind::Project
2249 } else {
2250 ItemKind::Task
2251 })
2252}
2253
2254fn state_input(target: &StatusTarget) -> Value {
2261 match target {
2262 StatusTarget::Closed(reason) => json!({"value":"CLOSED","stateReason":reason.reason()}),
2263 StatusTarget::Column(_) | StatusTarget::Disabled => json!({"value":"OPEN"}),
2264 }
2265}
2266
2267fn slot_metadata(
2274 incoming: &Incoming<'_>,
2275 own_repository: Option<&Repository>,
2276 fallback: &[DependencyEdge],
2277) -> BTreeMap<String, Value> {
2278 let mut metadata = incoming.metadata.clone();
2279 metadata.remove(ORIGIN_KEY);
2280 metadata.insert(
2281 ItemKind::METADATA_KEY.to_owned(),
2282 Value::String(incoming.kind.marker().to_owned()),
2283 );
2284 let derivable = own_repository
2285 .map(|own| incoming.repositories == [own.clone()])
2286 .unwrap_or(incoming.repositories.is_empty());
2287 if derivable {
2288 metadata.remove(Repository::METADATA_KEY);
2289 } else {
2290 metadata.insert(
2291 Repository::METADATA_KEY.to_owned(),
2292 Value::Array(
2293 incoming
2294 .repositories
2295 .iter()
2296 .map(|repository| Value::String(repository.as_str().to_owned()))
2297 .collect(),
2298 ),
2299 );
2300 }
2301 if fallback.is_empty() {
2302 metadata.remove(DependencyEdge::RECORDED_KEY);
2303 } else {
2304 metadata.insert(
2305 DependencyEdge::RECORDED_KEY.to_owned(),
2306 Value::Array(
2307 fallback
2308 .iter()
2309 .map(|edge| json!({"id":edge.to.id(),"kind":edge.to.kind}))
2310 .collect(),
2311 ),
2312 );
2313 }
2314 metadata
2315}
2316
2317fn labels(content: &Value, field_values: &[Value]) -> Result<Vec<Label>, SourceError> {
2318 let direct = optional_nodes(content.get("labels"), "content labels")?;
2319 let field = field_values
2320 .iter()
2321 .find_map(|value| value.get("labels"))
2322 .map(|labels| optional_nodes(Some(labels), "field labels"))
2323 .transpose()?
2324 .flatten();
2325 let labels = direct
2326 .into_iter()
2327 .flatten()
2328 .chain(field.into_iter().flatten())
2329 .map(|v| {
2330 Ok(Label {
2331 id: NativeId(required_str(v, "id")?.to_owned()),
2332 name: required_str(v, "name")?.to_owned(),
2333 color: optional_str(v, "color")?.map(str::to_owned),
2334 })
2335 })
2336 .collect::<Result<Vec<_>, SourceError>>()?
2337 .into_iter()
2338 .fold(Vec::new(), |mut labels, label| {
2339 if !labels.iter().any(|x: &Label| x.id == label.id) {
2340 labels.push(label);
2341 }
2342 labels
2343 });
2344 Ok(labels)
2345}
2346
2347fn text_field(field_values: &[Value], name: &str) -> Result<Option<String>, SourceError> {
2348 let Some(node) = field_values
2349 .iter()
2350 .find(|node| node.pointer("/field/name").and_then(Value::as_str) == Some(name))
2351 else {
2352 return Ok(None);
2353 };
2354 Ok(optional_str(node, "text")?.map(str::to_owned))
2355}
2356
2357fn valid_github_owner(owner: &str) -> bool {
2358 !owner.is_empty()
2359 && owner.len() <= 39
2360 && !owner.starts_with('-')
2361 && !owner.ends_with('-')
2362 && !owner.contains("--")
2363 && owner
2364 .bytes()
2365 .all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
2366}
2367
2368fn valid_github_repository_name(name: &str) -> bool {
2371 !name.is_empty()
2372 && name.len() <= 100
2373 && name != "."
2374 && name != ".."
2375 && name
2376 .bytes()
2377 .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
2378}
2379
2380fn valid_environment_name(name: &str) -> bool {
2381 let mut bytes = name.bytes();
2382 bytes
2383 .next()
2384 .is_some_and(|byte| byte.is_ascii_alphabetic() || byte == b'_')
2385 && bytes.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
2386}
2387
2388fn sub_issue_total(issue: &Value) -> Result<u64, SourceError> {
2395 let summary = issue
2396 .get("subIssuesSummary")
2397 .ok_or_else(|| SourceError::Malformed {
2398 message: "GitHub issue is missing subIssuesSummary".into(),
2399 })?;
2400 summary
2401 .get("total")
2402 .and_then(Value::as_u64)
2403 .ok_or_else(|| SourceError::Malformed {
2404 message: "GitHub issue subIssuesSummary.total is not an unsigned integer".into(),
2405 })
2406}
2407
2408fn required_str<'a>(value: &'a Value, field: &str) -> Result<&'a str, SourceError> {
2409 value
2410 .get(field)
2411 .and_then(Value::as_str)
2412 .ok_or_else(|| SourceError::Malformed {
2413 message: format!("GitHub response is missing string field {field}"),
2414 })
2415}
2416
2417const METADATA_OPEN: &str = "<!-- onetaskgraph.metadata\n";
2425const METADATA_CLOSE: &str = "\n-->";
2426
2427fn metadata_body(
2433 body: Option<String>,
2434) -> Result<(Option<String>, BTreeMap<String, Value>), SourceError> {
2435 let Some(body) = body else {
2436 return Ok((None, BTreeMap::new()));
2437 };
2438 let Some(start) = body.rfind(METADATA_OPEN) else {
2439 return Ok((Some(body), BTreeMap::new()));
2440 };
2441 let encoded_start = start + METADATA_OPEN.len();
2442 let Some(relative_end) = body[encoded_start..].find(METADATA_CLOSE) else {
2443 return Err(SourceError::Malformed {
2444 message: "unterminated onetaskgraph metadata slot in GitHub issue body".into(),
2445 });
2446 };
2447 let encoded_end = encoded_start + relative_end;
2448 if !body[encoded_end + METADATA_CLOSE.len()..].trim().is_empty() {
2449 return Ok((Some(body), BTreeMap::new()));
2450 }
2451 let metadata = serde_json::from_str(&body[encoded_start..encoded_end]).map_err(|error| {
2452 SourceError::Malformed {
2453 message: format!(
2454 "invalid canonical JSON in GitHub issue onetaskgraph metadata slot: {error}"
2455 ),
2456 }
2457 })?;
2458 let visible = body[..start].trim_end();
2459 Ok(((!visible.is_empty()).then(|| visible.to_owned()), metadata))
2460}
2461
2462fn compose_body(
2463 content: Option<&str>,
2464 metadata: &BTreeMap<String, Value>,
2465) -> Result<Option<String>, SourceError> {
2466 let visible = content.unwrap_or_default();
2467 if metadata.is_empty() {
2468 return Ok((!visible.is_empty()).then(|| visible.to_owned()));
2469 }
2470 let encoded = serde_json::to_string(metadata).map_err(|error| SourceError::Malformed {
2471 message: error.to_string(),
2472 })?;
2473 Ok(Some(if visible.is_empty() {
2474 format!("{METADATA_OPEN}{encoded}{METADATA_CLOSE}")
2475 } else {
2476 format!("{visible}\n\n{METADATA_OPEN}{encoded}{METADATA_CLOSE}")
2477 }))
2478}
2479
2480fn required_bool(value: &Value, field: &str) -> Result<bool, SourceError> {
2481 value
2482 .get(field)
2483 .and_then(Value::as_bool)
2484 .ok_or_else(|| SourceError::Malformed {
2485 message: format!("GitHub response is missing boolean field {field}"),
2486 })
2487}
2488fn optional_str<'a>(value: &'a Value, field: &str) -> Result<Option<&'a str>, SourceError> {
2489 match value.get(field) {
2490 None | Some(Value::Null) => Ok(None),
2491 Some(value) => value
2492 .as_str()
2493 .map(Some)
2494 .ok_or_else(|| SourceError::Malformed {
2495 message: format!("GitHub response field {field} is not a string or null"),
2496 }),
2497 }
2498}
2499fn optional_nodes<'a>(
2500 connection: Option<&'a Value>,
2501 name: &str,
2502) -> Result<Option<&'a Vec<Value>>, SourceError> {
2503 match connection {
2504 None | Some(Value::Null) => Ok(None),
2505 Some(value) => value
2506 .get("nodes")
2507 .and_then(Value::as_array)
2508 .map(Some)
2509 .ok_or_else(|| SourceError::Malformed {
2510 message: format!("GitHub {name}.nodes is not an array"),
2511 }),
2512 }
2513}
2514fn complete_connection(connection: &Value, name: &str) -> Result<(), SourceError> {
2515 let page_info = connection
2516 .get("pageInfo")
2517 .ok_or_else(|| SourceError::Malformed {
2518 message: format!("GitHub {name} has no pageInfo"),
2519 })?;
2520 if required_bool(page_info, "hasNextPage")? {
2521 return Err(SourceError::Malformed {
2522 message: format!(
2523 "GitHub {name} exceeds the supported nested connection size of {NESTED_PAGE_SIZE}"
2524 ),
2525 });
2526 }
2527 Ok(())
2528}
2529fn optional_time(value: &Value, field: &str) -> Result<Option<DateTime<Utc>>, SourceError> {
2530 optional_str(value, field)?
2531 .map(|timestamp| {
2532 timestamp.parse().map_err(|error| SourceError::Malformed {
2533 message: format!("GitHub response field {field} is not a timestamp: {error}"),
2534 })
2535 })
2536 .transpose()
2537}
2538fn validate_page(page: &PageRequest) -> Result<(), SourceError> {
2539 if page.limit == 0 {
2540 Err(SourceError::Config {
2541 message: "page limit must be at least 1".into(),
2542 })
2543 } else {
2544 Ok(())
2545 }
2546}
2547fn next_cursor(connection: &Value) -> Result<Option<Cursor>, SourceError> {
2548 let page = connection
2549 .get("pageInfo")
2550 .filter(|value| value.is_object())
2551 .ok_or_else(|| SourceError::Malformed {
2552 message: "GitHub connection is missing pageInfo".into(),
2553 })?;
2554 if required_bool(page, "hasNextPage")? {
2555 let cursor = required_str(page, "endCursor")?;
2556 validate_cursor_progress(None, cursor)?;
2557 Ok(Some(Cursor(cursor.into())))
2558 } else {
2559 Ok(None)
2560 }
2561}
2562fn validate_cursor_progress(previous: Option<&str>, next: &str) -> Result<(), SourceError> {
2563 if next.is_empty() || previous == Some(next) {
2564 Err(SourceError::Malformed {
2565 message: "GitHub pagination cursor is empty or did not advance".into(),
2566 })
2567 } else {
2568 Ok(())
2569 }
2570}
2571fn numeric_cursor(cursor: Option<&Cursor>) -> Result<usize, SourceError> {
2572 cursor.map_or(Ok(0), |c| {
2573 c.0.parse().map_err(|_| SourceError::Config {
2574 message: "page cursor is invalid".into(),
2575 })
2576 })
2577}
2578fn offset_page<T>(mut items: Vec<T>, offset: usize, limit: usize) -> Page<T> {
2579 if offset > items.len() {
2580 return Page::last(vec![]);
2581 }
2582 let tail = items.split_off(offset);
2583 let mut selected = tail;
2584 let next = (selected.len() > limit).then(|| Cursor((offset + limit).to_string()));
2585 selected.truncate(limit);
2586 Page {
2587 items: selected,
2588 next,
2589 }
2590}