use std::collections::{HashMap, HashSet};
use tracing::info;
use crate::collect::collector::CollectionStats;
use crate::collect::pm_adapter::{build_adapters, PmAdapter, PmSource, PmTicket};
use crate::core::config::Config;
use crate::core::db::{Database, WorkItemRow};
fn fetch_on_reference(config: &Config, source: PmSource) -> bool {
match source {
PmSource::Jira => config.jira.as_ref().is_some_and(|c| c.fetch_on_reference),
PmSource::GitHub => config.github.as_ref().is_some_and(|c| c.fetch_on_reference),
PmSource::AzureDevOps => config
.azure_devops_config()
.is_some_and(|c| c.fetch_on_reference),
PmSource::Linear => false,
}
}
pub(super) async fn fetch_and_persist_work_items(
db: &mut Database,
config: &Config,
stats: &mut CollectionStats,
) {
let adapters: Vec<Box<dyn PmAdapter>> = build_adapters(config)
.into_iter()
.filter(|a| fetch_on_reference(config, a.source()))
.collect();
if adapters.is_empty() {
return;
}
let commits = match read_commits(db) {
Ok(c) => c,
Err(e) => {
stats.fail_stage(format!("work_items: read commits failed: {e}"));
return;
}
};
for adapter in &adapters {
run_one_adapter(db, adapter.as_ref(), &commits, stats).await;
}
}
fn read_commits(db: &Database) -> crate::core::Result<Vec<(String, String)>> {
let conn = db.connection();
let mut stmt = conn.prepare("SELECT sha, message FROM commits")?;
let mapped = stmt.query_map([], |row| {
Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
})?;
let mut out = Vec::new();
for row in mapped {
out.push(row?);
}
Ok(out)
}
async fn run_one_adapter(
db: &mut Database,
adapter: &dyn PmAdapter,
commits: &[(String, String)],
stats: &mut CollectionStats,
) {
let name = adapter.name().to_string();
let mut commit_refs: HashMap<String, Vec<String>> = HashMap::new();
let mut unique: Vec<String> = Vec::new();
let mut seen: HashSet<String> = HashSet::new();
for (sha, message) in commits {
let refs = adapter.detect_ticket_refs(message);
if refs.is_empty() {
continue;
}
for r in &refs {
if seen.insert(r.clone()) {
unique.push(r.clone());
}
}
commit_refs.insert(sha.clone(), refs);
}
if unique.is_empty() {
info!(provider = %name, "no work-item references in commit messages");
return;
}
let id_refs: Vec<&str> = unique.iter().map(String::as_str).collect();
let results = adapter.fetch_tickets(&id_refs).await;
let mut tickets: Vec<PmTicket> = Vec::new();
let mut canonical: HashMap<String, String> = HashMap::new();
let mut first_error: Option<String> = None;
let mut error_count = 0usize;
for (requested, result) in unique.iter().zip(results) {
match result {
Ok(Some(ticket)) => {
canonical.insert(requested.clone(), ticket.id.clone());
tickets.push(ticket);
}
Ok(None) => {}
Err(e) => {
error_count += 1;
if first_error.is_none() {
first_error = Some(e.to_string());
}
}
}
}
if let Some(err) = first_error {
let message = format!(
"{name}: fetch work items failed for {error_count}/{} references: {err}",
unique.len()
);
if tickets.is_empty() {
stats.fail_stage(message);
} else {
stats.skip_item(message);
}
}
if tickets.is_empty() {
return;
}
let source = adapter.source().work_item_source();
match persist_tickets(db, source, &tickets, &commit_refs, &canonical) {
Ok(n) => info!(stored = n, provider = %name, source, "persisted work_items rows"),
Err(e) => stats.fail_stage(format!("{name}: store work_items failed: {e}")),
}
}
pub fn persist_tickets(
db: &mut Database,
source: &str,
tickets: &[PmTicket],
commit_refs: &HashMap<String, Vec<String>>,
canonical: &HashMap<String, String>,
) -> crate::core::Result<usize> {
use crate::core::db::work_items::{link_commit_work_item, upsert_work_item};
if tickets.is_empty() {
return Ok(0);
}
let tx = db.connection_mut().transaction()?;
for ticket in tickets {
upsert_work_item(&tx, &ticket_to_row(source, ticket))?;
}
let stored: HashSet<&str> = tickets.iter().map(|t| t.id.as_str()).collect();
for (sha, refs) in commit_refs {
for r in refs {
if let Some(id) = canonical.get(r) {
if stored.contains(id.as_str()) {
link_commit_work_item(&tx, sha, id, source)?;
}
}
}
}
tx.commit()?;
Ok(tickets.len())
}
fn ticket_to_row(source: &str, ticket: &PmTicket) -> WorkItemRow {
WorkItemRow {
id: ticket.id.clone(),
source: source.to_string(),
title: ticket.title.clone(),
status: ticket.status.clone(),
item_type: ticket.ticket_type.clone(),
tags: if ticket.labels.is_empty() {
None
} else {
Some(ticket.labels.join(","))
},
project: ticket.project.clone(),
url: ticket.url.clone(),
raw_json: serde_json::to_string(&ticket.raw).ok(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::core::config::{
AzureDevOpsConfig, GithubConfig, JiraConfig, LinearConfig, PmConfig,
};
fn ticket(id: &str) -> PmTicket {
PmTicket {
id: id.to_string(),
title: format!("Title for {id}"),
status: "Active".to_string(),
ticket_type: "Bug".to_string(),
labels: vec!["a".to_string(), "b".to_string()],
url: Some(format!("https://example.test/{id}")),
project: Some("P".to_string()),
source: PmSource::AzureDevOps,
raw: serde_json::json!({ "id": id }),
}
}
fn row_count(db: &Database, source: &str) -> i64 {
db.connection()
.query_row(
"SELECT COUNT(*) FROM work_items WHERE source = ?1",
[source],
|r| r.get(0),
)
.expect("count work_items")
}
fn insert_commit(db: &Database, sha: &str, message: &str) {
db.connection()
.execute(
"INSERT INTO commits (sha, author_name, author_email, timestamp, message, \
repository) \
VALUES (?1, 'A', 'a@x', '2026-01-01T00:00:00Z', ?2, 'repo')",
rusqlite::params![sha, message],
)
.expect("insert commit");
}
fn unauthenticated_azdo() -> AzureDevOpsConfig {
AzureDevOpsConfig {
organization_url: "https://dev.azure.com/myorg".into(),
pat: String::new(),
project: Some("P".into()),
projects: vec![],
ticket_regex: r"(?i)\bAB#(\d+)\b".into(),
team_keys: vec![],
fetch_on_reference: true,
fetch_prs: false,
}
}
#[test]
fn persist_tickets_writes_rows_and_links() {
let mut db = Database::open_in_memory().expect("db");
let refs = HashMap::from([("sha1".to_string(), vec!["AB#1".to_string()])]);
let canonical = HashMap::from([("AB#1".to_string(), "AB#1".to_string())]);
let n = persist_tickets(&mut db, "azdo", &[ticket("AB#1")], &refs, &canonical)
.expect("persist");
assert_eq!(n, 1);
assert_eq!(row_count(&db, "azdo"), 1);
let (title, status, item_type, tags, project, url): (
String,
String,
String,
Option<String>,
Option<String>,
Option<String>,
) = db
.connection()
.query_row(
"SELECT title, status, item_type, tags, project, url FROM work_items \
WHERE id = ?1 AND source = 'azdo'",
["AB#1"],
|r| {
Ok((
r.get(0)?,
r.get(1)?,
r.get(2)?,
r.get(3)?,
r.get(4)?,
r.get(5)?,
))
},
)
.expect("query row");
assert_eq!(title, "Title for AB#1");
assert_eq!(status, "Active");
assert_eq!(item_type, "Bug");
assert_eq!(tags.as_deref(), Some("a,b"), "labels become the tags CSV");
assert_eq!(project.as_deref(), Some("P"));
assert_eq!(url.as_deref(), Some("https://example.test/AB#1"));
let linked: i64 = db
.connection()
.query_row(
"SELECT COUNT(*) FROM commit_work_items \
WHERE work_item_id = 'AB#1' AND work_item_source = 'azdo'",
[],
|r| r.get(0),
)
.expect("count links");
assert_eq!(linked, 1);
}
#[test]
fn persist_tickets_maps_a_bare_ref_to_its_canonical_id() {
let mut db = Database::open_in_memory().expect("db");
let refs = HashMap::from([("sha1".to_string(), vec!["42".to_string()])]);
let canonical = HashMap::from([("42".to_string(), "AB#42".to_string())]);
persist_tickets(&mut db, "azdo", &[ticket("AB#42")], &refs, &canonical).expect("persist");
let linked: String = db
.connection()
.query_row(
"SELECT work_item_id FROM commit_work_items WHERE commit_sha = 'sha1'",
[],
|r| r.get(0),
)
.expect("query link");
assert_eq!(linked, "AB#42");
}
#[test]
fn persist_tickets_skips_unresolved_refs() {
let mut db = Database::open_in_memory().expect("db");
let refs = HashMap::from([(
"sha1".to_string(),
vec!["AB#1".to_string(), "AB#9".to_string()],
)]);
let canonical = HashMap::from([("AB#1".to_string(), "AB#1".to_string())]);
persist_tickets(&mut db, "azdo", &[ticket("AB#1")], &refs, &canonical).expect("persist");
let links: i64 = db
.connection()
.query_row("SELECT COUNT(*) FROM commit_work_items", [], |r| r.get(0))
.expect("count links");
assert_eq!(links, 1, "only the resolved reference is linked");
}
#[test]
fn persist_tickets_is_idempotent() {
let mut db = Database::open_in_memory().expect("db");
let refs = HashMap::from([("sha1".to_string(), vec!["AB#1".to_string()])]);
let canonical = HashMap::from([("AB#1".to_string(), "AB#1".to_string())]);
persist_tickets(&mut db, "azdo", &[ticket("AB#1")], &refs, &canonical).expect("first");
let mut updated = ticket("AB#1");
updated.status = "Closed".to_string();
persist_tickets(&mut db, "azdo", &[updated], &refs, &canonical).expect("second");
assert_eq!(row_count(&db, "azdo"), 1);
let status: String = db
.connection()
.query_row(
"SELECT status FROM work_items WHERE id = 'AB#1' AND source = 'azdo'",
[],
|r| r.get(0),
)
.expect("query status");
assert_eq!(
status, "Closed",
"re-running refreshes rather than duplicates"
);
let links: i64 = db
.connection()
.query_row("SELECT COUNT(*) FROM commit_work_items", [], |r| r.get(0))
.expect("count links");
assert_eq!(links, 1);
}
#[test]
fn persist_tickets_reports_write_failure() {
let mut db = Database::open_in_memory().expect("db");
db.connection()
.execute("DROP TABLE work_items", [])
.expect("drop table");
let err = persist_tickets(
&mut db,
"azdo",
&[ticket("AB#1")],
&HashMap::new(),
&HashMap::new(),
)
.expect_err("write against a missing table must fail");
assert!(
err.to_string().contains("work_items"),
"error names the failing table: {err}"
);
}
#[test]
fn persist_tickets_rolls_back_a_partial_write() {
let mut db = Database::open_in_memory().expect("db");
db.connection()
.execute("DROP TABLE commit_work_items", [])
.expect("drop join table");
let refs = HashMap::from([("sha1".to_string(), vec!["AB#1".to_string()])]);
let canonical = HashMap::from([("AB#1".to_string(), "AB#1".to_string())]);
let err = persist_tickets(&mut db, "azdo", &[ticket("AB#1")], &refs, &canonical)
.expect_err("link against a missing table must fail");
assert!(
err.to_string().contains("commit_work_items"),
"error names the failing table: {err}"
);
assert_eq!(
row_count(&db, "azdo"),
0,
"the upsert that already succeeded must roll back with the transaction"
);
}
#[test]
fn persist_tickets_no_op() {
let mut db = Database::open_in_memory().expect("db");
assert_eq!(
persist_tickets(&mut db, "azdo", &[], &HashMap::new(), &HashMap::new())
.expect("persist"),
0
);
assert_eq!(row_count(&db, "azdo"), 0);
}
#[test]
fn linear_is_never_fetched_here() {
let config = Config {
linear: Some(LinearConfig {
api_key: Some("k".into()),
fetch_on_reference: true,
..Default::default()
}),
..Default::default()
};
assert!(!fetch_on_reference(&config, PmSource::Linear));
}
#[test]
fn omitting_fetch_on_reference_in_yaml_enables_all_four_providers() {
let yaml = r#"
github:
repo: "o/r"
jira:
url: "https://example.atlassian.net"
linear:
api_key: "k"
pm:
azure_devops:
organization_url: "https://dev.azure.com/myorg"
pat: "x"
"#;
let cfg: Config = serde_yaml::from_str(yaml).expect("yaml parses");
assert!(fetch_on_reference(&cfg, PmSource::GitHub), "github");
assert!(fetch_on_reference(&cfg, PmSource::Jira), "jira");
assert!(fetch_on_reference(&cfg, PmSource::AzureDevOps), "azdo");
assert!(
cfg.linear.expect("linear parsed").fetch_on_reference,
"linear's own flag defaults on; its pipeline reads it directly"
);
}
#[test]
fn setting_fetch_on_reference_false_opts_a_provider_out() {
let yaml = r#"
github:
repo: "o/r"
fetch_on_reference: false
jira:
url: "https://example.atlassian.net"
fetch_on_reference: false
pm:
azure_devops:
organization_url: "https://dev.azure.com/myorg"
pat: "x"
fetch_on_reference: false
"#;
let cfg: Config = serde_yaml::from_str(yaml).expect("yaml parses");
assert!(!fetch_on_reference(&cfg, PmSource::GitHub));
assert!(!fetch_on_reference(&cfg, PmSource::Jira));
assert!(!fetch_on_reference(&cfg, PmSource::AzureDevOps));
}
#[test]
fn an_unconfigured_provider_is_never_fetched() {
let cfg = Config::default();
assert!(!fetch_on_reference(&cfg, PmSource::GitHub));
assert!(!fetch_on_reference(&cfg, PmSource::Jira));
assert!(!fetch_on_reference(&cfg, PmSource::AzureDevOps));
}
#[test]
fn derived_default_is_not_the_yaml_default() {
assert!(
!GithubConfig::default().fetch_on_reference,
"derived Default gives bool::default()"
);
assert!(!JiraConfig::default().fetch_on_reference);
assert!(!LinearConfig::default().fetch_on_reference);
let from_yaml: GithubConfig = serde_yaml::from_str("repo: o/r").expect("parses");
assert!(
from_yaml.fetch_on_reference,
"the YAML path is the one that carries the decision"
);
}
#[tokio::test]
async fn a_total_fetch_failure_is_a_stage_failure() {
let mut db = Database::open_in_memory().expect("db");
insert_commit(&db, "sha1", "fix: AB#42 broken thing");
let config = Config {
pm: Some(PmConfig {
azure_devops: Some(unauthenticated_azdo()),
}),
..Default::default()
};
let mut stats = CollectionStats::default();
fetch_and_persist_work_items(&mut db, &config, &mut stats).await;
assert_eq!(
stats.stage_failures().len(),
1,
"a provider that fetched nothing must reach the exit code: {:?}",
stats.errors
);
assert_eq!(row_count(&db, "azdo"), 0);
}
#[tokio::test]
async fn no_enabled_provider_is_a_noop() {
let mut db = Database::open_in_memory().expect("db");
insert_commit(&db, "sha1", "fix: AB#42 broken thing");
let mut stats = CollectionStats::default();
fetch_and_persist_work_items(&mut db, &Config::default(), &mut stats).await;
assert!(stats.errors.is_empty(), "faults: {:?}", stats.errors);
}
#[tokio::test]
async fn github_issues_reach_work_items() {
use crate::collect::pm_adapter::GitHubAdapter;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/repos/o/r/issues/42"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"number": 42,
"title": "Crash on startup",
"state": "open",
"html_url": "https://github.com/o/r/issues/42",
"labels": [{ "name": "bug" }],
"body": "steps to reproduce",
})))
.mount(&server)
.await;
let gh_cfg = GithubConfig {
repo: Some("o/r".into()),
fetch_on_reference: true,
..Default::default()
};
let client = crate::collect::github::GitHubClient::new(&gh_cfg)
.expect("client")
.with_api_base(server.uri());
let adapter = GitHubAdapter::with_ticket_regex(client, gh_cfg.ticket_regex.as_deref());
let mut db = Database::open_in_memory().expect("db");
let commits = vec![("sha1".to_string(), "fix: crash, closes #42".to_string())];
let mut stats = CollectionStats::default();
run_one_adapter(&mut db, &adapter, &commits, &mut stats).await;
assert!(stats.errors.is_empty(), "faults: {:?}", stats.errors);
assert_eq!(row_count(&db, "github"), 1);
let (id, title, status, tags): (String, String, String, Option<String>) = db
.connection()
.query_row(
"SELECT id, title, status, tags FROM work_items WHERE source = 'github'",
[],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
)
.expect("query row");
assert_eq!(id, "#42", "the form `extract_ticket_id` produces");
assert_eq!(title, "Crash on startup");
assert_eq!(status, "open");
assert_eq!(tags.as_deref(), Some("bug"));
let linked: String = db
.connection()
.query_row(
"SELECT commit_sha FROM commit_work_items WHERE work_item_source = 'github'",
[],
|r| r.get(0),
)
.expect("query link");
assert_eq!(linked, "sha1");
}
#[tokio::test]
async fn jira_issues_reach_work_items() {
use crate::collect::pm_adapter::JiraAdapter;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/rest/api/3/field"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/rest/api/3/issue/PROJ-7"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"key": "PROJ-7",
"fields": {
"summary": "Rework the importer",
"status": { "name": "In Progress" },
"issuetype": { "name": "Story" },
},
})))
.mount(&server)
.await;
let jira_cfg = JiraConfig {
url: Some(server.uri()),
username: Some("u".into()),
token: Some("t".into()),
fetch_on_reference: true,
..Default::default()
};
let client = crate::collect::jira::JiraClient::new(&jira_cfg).expect("client");
let adapter = JiraAdapter::with_ticket_regex(client, jira_cfg.ticket_regex.as_deref());
let mut db = Database::open_in_memory().expect("db");
let commits = vec![("sha1".to_string(), "PROJ-7: rework importer".to_string())];
let mut stats = CollectionStats::default();
run_one_adapter(&mut db, &adapter, &commits, &mut stats).await;
assert!(stats.errors.is_empty(), "faults: {:?}", stats.errors);
assert_eq!(row_count(&db, "jira"), 1);
let (id, title, status, item_type): (String, String, String, String) = db
.connection()
.query_row(
"SELECT id, title, status, item_type FROM work_items WHERE source = 'jira'",
[],
|r| Ok((r.get(0)?, r.get(1)?, r.get(2)?, r.get(3)?)),
)
.expect("query row");
assert_eq!(id, "PROJ-7");
assert_eq!(title, "Rework the importer");
assert_eq!(status, "In Progress");
assert_eq!(item_type, "Story");
}
#[tokio::test]
async fn a_partial_fetch_failure_skips_items() {
use crate::collect::pm_adapter::GitHubAdapter;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/repos/o/r/issues/1"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"number": 1,
"title": "Fine",
"state": "open",
"html_url": "https://github.com/o/r/issues/1",
"labels": [],
})))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/repos/o/r/issues/2"))
.respond_with(ResponseTemplate::new(500))
.mount(&server)
.await;
let gh_cfg = GithubConfig {
repo: Some("o/r".into()),
fetch_on_reference: true,
..Default::default()
};
let client = crate::collect::github::GitHubClient::new(&gh_cfg)
.expect("client")
.with_api_base(server.uri());
let adapter = GitHubAdapter::new(client);
let mut db = Database::open_in_memory().expect("db");
let commits = vec![("sha1".to_string(), "touches #1 and #2".to_string())];
let mut stats = CollectionStats::default();
run_one_adapter(&mut db, &adapter, &commits, &mut stats).await;
assert_eq!(stats.errors.len(), 1, "faults: {:?}", stats.errors);
assert!(
stats.stage_failures().is_empty(),
"a partial fetch keeps the exit code at zero: {:?}",
stats.errors
);
assert_eq!(
row_count(&db, "github"),
1,
"the issue that resolved is written"
);
}
#[tokio::test]
async fn a_real_slug_whose_fetch_fails_still_fails_the_stage_closed() {
use crate::collect::pm_adapter::GitHubAdapter;
use wiremock::matchers::method;
use wiremock::{Mock, MockServer, ResponseTemplate};
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(404))
.mount(&server)
.await;
let gh_cfg = GithubConfig {
repo: Some("bobmatnyc/trusty-tools".into()),
fetch_on_reference: true,
..Default::default()
};
let client = crate::collect::github::GitHubClient::new(&gh_cfg)
.expect("client")
.with_api_base(server.uri());
let adapter = GitHubAdapter::new(client);
let mut db = Database::open_in_memory().expect("db");
let commits = vec![("sha1".to_string(), "fix: closes #42".to_string())];
let mut stats = CollectionStats::default();
run_one_adapter(&mut db, &adapter, &commits, &mut stats).await;
assert_eq!(
stats.stage_failures().len(),
1,
"a total fetch failure against a real slug must reach the exit code: {:?}",
stats.errors
);
assert_eq!(row_count(&db, "github"), 0);
}
#[tokio::test]
async fn no_references_is_a_noop() {
let mut db = Database::open_in_memory().expect("db");
insert_commit(&db, "sha1", "chore: nothing to see");
let config = Config {
pm: Some(PmConfig {
azure_devops: Some(unauthenticated_azdo()),
}),
..Default::default()
};
let mut stats = CollectionStats::default();
fetch_and_persist_work_items(&mut db, &config, &mut stats).await;
assert!(stats.errors.is_empty(), "faults: {:?}", stats.errors);
}
}