#![deny(missing_docs)]
use std::collections::BTreeMap;
use chrono::{DateTime, Utc};
use onetaskgraph_plugin_api::{
Capabilities, Cursor, DependencyEdge, DependencyEndpoint, DependencyKind, DependencySupport,
Direction, Health, ItemKind, ItemWrite, Label, NativeId, Page, PageRequest, Project,
ProjectQuery, Repository, SecretResolver, SourceError, SourceName, SourcePlugin, Status,
StatusCategory, Support, Task, TaskQuery, TaskSource, WriteSupport,
};
use reqwest::{Client, StatusCode, Url};
use schemars::{Schema, schema_for};
use secrecy::{ExposeSecret, SecretString};
use serde::Deserialize;
use serde_json::{Value, json};
pub const KIND: &str = "github-projects";
pub const MAX_PAGE_SIZE: u32 = 100;
const NESTED_PAGE_SIZE: u32 = 50;
pub mod graphql {
pub const PROJECT: &str = r#"query($owner:String!,$number:Int!,$first:Int!,$after:String,$nestedFirst:Int!){
owner:repositoryOwner(login:$owner){
... on ProjectV2Owner{projectV2(number:$number){...Project}}
}
} fragment Project on ProjectV2 { id title shortDescription url createdAt updatedAt closed
fields(first:$nestedFirst){nodes{
... on ProjectV2SingleSelectField{__typename id name options{id name}}
... on ProjectV2Field{__typename id name}
}pageInfo{hasNextPage}}
items(first:$first,after:$after){nodes{id fieldValues(first:$nestedFirst){nodes{
... on ProjectV2ItemFieldSingleSelectValue{name field{
... on ProjectV2SingleSelectField{id name options{id name}}
}}
... on ProjectV2ItemFieldTextValue{text field{... on ProjectV2Field{id name}}}
... on ProjectV2ItemFieldLabelValue{labels(first:$nestedFirst){nodes{id name color}pageInfo{hasNextPage}}}
}pageInfo{hasNextPage}} content{
... on Issue{__typename id title body url createdAt updatedAt state repository{nameWithOwner} labels(first:$nestedFirst){nodes{id name color}pageInfo{hasNextPage}}}
... on PullRequest{__typename id title body url createdAt updatedAt state repository{nameWithOwner} labels(first:$nestedFirst){nodes{id name color}pageInfo{hasNextPage}}}
... on DraftIssue{__typename id title body createdAt updatedAt}
}} pageInfo{hasNextPage endCursor}}
}"#;
pub const TASK_DEPENDENCIES: &str = r#"query($id:ID!,$first:Int!,$after:String){node(id:$id){__typename ... on Issue{blockedBy(first:$first,after:$after){nodes{id}pageInfo{hasNextPage endCursor}}blocking(first:$first,after:$after){nodes{id}pageInfo{hasNextPage endCursor}}}}}"#;
pub const RELATED_PROJECTS: &str = r#"query($id:ID!,$first:Int!,$after:String!){node(id:$id){... on Issue{projectItems(first:$first,after:$after){nodes{project{id}}pageInfo{hasNextPage endCursor}}}}}"#;
pub const PROJECT_DEPENDENCIES: &str = r#"query($id:ID!,$first:Int!,$after:String,$nestedFirst:Int!){node(id:$id){... on Issue{blockedBy(first:$first,after:$after){nodes{id projectItems(first:$nestedFirst){nodes{project{id}}pageInfo{hasNextPage endCursor}}}pageInfo{hasNextPage endCursor}}blocking(first:$first,after:$after){nodes{id projectItems(first:$nestedFirst){nodes{project{id}}pageInfo{hasNextPage endCursor}}}pageInfo{hasNextPage endCursor}}}}}"#;
pub const CREATE_DRAFT: &str = r#"mutation($input:AddProjectV2DraftIssueInput!){addProjectV2DraftIssue(input:$input){projectItem{id content{... on DraftIssue{id}}}}}"#;
pub const UPDATE_DRAFT: &str = r#"mutation($input:UpdateProjectV2DraftIssueInput!){updateProjectV2DraftIssue(input:$input){draftIssue{id}}}"#;
pub const UPDATE_ISSUE: &str =
r#"mutation($input:UpdateIssueInput!){updateIssue(input:$input){issue{id}}}"#;
pub const UPDATE_FIELD: &str = r#"mutation($input:UpdateProjectV2ItemFieldValueInput!){updateProjectV2ItemFieldValue(input:$input){projectV2Item{id}}}"#;
pub const UPDATE_PROJECT: &str =
r#"mutation($input:UpdateProjectV2Input!){updateProjectV2(input:$input){projectV2{id}}}"#;
pub const ADD_BLOCKED_BY: &str = r#"mutation($input:AddBlockedByInput!){addBlockedBy(input:$input){issue{id} blockingIssue{id}}}"#;
pub const REMOVE_BLOCKED_BY: &str = r#"mutation($input:RemoveBlockedByInput!){removeBlockedBy(input:$input){issue{id} blockingIssue{id}}}"#;
}
fn default_token_env() -> String {
"GH_PROJECTS_TOKEN".to_owned()
}
fn default_endpoint() -> String {
"https://api.github.com/graphql".to_owned()
}
#[derive(Debug, Clone, Default, Deserialize, schemars::JsonSchema)]
#[serde(default, deny_unknown_fields)]
pub struct GitHubProjectsConfig {
pub owner: String, pub project_number: u32, #[serde(default = "default_token_env")]
pub token_env: String, #[serde(default = "default_endpoint")]
pub endpoint: String, #[serde(default)]
pub status_mapping: BTreeMap<String, StatusCategory>, }
#[derive(Debug, Clone, Copy, Default)]
pub struct Plugin;
impl SourcePlugin for Plugin {
fn kind(&self) -> &'static str {
KIND
}
fn config_schema(&self) -> Schema {
schema_for!(GitHubProjectsConfig)
}
fn build(
&self,
name: &SourceName,
config: &Value,
secrets: &dyn SecretResolver,
) -> Result<Box<dyn TaskSource>, SourceError> {
let config: GitHubProjectsConfig =
serde_json::from_value(config.clone()).map_err(|e| SourceError::Config {
message: format!("source {name}: {e}"),
})?;
let source =
GitHubProjectsSource::new(name, config, secrets).map_err(|error| match error {
SourceError::Config { message } => SourceError::Config {
message: format!("source {name}: {message}"),
},
SourceError::Auth { message } => SourceError::Auth {
message: format!("source {name}: {message}"),
},
other => other,
})?;
Ok(Box::new(source))
}
}
pub struct GitHubProjectsSource {
name: SourceName,
owner: String, project_number: u32, endpoint: Url,
token: SecretString,
credential_name: String, statuses: BTreeMap<StatusName, StatusCategory>,
client: Client,
}
impl GitHubProjectsSource {
pub fn new(
name: &SourceName,
config: GitHubProjectsConfig,
secrets: &dyn SecretResolver,
) -> Result<Self, SourceError> {
if !valid_github_owner(&config.owner) {
return Err(SourceError::Config {
message: "owner must be 1-39 ASCII letters, digits, or single hyphens, and cannot start or end with a hyphen".into(),
});
}
if config.project_number == 0 || config.project_number > i32::MAX as u32 {
return Err(SourceError::Config {
message: format!("project_number must be between 1 and {}", i32::MAX),
});
}
if !valid_environment_name(&config.token_env) {
return Err(SourceError::Config {
message: "token_env must be a valid environment-variable name".into(),
});
}
let endpoint = Url::parse(&config.endpoint).map_err(|e| SourceError::Config {
message: format!("endpoint is not a valid URL: {e}"),
})?;
if endpoint.scheme() != "https"
&& !(endpoint.scheme() == "http"
&& endpoint
.host_str()
.is_some_and(|h| h == "127.0.0.1" || h == "localhost" || h == "::1"))
{
return Err(SourceError::Config {
message:
"endpoint must use HTTPS (HTTP is accepted only for a loopback test server)"
.into(),
});
}
let token = secrets.get(&config.token_env).filter(|token| !token.expose_secret().trim().is_empty()).ok_or_else(|| SourceError::Auth {
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),
})?;
Ok(Self {
name: name.clone(),
owner: config.owner,
project_number: config.project_number,
endpoint,
token,
credential_name: config.token_env,
statuses: normalize_status_mapping(config.status_mapping)?,
client: Client::builder()
.user_agent("onetaskgraph")
.build()
.map_err(|e| SourceError::Config {
message: format!("cannot build HTTP client: {e}"),
})?,
})
}
async fn graphql(&self, query: &str, variables: Value) -> Result<Value, SourceError> {
let response = self
.client
.post(self.endpoint.clone())
.bearer_auth(self.token.expose_secret())
.json(&json!({"query": query, "variables": variables}))
.send()
.await
.map_err(|e| SourceError::Unavailable {
message: format!("GitHub GraphQL request failed: {e}"),
})?;
let status = response.status();
let retry_after = response
.headers()
.get("retry-after")
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse().ok());
let exhausted = response
.headers()
.get("x-ratelimit-remaining")
.and_then(|v| v.to_str().ok())
== Some("0");
if status == StatusCode::TOO_MANY_REQUESTS || exhausted {
return Err(SourceError::RateLimited {
retry_after_seconds: retry_after,
});
}
if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN {
return Err(SourceError::Auth {
message: format!(
"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"
),
});
}
if !status.is_success() {
return Err(SourceError::Unavailable {
message: format!("GitHub GraphQL returned HTTP {status}"),
});
}
let body: Value = response.json().await.map_err(|e| SourceError::Malformed {
message: format!("GitHub returned invalid JSON: {e}"),
})?;
let errors = body
.get("errors")
.map(|value| {
value.as_array().ok_or_else(|| SourceError::Malformed {
message: "GitHub response errors is not an array".into(),
})
})
.transpose()?;
if let Some(errors) = errors.filter(|errors| !errors.is_empty()) {
let messages = errors
.iter()
.filter_map(|e| e.get("message").and_then(Value::as_str))
.collect::<Vec<_>>()
.join("; ");
let message = if messages.is_empty() {
"GitHub returned GraphQL errors".into()
} else {
messages
};
let normalized = message.to_ascii_lowercase();
if normalized.contains("resource not accessible") || normalized.contains("scope") {
return Err(SourceError::Auth {
message: format!(
"{message}; grant {} Projects and Issues read/write plus Pull requests read-only access for every repository represented on the board",
self.credential_name
),
});
}
return Err(SourceError::Refused { message });
}
body.get("data")
.filter(|data| data.is_object())
.cloned()
.ok_or_else(|| SourceError::Malformed {
message: "GitHub response has no data object".into(),
})
}
async fn project_value(
&self,
items_after: Option<&str>,
items_first: u32,
) -> Result<Value, SourceError> {
let data = self.graphql(graphql::PROJECT, json!({"owner":self.owner,"number":self.project_number,"first":items_first.min(MAX_PAGE_SIZE),"after":items_after,"nestedFirst":NESTED_PAGE_SIZE})).await?;
let project = data
.pointer("/owner/projectV2")
.filter(|v| !v.is_null())
.cloned()
.ok_or_else(|| SourceError::Refused {
message: format!(
"GitHub project {}/{} was not found or is not visible to the token",
self.owner, self.project_number
),
})?;
Ok(project)
}
fn status(&self, item: &Value) -> Result<Status, SourceError> {
let fields = item
.pointer("/fieldValues/nodes")
.and_then(Value::as_array)
.expect("task validates fieldValues.nodes before mapping status");
let name = fields
.iter()
.find(|v| v.pointer("/field/name").and_then(Value::as_str) == Some("Status"))
.map(|value| required_str(value, "name"))
.transpose()?
.or(optional_str(
item.get("content").unwrap_or(&Value::Null),
"state",
)?)
.unwrap_or("Unknown")
.to_owned();
let category = self
.statuses
.get(&StatusName::new(&name))
.copied()
.unwrap_or_else(|| match name.to_ascii_lowercase().as_str() {
"backlog" => StatusCategory::Backlog,
"todo" | "open" => StatusCategory::Todo,
"in progress" | "in review" => StatusCategory::InProgress,
"done" | "closed" | "merged" => StatusCategory::Done,
"cancelled" | "canceled" => StatusCategory::Cancelled,
_ => StatusCategory::Unknown,
});
Ok(Status { category, name })
}
fn labels(item: &Value) -> Result<Vec<Label>, SourceError> {
let direct = optional_nodes(item.pointer("/content/labels"), "content labels")?;
let field_values = item
.pointer("/fieldValues/nodes")
.and_then(Value::as_array)
.expect("task validates fieldValues.nodes before mapping labels");
let field = field_values
.iter()
.find_map(|value| value.get("labels"))
.map(|labels| optional_nodes(Some(labels), "field labels"))
.transpose()?
.flatten();
let labels = direct
.into_iter()
.flatten()
.chain(field.into_iter().flatten())
.map(|v| {
Ok(Label {
id: NativeId(required_str(v, "id")?.to_owned()),
name: required_str(v, "name")?.to_owned(),
color: optional_str(v, "color")?.map(str::to_owned),
})
})
.collect::<Result<Vec<_>, SourceError>>()?
.into_iter()
.fold(Vec::new(), |mut labels, label| {
if !labels.iter().any(|x: &Label| x.id == label.id) {
labels.push(label);
}
labels
});
Ok(labels)
}
fn task(&self, project_id: &str, item: &Value) -> Result<Option<Task>, SourceError> {
let content = item.get("content").ok_or_else(|| SourceError::Malformed {
message: "GitHub project item is missing content".into(),
})?;
if content.is_null() {
return Ok(None);
}
let field_values = item
.get("fieldValues")
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project item is missing fieldValues".into(),
})?;
complete_connection(field_values, "project item field values")?;
field_values
.get("nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project item fieldValues.nodes is not an array".into(),
})?;
if let Some(labels) = content.get("labels") {
complete_connection(labels, "content labels")?;
}
for field_value in field_values["nodes"].as_array().expect("validated above") {
if let Some(labels) = field_value.get("labels") {
complete_connection(labels, "project item field labels")?;
}
}
Ok(Some(Task {
id: NativeId(required_str(content, "id")?.to_owned()),
title: required_str(content, "title")?.to_owned(),
content: optional_str(content, "body")?
.filter(|s| !s.is_empty())
.map(str::to_owned),
status: self.status(item)?,
labels: Self::labels(item)?,
project: Some(NativeId(project_id.to_owned())),
url: optional_str(content, "url")?.map(str::to_owned),
created_at: optional_time(content, "createdAt")?,
updated_at: optional_time(content, "updatedAt")?,
metadata: metadata_field(field_values)?,
repositories: repositories(content, field_values)?,
}))
}
fn project(&self, value: &Value) -> Result<Project, SourceError> {
let id = required_str(value, "id")?;
let (content, metadata) =
metadata_description(optional_str(value, "shortDescription")?.map(str::to_owned))?;
let repositories = repositories_from_metadata(&metadata)?;
Ok(Project {
id: NativeId(id.into()),
title: required_str(value, "title")?.into(),
content,
status: Status {
category: if required_bool(value, "closed")? {
StatusCategory::Done
} else {
StatusCategory::InProgress
},
name: if required_bool(value, "closed")? {
"Closed"
} else {
"Open"
}
.into(),
},
labels: vec![],
url: optional_str(value, "url")?.map(str::to_owned),
created_at: optional_time(value, "createdAt")?,
updated_at: optional_time(value, "updatedAt")?,
metadata,
repositories,
})
}
async fn all_tasks(&self) -> Result<Vec<Task>, SourceError> {
let mut after = None;
let mut tasks = Vec::new();
loop {
let project = self.project_value(after.as_deref(), MAX_PAGE_SIZE).await?;
let project_id = required_str(&project, "id")?;
let items = project
.pointer("/items/nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project items.nodes is not an array".into(),
})?;
for item in items {
if let Some(task) = self.task(project_id, item)? {
tasks.push(task);
}
}
let page =
project
.pointer("/items/pageInfo")
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project items have no pageInfo".into(),
})?;
if !required_bool(page, "hasNextPage")? {
break;
}
let next = required_str(page, "endCursor")?;
validate_cursor_progress(after.as_deref(), next)?;
after = Some(next.to_owned());
}
Ok(tasks)
}
async fn board_and_item(
&self,
content_id: Option<&NativeId>,
) -> Result<(Value, Option<Value>), SourceError> {
let mut after = None;
loop {
let project = self.project_value(after.as_deref(), MAX_PAGE_SIZE).await?;
let nodes = project
.pointer("/items/nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project items.nodes is not an array".into(),
})?;
let found = content_id.and_then(|wanted| {
nodes
.iter()
.find(|item| {
item.pointer("/content/id").and_then(Value::as_str)
== Some(wanted.0.as_str())
})
.cloned()
});
if found.is_some() || content_id.is_none() {
return Ok((project, found));
}
let page =
project
.pointer("/items/pageInfo")
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project items have no pageInfo".into(),
})?;
if !required_bool(page, "hasNextPage")? {
return Ok((project, None));
}
let next = required_str(page, "endCursor")?;
validate_cursor_progress(after.as_deref(), next)?;
after = Some(next.to_owned());
}
}
async fn set_item_field(
&self,
project_id: &str,
item_id: &str,
field_id: &str,
value: Value,
) -> Result<(), SourceError> {
let data = self
.graphql(
graphql::UPDATE_FIELD,
json!({"input":{
"projectId":project_id,"itemId":item_id,"fieldId":field_id,"value":value
}}),
)
.await?;
let returned = data
.pointer("/updateProjectV2ItemFieldValue/projectV2Item")
.ok_or_else(|| SourceError::Malformed {
message: "GitHub field update returned no project item".into(),
})?;
if required_str(returned, "id")? != item_id {
return Err(SourceError::Malformed {
message: "GitHub field update returned the wrong project item".into(),
});
}
Ok(())
}
async fn native_dependency_ids(&self, id: &NativeId) -> Result<Vec<String>, SourceError> {
let mut after = None;
let mut ids = Vec::new();
loop {
let data = self
.graphql(
graphql::TASK_DEPENDENCIES,
json!({"id":id.0,"first":MAX_PAGE_SIZE,"after":after}),
)
.await?;
let connection =
data.pointer("/node/blockedBy")
.ok_or_else(|| SourceError::Malformed {
message: "GitHub dependency response has no blockedBy connection".into(),
})?;
ids.extend(
connection
.get("nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub dependency response nodes is not an array".into(),
})?
.iter()
.map(|value| required_str(value, "id").map(str::to_owned))
.collect::<Result<Vec<_>, _>>()?,
);
let next = next_cursor(connection)?;
if let Some(next) = &next {
validate_cursor_progress(after.as_deref(), &next.0)?;
}
after = next.map(|cursor| cursor.0);
if after.is_none() {
return Ok(ids);
}
}
}
fn field<'a>(project: &'a Value, name: &str) -> Result<Option<&'a Value>, SourceError> {
complete_connection(
project.get("fields").unwrap_or(&Value::Null),
"project fields",
)?;
let fields = project
.pointer("/fields/nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project fields.nodes is not an array".into(),
})?;
Ok(fields
.iter()
.find(|field| field.get("name").and_then(Value::as_str) == Some(name)))
}
fn task_metadata(
write: &ItemWrite<Task>,
repositories: RepositoryStorage,
) -> Result<BTreeMap<String, Value>, SourceError> {
let mut metadata = write.item.metadata.clone();
if repositories == RepositoryStorage::Recorded && !write.item.repositories.is_empty() {
metadata.insert(
Repository::METADATA_KEY.into(),
Value::Array(
write
.item
.repositories
.iter()
.map(|repository| Value::String(repository.as_str().to_owned()))
.collect(),
),
);
} else {
metadata.remove(Repository::METADATA_KEY);
}
if !write.depends_on.is_empty() {
metadata.insert(
DependencyEdge::RECORDED_KEY.into(),
Value::Array(
write
.depends_on
.iter()
.map(|edge| endpoint_value(&edge.to))
.collect(),
),
);
} else {
metadata.remove(DependencyEdge::RECORDED_KEY);
}
Ok(metadata)
}
async fn dependencies(
&self,
id: &NativeId,
direction: Direction,
page: &PageRequest,
) -> Result<Page<DependencyEdge>, SourceError> {
validate_page(page)?;
let limit = page.limit.min(MAX_PAGE_SIZE) as usize;
let cursor = page.cursor.as_ref().map(|c| c.0.as_str());
let recorded = recorded_offset(cursor, direction)?;
let data = self
.graphql(
graphql::TASK_DEPENDENCIES,
json!({"id":id.0,"first":page.limit.min(MAX_PAGE_SIZE),
"after":if recorded.is_some() {None} else {cursor}}),
)
.await?;
let node =
data.get("node")
.filter(|v| !v.is_null())
.ok_or_else(|| SourceError::Refused {
message: format!(
"GitHub item {} was not found or does not support dependencies",
id.0
),
})?;
let connection_name = match direction {
Direction::DependsOn => "blockedBy",
Direction::DependedOnBy => "blocking",
};
let natively_names =
(required_str(node, "__typename")? == "Issue").then_some(ItemKind::Task);
if let Some(offset) = recorded {
return Ok(recorded_page(
self.recorded_task_edges(id, direction, natively_names)
.await?,
offset,
limit,
));
}
if natively_names.is_none() {
return Ok(recorded_page(
self.recorded_task_edges(id, direction, natively_names)
.await?,
0,
limit,
));
}
let connection = node
.get(connection_name)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub dependency response is missing its connection".into(),
})?;
let nodes = connection
.get("nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub dependency response nodes is not an array".into(),
})?;
let items = nodes
.iter()
.map(|value| {
let related = NativeId(required_str(value, "id")?.into());
let (from, to) = match direction {
Direction::DependsOn => (id.clone(), related),
Direction::DependedOnBy => (related, id.clone()),
};
Ok(DependencyEdge {
from: DependencyEndpoint::from_native(from, ItemKind::Task),
to: DependencyEndpoint::from_native(to, ItemKind::Task),
kind: DependencyKind::Blocks,
})
})
.collect::<Result<Vec<_>, SourceError>>()?;
let mut next = next_cursor(connection)?;
if let Some(next) = &next {
validate_cursor_progress(cursor, &next.0)?;
}
if next.is_none()
&& !self
.recorded_task_edges(id, direction, natively_names)
.await?
.is_empty()
{
next = Some(Cursor(format!("{RECORDED_CURSOR}0")));
}
Ok(Page { items, next })
}
async fn recorded_task_edges(
&self,
id: &NativeId,
direction: Direction,
natively_names: Option<ItemKind>,
) -> Result<Vec<DependencyEdge>, SourceError> {
if direction != Direction::DependsOn {
return Ok(Vec::new());
}
let Some(task) = self
.all_tasks()
.await?
.into_iter()
.find(|task| task.id == *id)
else {
return Ok(Vec::new());
};
DependencyEdge::recorded(
&task.metadata,
id,
ItemKind::Task,
&self.name,
natively_names,
)
.map_err(|message| SourceError::Malformed { message })
}
async fn related_issue_projects(&self, issue: &Value) -> Result<Vec<NativeId>, SourceError> {
let issue_id = required_str(issue, "id")?;
let mut connection =
issue
.get("projectItems")
.cloned()
.ok_or_else(|| SourceError::Malformed {
message: "GitHub related issue is missing projectItems".into(),
})?;
let mut projects = Vec::new();
let mut previous = None;
loop {
let nodes = connection
.get("nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub related issue projectItems.nodes is not an array".into(),
})?;
for item in nodes {
projects.push(NativeId(
required_str(
item.get("project").ok_or_else(|| SourceError::Malformed {
message: "GitHub dependency project item has no project".into(),
})?,
"id",
)?
.into(),
));
}
let Some(cursor) = next_cursor(&connection)? else {
break;
};
validate_cursor_progress(previous.as_deref(), &cursor.0)?;
previous = Some(cursor.0.clone());
let data = self
.graphql(
graphql::RELATED_PROJECTS,
json!({"id":issue_id,"first":MAX_PAGE_SIZE,"after":cursor.0}),
)
.await?;
connection = data.pointer("/node/projectItems").cloned().ok_or_else(|| {
SourceError::Malformed {
message: "GitHub related issue response is missing projectItems".into(),
}
})?;
}
Ok(projects)
}
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum ContentKind {
DraftIssue,
Issue,
}
#[derive(Clone, Copy, PartialEq, Eq)]
enum RepositoryStorage {
Native,
Recorded,
}
impl ContentKind {
fn parse(content: &Value) -> Result<Self, SourceError> {
match required_str(content, "__typename")? {
"DraftIssue" => Ok(Self::DraftIssue),
"Issue" => Ok(Self::Issue),
other => Err(SourceError::Refused {
message: format!("GitHub {other} items cannot be updated by this destination"),
}),
}
}
}
#[async_trait::async_trait]
impl TaskSource for GitHubProjectsSource {
fn kind(&self) -> &'static str {
KIND
}
fn capabilities(&self) -> Capabilities {
Capabilities {
projects: Support::Native,
orphan_tasks: Support::Unsupported,
filter_by_label: Support::Unsupported,
filter_by_status: Support::Unsupported,
search_title: Support::Unsupported,
search_content: Support::Unsupported,
task_dependencies: DependencySupport::BothDirections,
project_dependencies: DependencySupport::BothDirections,
max_page_size: MAX_PAGE_SIZE,
}
}
async fn health(&self) -> Result<Health, SourceError> {
let project = self.project_value(None, 1).await?;
Ok(Health {
reachable: true,
detail: Some(format!(
"reading GitHub project {}/{} ({})",
self.owner,
self.project_number,
required_str(&project, "title")?
)),
})
}
async fn get_task(&self, id: &NativeId) -> Result<Option<Task>, SourceError> {
Ok(self
.all_tasks()
.await?
.into_iter()
.find(|task| task.id == *id))
}
async fn get_project(&self, id: &NativeId) -> Result<Option<Project>, SourceError> {
let value = self.project_value(None, 1).await?;
let project = self.project(&value)?;
Ok((project.id == *id).then_some(project))
}
async fn query_tasks(
&self,
_query: &TaskQuery,
page: &PageRequest,
) -> Result<Page<Task>, SourceError> {
validate_page(page)?;
let value = self
.project_value(page.cursor.as_ref().map(|c| c.0.as_str()), page.limit)
.await?;
let id = required_str(&value, "id")?;
let items_connection = value.get("items").ok_or_else(|| SourceError::Malformed {
message: "GitHub project response is missing items".into(),
})?;
let nodes = items_connection
.get("nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project items.nodes is not an array".into(),
})?;
let items = nodes
.iter()
.map(|item| self.task(id, item))
.collect::<Result<Vec<_>, SourceError>>()?
.into_iter()
.flatten()
.collect();
let next = next_cursor(items_connection)?;
if let Some(next) = &next {
validate_cursor_progress(
page.cursor.as_ref().map(|cursor| cursor.0.as_str()),
&next.0,
)?;
}
Ok(Page { items, next })
}
async fn query_projects(
&self,
_query: &ProjectQuery,
page: &PageRequest,
) -> Result<Page<Project>, SourceError> {
validate_page(page)?;
if page.cursor.is_some() {
return Err(SourceError::Config {
message: "GitHub project listing does not issue page cursors".into(),
});
}
Ok(Page::last(vec![
self.project(&self.project_value(None, 1).await?)?,
]))
}
async fn labels(&self, page: &PageRequest) -> Result<Page<Label>, SourceError> {
validate_page(page)?;
let offset = numeric_cursor(page.cursor.as_ref())?;
let mut labels = self
.all_tasks()
.await?
.into_iter()
.flat_map(|t| t.labels)
.fold(Vec::new(), |mut all, label| {
if !all.iter().any(|x: &Label| x.id == label.id) {
all.push(label);
}
all
});
labels.sort_by(|a, b| a.name.cmp(&b.name).then(a.id.0.cmp(&b.id.0)));
Ok(offset_page(
labels,
offset,
page.limit.min(MAX_PAGE_SIZE) as usize,
))
}
async fn task_dependencies(
&self,
id: &NativeId,
direction: Direction,
page: &PageRequest,
) -> Result<Page<DependencyEdge>, SourceError> {
self.dependencies(id, direction, page).await
}
async fn project_dependencies(
&self,
id: &NativeId,
direction: Direction,
page: &PageRequest,
) -> Result<Page<DependencyEdge>, SourceError> {
validate_page(page)?;
let project = self.project_value(None, 1).await?;
if required_str(&project, "id")? != id.0 {
return Err(SourceError::Refused {
message: format!("GitHub project {} was not found", id.0),
});
}
let mut edges = Vec::new();
for task in self.all_tasks().await? {
let mut cursor = None;
loop {
let data = self.graphql(graphql::PROJECT_DEPENDENCIES, json!({"id":task.id.0,"first":MAX_PAGE_SIZE,"after":cursor.as_ref().map(|cursor: &Cursor| cursor.0.as_str()),"nestedFirst":MAX_PAGE_SIZE})).await?;
let connection_name = match direction {
Direction::DependsOn => "blockedBy",
Direction::DependedOnBy => "blocking",
};
let Some(connection) = data.pointer(&format!("/node/{connection_name}")) else {
break;
};
let related_issues = connection
.get("nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project dependency nodes is not an array".into(),
})?;
for related_issue in related_issues {
for related in self.related_issue_projects(related_issue).await? {
if related != *id {
let (from, to) = match direction {
Direction::DependsOn => (id.clone(), related),
Direction::DependedOnBy => (related, id.clone()),
};
edges.push(DependencyEdge {
from: DependencyEndpoint::from_native(from, ItemKind::Project),
to: DependencyEndpoint::from_native(to, ItemKind::Project),
kind: DependencyKind::Blocks,
});
}
}
}
let next = next_cursor(connection)?;
if let Some(next) = &next {
validate_cursor_progress(
cursor.as_ref().map(|value: &Cursor| value.0.as_str()),
&next.0,
)?;
}
cursor = next;
if cursor.is_none() {
break;
}
}
}
if direction == Direction::DependsOn {
edges.extend(
DependencyEdge::recorded(
&self.project(&project)?.metadata,
id,
ItemKind::Project,
&self.name,
Some(ItemKind::Project),
)
.map_err(|message| SourceError::Malformed { message })?,
);
}
let offset = numeric_cursor(page.cursor.as_ref())?;
Ok(offset_page(edges, offset, page.limit as usize))
}
fn writes(&self) -> WriteSupport {
WriteSupport::Supported
}
async fn write_task(&self, write: &ItemWrite<Task>) -> Result<NativeId, SourceError> {
let (project, existing) = self.board_and_item(write.target.as_ref()).await?;
let project_id = required_str(&project, "id")?;
let metadata_field =
Self::field(&project, METADATA_FIELD)?.ok_or_else(|| SourceError::Refused {
message: format!("GitHub project has no source-owned {METADATA_FIELD} text field"),
})?;
if required_str(metadata_field, "__typename")? != "ProjectV2Field" {
return Err(SourceError::Refused {
message: format!(
"GitHub project source-owned {METADATA_FIELD} field is not a text field"
),
});
}
let status_selection =
Some(
Self::field(&project, "Status")?.ok_or_else(|| SourceError::Refused {
message: "GitHub project has no Status field".into(),
})?,
)
.map(|field| {
if required_str(field, "__typename")? != "ProjectV2SingleSelectField" {
return Err(SourceError::Refused {
message: "GitHub project Status field is not a single-select field".into(),
});
}
let option = field
.get("options")
.and_then(Value::as_array)
.and_then(|options| {
options.iter().find(|option| {
option
.get("name")
.and_then(Value::as_str)
.is_some_and(|name| {
name.eq_ignore_ascii_case(&write.item.status.name)
})
})
})
.ok_or_else(|| SourceError::Refused {
message: format!(
"GitHub Status field cannot represent status {}",
write.item.status.name
),
})?;
Ok::<_, SourceError>((
required_str(field, "id")?.to_owned(),
required_str(option, "id")?.to_owned(),
))
})
.transpose()?;
let (content_id, item_id, content_kind) = if let Some(target) = &write.target {
let item = existing.ok_or_else(|| SourceError::Refused {
message: format!("GitHub destination item {} was not found", target.0),
})?;
let content_kind = ContentKind::parse(item.get("content").unwrap_or(&Value::Null))?;
if content_kind == ContentKind::DraftIssue && !write.item.labels.is_empty() {
return Err(SourceError::Refused {
message: "GitHub draft items cannot represent labels".into(),
});
}
if content_kind == ContentKind::Issue {
let held = Self::labels(&item)?;
if held != write.item.labels {
return Err(SourceError::Refused {
message: "GitHub issue labels differ from the labels being written".into(),
});
}
let native = item
.pointer("/content/repository/nameWithOwner")
.and_then(Value::as_str)
.map(|value| format!("github.com/{value}"));
if write
.item
.repositories
.iter()
.map(Repository::as_str)
.collect::<Vec<_>>()
!= native.iter().map(String::as_str).collect::<Vec<_>>()
{
return Err(SourceError::Refused {
message:
"GitHub issue repository differs from the repositories being written"
.into(),
});
}
}
let operation = match content_kind {
ContentKind::DraftIssue => graphql::UPDATE_DRAFT,
ContentKind::Issue => graphql::UPDATE_ISSUE,
};
let input = if content_kind == ContentKind::DraftIssue {
json!({"draftIssueId":target.0,"title":write.item.title,"body":write.item.content})
} else {
json!({"id":target.0,"title":write.item.title,"body":write.item.content})
};
let data = self.graphql(operation, json!({"input":input})).await?;
let pointer = if content_kind == ContentKind::DraftIssue {
"/updateProjectV2DraftIssue/draftIssue"
} else {
"/updateIssue/issue"
};
let returned = data
.pointer(pointer)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub item update returned no item".into(),
})?;
if required_str(returned, "id")? != target.0 {
return Err(SourceError::Malformed {
message: "GitHub item update returned the wrong item".into(),
});
}
(
target.clone(),
NativeId(required_str(&item, "id")?.into()),
content_kind,
)
} else {
if !write.item.labels.is_empty() {
return Err(SourceError::Refused {
message: "GitHub draft items cannot represent labels".into(),
});
}
let data = self
.graphql(
graphql::CREATE_DRAFT,
json!({"input":{
"projectId":project_id,"title":write.item.title,"body":write.item.content
}}),
)
.await?;
let created = data
.pointer("/addProjectV2DraftIssue/projectItem")
.ok_or_else(|| SourceError::Malformed {
message: "GitHub draft creation returned no project item".into(),
})?;
(
NativeId(
required_str(created.pointer("/content").unwrap_or(&Value::Null), "id")?.into(),
),
NativeId(required_str(created, "id")?.into()),
ContentKind::DraftIssue,
)
};
let mut fallback = Vec::new();
let mut native = Vec::new();
for edge in &write.depends_on {
let same_source = edge
.to
.source()
.is_none_or(|source| source == self.name.as_str());
let far_id = edge
.to
.id()
.rsplit_once(':')
.map_or(edge.to.id(), |(_, id)| id);
let far_issue = if same_source {
let far = self
.board_and_item(Some(&NativeId(far_id.into())))
.await?
.1
.ok_or_else(|| SourceError::Refused {
message: format!("GitHub dependency item {far_id} was not found"),
})?;
match required_str(far.get("content").unwrap_or(&Value::Null), "__typename")? {
"Issue" => true,
"DraftIssue" | "PullRequest" => false,
other => {
return Err(SourceError::Malformed {
message: format!(
"GitHub dependency item has unknown content type {other}"
),
});
}
}
} else {
false
};
if content_kind == ContentKind::Issue && far_issue && edge.to.kind == ItemKind::Task {
native.push(far_id.to_owned());
} else {
fallback.push(edge.clone());
}
}
if content_kind == ContentKind::Issue {
let current = self.native_dependency_ids(&content_id).await?;
for (operation, far_id) in current
.iter()
.filter(|id| !native.contains(id))
.map(|id| (graphql::REMOVE_BLOCKED_BY, id))
.chain(
native
.iter()
.filter(|id| !current.contains(id))
.map(|id| (graphql::ADD_BLOCKED_BY, id)),
)
{
let data = self
.graphql(
operation,
json!({"input":{"issueId":content_id.0,"blockingIssueId":far_id}}),
)
.await?;
let root = if operation == graphql::ADD_BLOCKED_BY {
"addBlockedBy"
} else {
"removeBlockedBy"
};
let issue = data.pointer(&format!("/{root}/issue")).ok_or_else(|| {
SourceError::Malformed {
message: "GitHub dependency update returned no issue".into(),
}
})?;
let blocker = data
.pointer(&format!("/{root}/blockingIssue"))
.ok_or_else(|| SourceError::Malformed {
message: "GitHub dependency update returned no blocking issue".into(),
})?;
if required_str(issue, "id")? != content_id.0
|| required_str(blocker, "id")? != far_id
{
return Err(SourceError::Malformed {
message: "GitHub dependency update returned the wrong issues".into(),
});
}
}
}
let metadata_write = ItemWrite {
target: write.target.clone(),
item: write.item.clone(),
depends_on: fallback,
};
let storage = if content_kind == ContentKind::Issue {
RepositoryStorage::Native
} else {
RepositoryStorage::Recorded
};
let metadata = Self::task_metadata(&metadata_write, storage)?;
self.set_item_field(
project_id,
&item_id.0,
required_str(metadata_field, "id")?,
json!({"text":Value::Object(metadata.clone().into_iter().collect()).to_string()}),
)
.await?;
if let Some((field_id, option_id)) = status_selection {
self.set_item_field(
project_id,
&item_id.0,
&field_id,
json!({"singleSelectOptionId":option_id}),
)
.await?;
}
Ok(content_id)
}
async fn write_project(&self, write: &ItemWrite<Project>) -> Result<NativeId, SourceError> {
let project = self.project_value(None, 1).await?;
let id = NativeId(required_str(&project, "id")?.into());
if write.target.as_ref().is_some_and(|target| target != &id) {
return Err(SourceError::Refused {
message: format!(
"GitHub project {} was not found",
write.target.as_ref().unwrap().0
),
});
}
if !write.item.labels.is_empty() {
return Err(SourceError::Refused {
message: "GitHub Projects v2 cannot represent project labels".into(),
});
}
let mut metadata = write.item.metadata.clone();
metadata.insert(
Repository::METADATA_KEY.into(),
Value::Array(
write
.item
.repositories
.iter()
.map(|repository| Value::String(repository.as_str().to_owned()))
.collect(),
),
);
metadata.insert(
DependencyEdge::RECORDED_KEY.into(),
Value::Array(
write
.depends_on
.iter()
.map(|edge| endpoint_value(&edge.to))
.collect(),
),
);
let description = project_metadata_description(write.item.content.as_deref(), &metadata)?;
let data = self
.graphql(
graphql::UPDATE_PROJECT,
json!({"input":{
"projectId":id.0,"title":write.item.title,"shortDescription":description,
"closed":write.item.status.category == StatusCategory::Done
}}),
)
.await?;
let returned =
data.pointer("/updateProjectV2/projectV2")
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project update returned no project".into(),
})?;
if required_str(returned, "id")? != id.0 {
return Err(SourceError::Malformed {
message: "GitHub project update returned the wrong project".into(),
});
}
Ok(id)
}
}
const RECORDED_CURSOR: &str = "onetaskgraph.depends_on:";
fn recorded_offset(
cursor: Option<&str>,
direction: Direction,
) -> Result<Option<usize>, SourceError> {
cursor
.and_then(|cursor| cursor.strip_prefix(RECORDED_CURSOR))
.map(|offset| {
if direction != Direction::DependsOn {
return Err(SourceError::Config {
message: format!(
"{RECORDED_CURSOR}{offset} resumes recorded forward edges, which a \
reverse dependency read never issues; resume it in the direction \
that reported it"
),
});
}
offset.parse().map_err(|_| SourceError::Config {
message: format!("{RECORDED_CURSOR}{offset} is not a recorded-edge cursor"),
})
})
.transpose()
}
fn recorded_page(edges: Vec<DependencyEdge>, offset: usize, limit: usize) -> Page<DependencyEdge> {
let mut page = offset_page(edges, offset, limit.max(1));
page.next = page
.next
.map(|cursor| Cursor(format!("{RECORDED_CURSOR}{}", cursor.0)));
page
}
fn normalize_status_mapping(
mapping: BTreeMap<String, StatusCategory>,
) -> Result<BTreeMap<StatusName, StatusCategory>, SourceError> {
let mut normalized = BTreeMap::new();
for (name, category) in mapping {
if name.trim().is_empty() {
return Err(SourceError::Config {
message: "status_mapping contains a blank status name".into(),
});
}
let key = StatusName::new(&name);
if normalized.insert(key, category).is_some() {
return Err(SourceError::Config {
message: format!("status_mapping contains case-insensitive duplicate {name}"),
});
}
}
Ok(normalized)
}
fn valid_github_owner(owner: &str) -> bool {
!owner.is_empty()
&& owner.len() <= 39
&& !owner.starts_with('-')
&& !owner.ends_with('-')
&& !owner.contains("--")
&& owner
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'-')
}
fn valid_environment_name(name: &str) -> bool {
let mut bytes = name.bytes();
bytes
.next()
.is_some_and(|byte| byte.is_ascii_alphabetic() || byte == b'_')
&& bytes.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
struct StatusName(String);
impl StatusName {
fn new(name: &str) -> Self {
Self(name.to_lowercase())
}
}
fn required_str<'a>(value: &'a Value, field: &str) -> Result<&'a str, SourceError> {
value
.get(field)
.and_then(Value::as_str)
.ok_or_else(|| SourceError::Malformed {
message: format!("GitHub response is missing string field {field}"),
})
}
const METADATA_FIELD: &str = "onetaskgraph.metadata";
fn endpoint_value(endpoint: &DependencyEndpoint) -> Value {
json!({"id":endpoint.id(), "kind":match endpoint.kind { ItemKind::Task => "task", ItemKind::Project => "project" }})
}
fn metadata_field(field_values: &Value) -> Result<BTreeMap<String, Value>, SourceError> {
let nodes = field_values
.get("nodes")
.and_then(Value::as_array)
.ok_or_else(|| SourceError::Malformed {
message: "GitHub project item fieldValues.nodes is not an array".into(),
})?;
let Some(text) = nodes
.iter()
.find(|node| node.pointer("/field/name").and_then(Value::as_str) == Some(METADATA_FIELD))
.and_then(|node| node.get("text"))
else {
return Ok(BTreeMap::new());
};
text.as_str()
.ok_or_else(|| SourceError::Malformed {
message: format!("GitHub {METADATA_FIELD} field text is not a string"),
})
.and_then(|text| {
serde_json::from_str(text).map_err(|error| SourceError::Malformed {
message: format!(
"GitHub {METADATA_FIELD} field is not canonical JSON metadata: {error}"
),
})
})
}
fn repositories(content: &Value, field_values: &Value) -> Result<Vec<Repository>, SourceError> {
if let Some(origin) = content
.pointer("/repository/nameWithOwner")
.and_then(Value::as_str)
{
return Repository::try_from(format!("github.com/{origin}"))
.map(|repository| vec![repository])
.map_err(|message| SourceError::Malformed { message });
}
repositories_from_metadata(&metadata_field(field_values)?)
}
const PROJECT_METADATA_OPEN: &str = "<!-- onetaskgraph.metadata\n";
const PROJECT_METADATA_CLOSE: &str = "\n-->";
fn metadata_description(
description: Option<String>,
) -> Result<(Option<String>, BTreeMap<String, Value>), SourceError> {
let Some(description) = description else {
return Ok((None, BTreeMap::new()));
};
let Some(start) = description.rfind(PROJECT_METADATA_OPEN) else {
return Ok((Some(description), BTreeMap::new()));
};
let value_start = start + PROJECT_METADATA_OPEN.len();
let Some(relative_end) = description[value_start..].find(PROJECT_METADATA_CLOSE) else {
return Err(SourceError::Malformed {
message: "unterminated onetaskgraph metadata slot in GitHub project description".into(),
});
};
let value_end = value_start + relative_end;
if !description[value_end + PROJECT_METADATA_CLOSE.len()..]
.trim()
.is_empty()
{
return Ok((Some(description), BTreeMap::new()));
}
let metadata = serde_json::from_str(&description[value_start..value_end]).map_err(|error| {
SourceError::Malformed {
message: format!("invalid canonical JSON in GitHub project metadata slot: {error}"),
}
})?;
let visible = description[..start].trim_end();
Ok(((!visible.is_empty()).then(|| visible.into()), metadata))
}
fn project_metadata_description(
content: Option<&str>,
metadata: &BTreeMap<String, Value>,
) -> Result<Option<String>, SourceError> {
if metadata.is_empty() {
return Ok(content.filter(|value| !value.is_empty()).map(str::to_owned));
}
let encoded = Value::Object(metadata.clone().into_iter().collect()).to_string();
Ok(Some(format!(
"{}{}{}\n{}",
content.unwrap_or_default(),
if content.is_some_and(|value| !value.is_empty()) {
"\n\n"
} else {
""
},
PROJECT_METADATA_OPEN,
format_args!("{encoded}\n-->")
)))
}
fn repositories_from_metadata(
metadata: &BTreeMap<String, Value>,
) -> Result<Vec<Repository>, SourceError> {
Repository::from_metadata(metadata).map_err(|message| SourceError::Malformed { message })
}
fn required_bool(value: &Value, field: &str) -> Result<bool, SourceError> {
value
.get(field)
.and_then(Value::as_bool)
.ok_or_else(|| SourceError::Malformed {
message: format!("GitHub response is missing boolean field {field}"),
})
}
fn optional_str<'a>(value: &'a Value, field: &str) -> Result<Option<&'a str>, SourceError> {
match value.get(field) {
None | Some(Value::Null) => Ok(None),
Some(value) => value
.as_str()
.map(Some)
.ok_or_else(|| SourceError::Malformed {
message: format!("GitHub response field {field} is not a string or null"),
}),
}
}
fn optional_nodes<'a>(
connection: Option<&'a Value>,
name: &str,
) -> Result<Option<&'a Vec<Value>>, SourceError> {
match connection {
None | Some(Value::Null) => Ok(None),
Some(value) => value
.get("nodes")
.and_then(Value::as_array)
.map(Some)
.ok_or_else(|| SourceError::Malformed {
message: format!("GitHub {name}.nodes is not an array"),
}),
}
}
fn complete_connection(connection: &Value, name: &str) -> Result<(), SourceError> {
let page_info = connection
.get("pageInfo")
.ok_or_else(|| SourceError::Malformed {
message: format!("GitHub {name} has no pageInfo"),
})?;
if required_bool(page_info, "hasNextPage")? {
return Err(SourceError::Malformed {
message: format!(
"GitHub {name} exceeds the supported nested connection size of {NESTED_PAGE_SIZE}"
),
});
}
Ok(())
}
fn optional_time(value: &Value, field: &str) -> Result<Option<DateTime<Utc>>, SourceError> {
optional_str(value, field)?
.map(|timestamp| {
timestamp.parse().map_err(|error| SourceError::Malformed {
message: format!("GitHub response field {field} is not a timestamp: {error}"),
})
})
.transpose()
}
fn validate_page(page: &PageRequest) -> Result<(), SourceError> {
if page.limit == 0 {
Err(SourceError::Config {
message: "page limit must be at least 1".into(),
})
} else {
Ok(())
}
}
fn next_cursor(connection: &Value) -> Result<Option<Cursor>, SourceError> {
let page = connection
.get("pageInfo")
.filter(|value| value.is_object())
.ok_or_else(|| SourceError::Malformed {
message: "GitHub connection is missing pageInfo".into(),
})?;
if required_bool(page, "hasNextPage")? {
let cursor = required_str(page, "endCursor")?;
validate_cursor_progress(None, cursor)?;
Ok(Some(Cursor(cursor.into())))
} else {
Ok(None)
}
}
fn validate_cursor_progress(previous: Option<&str>, next: &str) -> Result<(), SourceError> {
if next.is_empty() || previous == Some(next) {
Err(SourceError::Malformed {
message: "GitHub pagination cursor is empty or did not advance".into(),
})
} else {
Ok(())
}
}
fn numeric_cursor(cursor: Option<&Cursor>) -> Result<usize, SourceError> {
cursor.map_or(Ok(0), |c| {
c.0.parse().map_err(|_| SourceError::Config {
message: "label cursor is invalid".into(),
})
})
}
fn offset_page<T>(mut items: Vec<T>, offset: usize, limit: usize) -> Page<T> {
if offset > items.len() {
return Page::last(vec![]);
}
let tail = items.split_off(offset);
let mut selected = tail;
let next = (selected.len() > limit).then(|| Cursor((offset + limit).to_string()));
selected.truncate(limit);
Page {
items: selected,
next,
}
}