diff --git a/crates/api/src/github_access/cached.rs b/crates/api/src/github_access/cached.rs index 49196f8..7bc4576 100644 --- a/crates/api/src/github_access/cached.rs +++ b/crates/api/src/github_access/cached.rs @@ -8,10 +8,13 @@ use super::types::{AccessError, AccessResult, GithubAccessChecker}; pub(super) const GITHUB_API_BASE_URL: &str = "https://api.github.com"; pub(super) const CACHE_TYPE_REPO_IDS: &str = "accessible_repo_ids"; -pub(super) const CACHE_TTL_SECONDS: i64 = 600; // 10 minutes -pub(super) const STALE_MAX_SECONDS: i64 = 3600; // 1 hour +// Issue #105: relaxed cache lifetimes. The repositories table is reconciled +// independently by webhook + worker sync, so we can serve cached user→repo +// mappings for longer without risking long-term staleness. +pub(super) const CACHE_TTL_SECONDS: i64 = 3600; // 1 hour (valid) +pub(super) const STALE_MAX_SECONDS: i64 = 86_400; // 24 hours (stale-while-error) pub(super) const SYNC_CACHE_TYPE: &str = "installation_repos_sync"; -pub(super) const SYNC_TTL_SECONDS: i64 = 600; // 10 minutes +pub(super) const SYNC_TTL_SECONDS: i64 = 1800; // 30 minutes /// Caching decorator that stores `list_accessible_repo_ids` results in PostgreSQL. /// Falls back to stale cache on rate-limit errors (stale-while-error). @@ -148,8 +151,11 @@ impl GithubAccessChecker for CachedGithubAccessChecker { } Ok(result) } - Err(ref e @ AccessError::RateLimited) => { - // 5. On RateLimited only – try stale cache + Err(e) => { + // 5. On any error – try stale cache so a transient GitHub failure + // (rate limit, expired user token, upstream blip) does not break the + // repository list. The repositories table itself is reconciled by + // webhook + worker sync, so the cached id set is still meaningful. let stale_duration = chrono::Duration::seconds(STALE_MAX_SECONDS); if let Ok(Some(stale)) = boardflow_db::queries::github_api_cache::get_stale_cache( &self.pool, @@ -163,15 +169,11 @@ impl GithubAccessChecker for CachedGithubAccessChecker { tracing::warn!( user_id = %user_id, error = %format!("{e:?}"), - "using stale cache for accessible_repo_ids due to rate limiting" + "using stale cache for accessible_repo_ids after GitHub error" ); return Ok(Some(ids)); } } - Err(AccessError::RateLimited) - } - Err(e) => { - // 6. Non-rate-limit errors (TokenExpired, Upstream) – propagate directly Err(e) } } diff --git a/crates/api/src/routes/webhook.rs b/crates/api/src/routes/webhook.rs index a7713de..90e95bf 100644 --- a/crates/api/src/routes/webhook.rs +++ b/crates/api/src/routes/webhook.rs @@ -130,6 +130,19 @@ async fn handle_installation_event(pool: &PgPool, body: &[u8]) -> StatusCode { } }; + if let Err(err) = boardflow_db::queries::github_installation_sync_state::upsert_webhook_seen( + pool, + event.installation.id, + ) + .await + { + tracing::warn!( + error = %err, + installation_id = event.installation.id, + "failed to stamp installation webhook_seen_at" + ); + } + match event.action.as_str() { "created" => { for repo in &event.repositories { @@ -190,6 +203,19 @@ async fn handle_installation_repositories_event(pool: &PgPool, body: &[u8]) -> S } }; + if let Err(err) = boardflow_db::queries::github_installation_sync_state::upsert_webhook_seen( + pool, + event.installation.id, + ) + .await + { + tracing::warn!( + error = %err, + installation_id = event.installation.id, + "failed to stamp installation webhook_seen_at" + ); + } + match event.action.as_str() { "added" => { for repo in &event.repositories_added { diff --git a/crates/api/tests/github_cache_test.rs b/crates/api/tests/github_cache_test.rs index 89d350e..9f4116b 100644 --- a/crates/api/tests/github_cache_test.rs +++ b/crates/api/tests/github_cache_test.rs @@ -279,10 +279,11 @@ async fn test_cleanup_expired_cache() { }; let user_id = create_test_user_with_token(&pool, "gho_cache_test_8").await; - // Insert expired >1 hour ago (should be cleaned) + // Issue #105: cleanup preserves entries up to 24h past expiration so they + // remain available as stale-while-error fallbacks. Insert expired >24h ago. sqlx::query( "INSERT INTO github_api_cache (user_id, cache_type, value_json, expires_at, created_at, updated_at) \ - VALUES ($1, 'cleanup_type', '[1]'::jsonb, NOW() - INTERVAL '2 hours', NOW(), NOW()) \ + VALUES ($1, 'cleanup_type', '[1]'::jsonb, NOW() - INTERVAL '36 hours', NOW(), NOW()) \ ON CONFLICT (user_id, cache_type) DO UPDATE SET value_json = EXCLUDED.value_json, expires_at = EXCLUDED.expires_at", ) .bind(user_id) @@ -296,6 +297,40 @@ async fn test_cleanup_expired_cache() { assert!(deleted >= 1); } +#[tokio::test] +#[serial] +async fn test_cleanup_preserves_recent_stale_cache() { + let Some(pool) = setup_pool().await else { + return; + }; + let user_id = create_test_user_with_token(&pool, "gho_cache_cleanup_preserve").await; + + // Expired 2 hours ago — within the 24h stale window, must survive cleanup. + sqlx::query( + "INSERT INTO github_api_cache (user_id, cache_type, value_json, expires_at, created_at, updated_at) \ + VALUES ($1, 'preserve_type', '[2]'::jsonb, NOW() - INTERVAL '2 hours', NOW(), NOW()) \ + ON CONFLICT (user_id, cache_type) DO UPDATE SET value_json = EXCLUDED.value_json, expires_at = EXCLUDED.expires_at", + ) + .bind(user_id) + .execute(&pool) + .await + .unwrap(); + + boardflow_db::queries::github_api_cache::cleanup_expired_cache(&pool) + .await + .unwrap(); + + let stale = boardflow_db::queries::github_api_cache::get_stale_cache( + &pool, + user_id, + "preserve_type", + chrono::Duration::hours(24), + ) + .await + .unwrap(); + assert_eq!(stale, Some(serde_json::json!([2]))); +} + // ─── CachedGithubAccessChecker integration tests ───────────────────────────── /// Test: CachedGithubAccessChecker.invalidate_cache removes all user's cache @@ -450,10 +485,11 @@ async fn test_cached_checker_stale_fallback_with_mock_inner_rate_limited() { assert_eq!(result, Ok(Some(vec![5001, 5002, 5003]))); } -/// Test: TokenExpired from inner does NOT use stale cache – propagates error +/// Issue #105: TokenExpired from inner falls back to stale cache so a +/// transient GitHub token failure does not break the repository list. #[tokio::test] #[serial] -async fn test_cached_checker_token_expired_no_stale_fallback() { +async fn test_cached_checker_token_expired_stale_fallback() { use boardflow_api::github_access::TokenExpiredGithubAccessChecker; let Some(pool) = setup_pool().await else { @@ -478,14 +514,13 @@ async fn test_cached_checker_token_expired_no_stale_fallback() { let checker = CachedGithubAccessChecker::with_inner(inner, pool.clone(), None); let result = checker.list_accessible_repo_ids(token).await; - // Should propagate TokenExpired error, NOT return stale cache - assert_eq!(result, Err(AccessError::TokenExpired)); + assert_eq!(result, Ok(Some(vec![6001, 6002]))); } -/// Test: Upstream error from inner does NOT use stale cache – propagates error +/// Issue #105: Upstream error from inner falls back to stale cache. #[tokio::test] #[serial] -async fn test_cached_checker_upstream_error_no_stale_fallback() { +async fn test_cached_checker_upstream_error_stale_fallback() { use boardflow_api::github_access::UpstreamErrorGithubAccessChecker; let Some(pool) = setup_pool().await else { @@ -510,7 +545,46 @@ async fn test_cached_checker_upstream_error_no_stale_fallback() { let checker = CachedGithubAccessChecker::with_inner(inner, pool.clone(), None); let result = checker.list_accessible_repo_ids(token).await; - // Should propagate Upstream error, NOT return stale cache + assert_eq!(result, Ok(Some(vec![7001, 7002]))); +} + +/// Issue #105: TokenExpired with no stale cache still propagates the error +/// (the caller decides — e.g. list_repositories — but the cache layer must +/// not fabricate an empty list). +#[tokio::test] +#[serial] +async fn test_cached_checker_token_expired_no_stale_returns_error() { + use boardflow_api::github_access::TokenExpiredGithubAccessChecker; + + let Some(pool) = setup_pool().await else { + return; + }; + let token = "gho_mock_token_expired_no_stale"; + let _user_id = create_test_user_with_token(&pool, token).await; + + let inner: Arc = Arc::new(TokenExpiredGithubAccessChecker); + let checker = CachedGithubAccessChecker::with_inner(inner, pool.clone(), None); + + let result = checker.list_accessible_repo_ids(token).await; + assert_eq!(result, Err(AccessError::TokenExpired)); +} + +/// Issue #105: Upstream error with no stale cache still propagates the error. +#[tokio::test] +#[serial] +async fn test_cached_checker_upstream_error_no_stale_returns_error() { + use boardflow_api::github_access::UpstreamErrorGithubAccessChecker; + + let Some(pool) = setup_pool().await else { + return; + }; + let token = "gho_mock_upstream_error_no_stale"; + let _user_id = create_test_user_with_token(&pool, token).await; + + let inner: Arc = Arc::new(UpstreamErrorGithubAccessChecker); + let checker = CachedGithubAccessChecker::with_inner(inner, pool.clone(), None); + + let result = checker.list_accessible_repo_ids(token).await; assert!(matches!(result, Err(AccessError::Upstream(_)))); } diff --git a/crates/api/tests/webhook_test.rs b/crates/api/tests/webhook_test.rs index 2985b24..23c00e0 100644 --- a/crates/api/tests/webhook_test.rs +++ b/crates/api/tests/webhook_test.rs @@ -457,6 +457,95 @@ async fn test_webhook_repos_removed_different_installation() { ); } +// --- Issue #105: webhook events stamp installation sync_state --- + +#[tokio::test] +#[serial] +async fn test_webhook_installation_created_stamps_sync_state() { + let Some(pool) = setup_pool().await else { + return; + }; + + let installation_id = rand_i64(); + let body = serde_json::json!({ + "action": "created", + "installation": { "id": installation_id }, + "repositories": [] + }); + let body_bytes = serde_json::to_vec(&body).unwrap(); + let signature = compute_signature(WEBHOOK_SECRET, &body_bytes); + + let app = create_test_app(pool.clone()); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/v1/github/webhook") + .header("Content-Type", "application/json") + .header("X-GitHub-Event", "installation") + .header("X-Hub-Signature-256", &signature) + .body(Body::from(body_bytes)) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + + let state = boardflow_db::queries::github_installation_sync_state::find_by_installation_id( + &pool, + installation_id, + ) + .await + .unwrap() + .expect("sync_state row should exist after webhook"); + assert!(state.webhook_seen_at.is_some()); +} + +#[tokio::test] +#[serial] +async fn test_webhook_repos_added_stamps_sync_state() { + let Some(pool) = setup_pool().await else { + return; + }; + + let installation_id = rand_i64(); + let body = serde_json::json!({ + "action": "added", + "installation": { "id": installation_id }, + "repositories_added": [], + "repositories_removed": [] + }); + let body_bytes = serde_json::to_vec(&body).unwrap(); + let signature = compute_signature(WEBHOOK_SECRET, &body_bytes); + + let app = create_test_app(pool.clone()); + let response = app + .oneshot( + Request::builder() + .method("POST") + .uri("/api/v1/github/webhook") + .header("Content-Type", "application/json") + .header("X-GitHub-Event", "installation_repositories") + .header("X-Hub-Signature-256", &signature) + .body(Body::from(body_bytes)) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::OK); + + let state = boardflow_db::queries::github_installation_sync_state::find_by_installation_id( + &pool, + installation_id, + ) + .await + .unwrap() + .expect("sync_state row should exist after webhook"); + assert!(state.webhook_seen_at.is_some()); +} + // --- Test: unknown event returns 200 --- #[tokio::test] diff --git a/crates/config/src/worker.rs b/crates/config/src/worker.rs index 9b51244..fa34b58 100644 --- a/crates/config/src/worker.rs +++ b/crates/config/src/worker.rs @@ -8,6 +8,10 @@ pub struct WorkerConfig { pub poll_interval_secs: u64, pub timeout_sweep_interval_secs: u64, pub cache_cleanup_interval_secs: u64, + pub installation_sync_interval_secs: u64, + pub installation_sync_stale_after_secs: u64, + pub installation_sync_min_interval_secs: u64, + pub installation_sync_max_per_sweep: u64, pub github_app_id: Option, pub github_private_key_pem: Option, pub app_domain: String, @@ -33,12 +37,34 @@ impl WorkerConfig { }); } + let installation_sync_interval_secs = + parse_env_or("INSTALLATION_SYNC_INTERVAL_SECS", 1800u64)?; + if installation_sync_interval_secs == 0 { + return Err(ConfigError::InvalidValue { + var: "INSTALLATION_SYNC_INTERVAL_SECS".to_string(), + reason: "must be greater than 0".to_string(), + }); + } + Ok(Self { db: DatabaseConfig::from_env()?, s3: S3Config::from_env(), poll_interval_secs: parse_env_or("POLL_INTERVAL_SECS", 2u64)?, timeout_sweep_interval_secs: parse_env_or("TIMEOUT_SWEEP_INTERVAL_SECS", 60u64)?, cache_cleanup_interval_secs, + installation_sync_interval_secs, + installation_sync_stale_after_secs: parse_env_or( + "INSTALLATION_SYNC_STALE_AFTER_SECS", + 86_400u64, + )?, + installation_sync_min_interval_secs: parse_env_or( + "INSTALLATION_SYNC_MIN_INTERVAL_SECS", + 3600u64, + )?, + installation_sync_max_per_sweep: parse_env_or( + "INSTALLATION_SYNC_MAX_PER_SWEEP", + 50u64, + )?, github_app_id, github_private_key_pem: optional_env("GITHUB_PRIVATE_KEY_PEM"), app_domain: std::env::var("BOARDFLOW_APP_DOMAIN").unwrap_or_else(|_| { diff --git a/crates/db/migrations/20260504000000_add_installation_sync_state.down.sql b/crates/db/migrations/20260504000000_add_installation_sync_state.down.sql new file mode 100644 index 0000000..095dad0 --- /dev/null +++ b/crates/db/migrations/20260504000000_add_installation_sync_state.down.sql @@ -0,0 +1,3 @@ +DROP INDEX IF EXISTS idx_repositories_updated_at_github_id; +DROP INDEX IF EXISTS idx_github_installation_sync_state_stale; +DROP TABLE IF EXISTS github_installation_sync_state; diff --git a/crates/db/migrations/20260504000000_add_installation_sync_state.up.sql b/crates/db/migrations/20260504000000_add_installation_sync_state.up.sql new file mode 100644 index 0000000..b7803db --- /dev/null +++ b/crates/db/migrations/20260504000000_add_installation_sync_state.up.sql @@ -0,0 +1,23 @@ +-- Issue #105: per-installation sync state for webhook freshness / worker reconciliation +CREATE TABLE github_installation_sync_state ( + installation_id BIGINT PRIMARY KEY, + webhook_seen_at TIMESTAMPTZ, + last_sync_started_at TIMESTAMPTZ, + last_sync_completed_at TIMESTAMPTZ, + last_sync_status TEXT CONSTRAINT github_installation_sync_state_status_check + CHECK (last_sync_status IN ('success', 'failed')), + last_error TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- Worker periodically picks installations needing reconciliation. The selector +-- is "webhook_seen_at IS NULL OR webhook_seen_at < cutoff OR last_sync_status = 'failed'", +-- so an index on webhook_seen_at + last_sync_status keeps that scan cheap as +-- installations accumulate. +CREATE INDEX idx_github_installation_sync_state_stale + ON github_installation_sync_state (webhook_seen_at NULLS FIRST, last_sync_status); + +-- Issue #105: list_repositories paginates by (updated_at DESC, github_repository_id DESC). +-- Adding a composite index lets the cursor pagination be index-only. +CREATE INDEX idx_repositories_updated_at_github_id + ON repositories (updated_at DESC, github_repository_id DESC); diff --git a/crates/db/src/queries/github_api_cache.rs b/crates/db/src/queries/github_api_cache.rs index a37ab38..94bad8e 100644 --- a/crates/db/src/queries/github_api_cache.rs +++ b/crates/db/src/queries/github_api_cache.rs @@ -86,8 +86,11 @@ pub async fn delete_cache( pub async fn cleanup_expired_cache( executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>, ) -> Result { + // Issue #105: the stale-while-error window for accessible_repo_ids is 24h + // past expiration. Keep entries around for at least that long so the + // cleanup job does not delete data still usable as fallback. let result = - sqlx::query("DELETE FROM github_api_cache WHERE expires_at < NOW() - INTERVAL '1 hour'") + sqlx::query("DELETE FROM github_api_cache WHERE expires_at < NOW() - INTERVAL '24 hours'") .execute(executor) .await?; Ok(result.rows_affected()) diff --git a/crates/db/src/queries/github_installation_sync_state.rs b/crates/db/src/queries/github_installation_sync_state.rs new file mode 100644 index 0000000..415b320 --- /dev/null +++ b/crates/db/src/queries/github_installation_sync_state.rs @@ -0,0 +1,114 @@ +use chrono::{DateTime, Duration, Utc}; + +#[derive(Debug, Clone, sqlx::FromRow)] +pub struct InstallationSyncState { + pub installation_id: i64, + pub webhook_seen_at: Option>, + pub last_sync_started_at: Option>, + pub last_sync_completed_at: Option>, + pub last_sync_status: Option, + pub last_error: Option, + pub updated_at: DateTime, +} + +pub async fn upsert_webhook_seen( + executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>, + installation_id: i64, +) -> Result<(), sqlx::Error> { + sqlx::query( + r#"INSERT INTO github_installation_sync_state (installation_id, webhook_seen_at, updated_at) + VALUES ($1, NOW(), NOW()) + ON CONFLICT (installation_id) DO UPDATE SET + webhook_seen_at = NOW(), + updated_at = NOW()"#, + ) + .bind(installation_id) + .execute(executor) + .await?; + Ok(()) +} + +pub async fn mark_sync_started( + executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>, + installation_id: i64, +) -> Result<(), sqlx::Error> { + sqlx::query( + r#"INSERT INTO github_installation_sync_state (installation_id, last_sync_started_at, updated_at) + VALUES ($1, NOW(), NOW()) + ON CONFLICT (installation_id) DO UPDATE SET + last_sync_started_at = NOW(), + updated_at = NOW()"#, + ) + .bind(installation_id) + .execute(executor) + .await?; + Ok(()) +} + +pub async fn mark_sync_completed( + executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>, + installation_id: i64, + status: &str, + error: Option<&str>, +) -> Result<(), sqlx::Error> { + sqlx::query( + r#"INSERT INTO github_installation_sync_state + (installation_id, last_sync_completed_at, last_sync_status, last_error, updated_at) + VALUES ($1, NOW(), $2, $3, NOW()) + ON CONFLICT (installation_id) DO UPDATE SET + last_sync_completed_at = NOW(), + last_sync_status = EXCLUDED.last_sync_status, + last_error = EXCLUDED.last_error, + updated_at = NOW()"#, + ) + .bind(installation_id) + .bind(status) + .bind(error) + .execute(executor) + .await?; + Ok(()) +} + +/// Installations that should be reconciled by the worker: +/// - never observed via webhook, or webhook is older than `stale_after`, +/// - AND not already synced more recently than `min_sync_interval`. +/// +/// Failed syncs are eligible regardless of the throttle so they can recover. +pub async fn list_installations_needing_sync( + executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>, + stale_after: Duration, + min_sync_interval: Duration, + limit: i64, +) -> Result, sqlx::Error> { + let webhook_cutoff = Utc::now() - stale_after; + let sync_cutoff = Utc::now() - min_sync_interval; + sqlx::query_as::<_, InstallationSyncState>( + r#"SELECT * FROM github_installation_sync_state + WHERE (webhook_seen_at IS NULL OR webhook_seen_at < $1) + AND ( + last_sync_status = 'failed' + OR last_sync_completed_at IS NULL + OR last_sync_completed_at < $2 + ) + AND (last_sync_started_at IS NULL OR last_sync_started_at < $2) + ORDER BY webhook_seen_at NULLS FIRST, installation_id + LIMIT $3"#, + ) + .bind(webhook_cutoff) + .bind(sync_cutoff) + .bind(limit) + .fetch_all(executor) + .await +} + +pub async fn find_by_installation_id( + executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>, + installation_id: i64, +) -> Result, sqlx::Error> { + sqlx::query_as::<_, InstallationSyncState>( + "SELECT * FROM github_installation_sync_state WHERE installation_id = $1", + ) + .bind(installation_id) + .fetch_optional(executor) + .await +} diff --git a/crates/db/src/queries/mod.rs b/crates/db/src/queries/mod.rs index 005c3b6..a2aeee6 100644 --- a/crates/db/src/queries/mod.rs +++ b/crates/db/src/queries/mod.rs @@ -5,6 +5,7 @@ pub mod board_project; pub mod board_run; pub mod diff; pub mod github_api_cache; +pub mod github_installation_sync_state; pub mod github_job; pub mod repository; pub mod run_check; diff --git a/crates/db/src/queries/repository.rs b/crates/db/src/queries/repository.rs index 2202191..bca471a 100644 --- a/crates/db/src/queries/repository.rs +++ b/crates/db/src/queries/repository.rs @@ -182,3 +182,15 @@ pub async fn find_existing_github_ids( .fetch_all(executor) .await } + +pub async fn list_github_ids_for_installation( + executor: impl sqlx::Executor<'_, Database = sqlx::Postgres>, + installation_id: i64, +) -> Result, sqlx::Error> { + sqlx::query_scalar::<_, i64>( + "SELECT github_repository_id FROM repositories WHERE installation_id = $1", + ) + .bind(installation_id) + .fetch_all(executor) + .await +} diff --git a/crates/github/src/client.rs b/crates/github/src/client.rs index dc22126..703b491 100644 --- a/crates/github/src/client.rs +++ b/crates/github/src/client.rs @@ -2,7 +2,7 @@ use secrecy::{ExposeSecret, SecretString}; use crate::config::GitHubAppConfig; use crate::error::GitHubClientError; -use crate::types::{CreatedComment, CreatedIssue, IssueInfo, IssueState}; +use crate::types::{CreatedComment, CreatedIssue, InstallationRepoInfo, IssueInfo, IssueState}; /// Trait for GitHub App client operations. /// Production implementation uses octocrab; tests can mock this trait. @@ -52,6 +52,21 @@ pub trait GitHubAppClient: Send + Sync { comment_id: u64, body: &str, ) -> Result<(), GitHubClientError>; + + /// List installations for the authenticated GitHub App. + /// Default returns an empty list; the worker reconciler tolerates that as "nothing to sync". + async fn list_installation_ids(&self) -> Result, GitHubClientError> { + Ok(Vec::new()) + } + + /// List repositories accessible to the given installation. + /// Default returns an empty list. + async fn list_installation_repositories( + &self, + _installation_id: u64, + ) -> Result, GitHubClientError> { + Ok(Vec::new()) + } } /// Production implementation backed by octocrab. @@ -200,6 +215,76 @@ impl GitHubAppClient for OctocrabGitHubAppClient { Ok(()) } + + async fn list_installation_ids(&self) -> Result, GitHubClientError> { + let mut ids = Vec::new(); + let mut page = 1u32; + loop { + let result = self + .octocrab + .apps() + .installations() + .per_page(100u8) + .page(page) + .send() + .await + .map_err(GitHubClientError::from)?; + + let items = result.items; + if items.is_empty() { + break; + } + for inst in &items { + ids.push(inst.id.0); + } + if items.len() < 100 { + break; + } + page += 1; + } + Ok(ids) + } + + async fn list_installation_repositories( + &self, + installation_id: u64, + ) -> Result, GitHubClientError> { + let installation_crab = self + .octocrab + .installation(octocrab::models::InstallationId(installation_id)) + .map_err(GitHubClientError::from)?; + + let mut repos = Vec::new(); + let mut page = 1u32; + loop { + let route = format!("/installation/repositories?per_page=100&page={page}"); + let resp: octocrab::models::InstallationRepositories = installation_crab + .get(&route, None::<&()>) + .await + .map_err(GitHubClientError::from)?; + + let count = resp.repositories.len(); + for r in &resp.repositories { + let owner = match r.owner.as_ref().map(|a| a.login.clone()) { + Some(o) => o, + None => match r.full_name.as_ref().and_then(|fn_| fn_.split_once('/')) { + Some((o, _)) => o.to_string(), + None => continue, + }, + }; + repos.push(InstallationRepoInfo { + id: r.id.0 as i64, + owner, + name: r.name.clone(), + }); + } + if count < 100 { + break; + } + page += 1; + } + Ok(repos) + } } #[cfg(test)] diff --git a/crates/github/src/lib.rs b/crates/github/src/lib.rs index 8b5463f..8506283 100644 --- a/crates/github/src/lib.rs +++ b/crates/github/src/lib.rs @@ -6,4 +6,4 @@ pub mod types; pub use client::{GitHubAppClient, OctocrabGitHubAppClient}; pub use config::GitHubAppConfig; pub use error::GitHubClientError; -pub use types::{CreatedComment, CreatedIssue, IssueInfo, IssueState}; +pub use types::{CreatedComment, CreatedIssue, InstallationRepoInfo, IssueInfo, IssueState}; diff --git a/crates/github/src/types.rs b/crates/github/src/types.rs index a18b3b1..ce6dc6a 100644 --- a/crates/github/src/types.rs +++ b/crates/github/src/types.rs @@ -23,3 +23,10 @@ pub enum IssueState { pub struct CreatedComment { pub id: u64, } + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct InstallationRepoInfo { + pub id: i64, + pub owner: String, + pub name: String, +} diff --git a/crates/worker/src/handlers/create_issue.rs b/crates/worker/src/handlers/create_issue.rs index 6fc72f5..105ce64 100644 --- a/crates/worker/src/handlers/create_issue.rs +++ b/crates/worker/src/handlers/create_issue.rs @@ -319,6 +319,10 @@ mod tests { poll_interval_secs: 2, timeout_sweep_interval_secs: 60, cache_cleanup_interval_secs: 3600, + installation_sync_interval_secs: 1800, + installation_sync_stale_after_secs: 86_400, + installation_sync_min_interval_secs: 3600, + installation_sync_max_per_sweep: 50, github_app_id: None, github_private_key_pem: None, app_domain: "https://test.example.com".into(), diff --git a/crates/worker/src/installation_sync.rs b/crates/worker/src/installation_sync.rs new file mode 100644 index 0000000..3a850d0 --- /dev/null +++ b/crates/worker/src/installation_sync.rs @@ -0,0 +1,305 @@ +//! Issue #105: Worker-side reconciliation of the `repositories` table for +//! installations whose webhook stream is stale. +//! +//! The repository list API serves data from the local `repositories` table. +//! Webhook events are the primary source of truth for which repositories belong +//! to which installation; the worker here is the safety net for missed webhooks +//! and for the first-time view of an installation we have never seen. + +use std::collections::HashSet; +use std::time::Duration as StdDuration; + +use boardflow_db::queries::{github_installation_sync_state, repository}; +use boardflow_github::{GitHubAppClient, GitHubClientError, InstallationRepoInfo}; +use chrono::Duration as ChronoDuration; +use sqlx::PgPool; + +use crate::config::WorkerConfig; + +/// Single sweep: reconcile installations that are due. Safe to call repeatedly. +pub async fn sync_stale_installations( + pool: &PgPool, + config: &WorkerConfig, + github_client: &dyn GitHubAppClient, +) { + let max_per_sweep = config.installation_sync_max_per_sweep as usize; + if max_per_sweep == 0 { + return; + } + + let installation_ids = match github_client.list_installation_ids().await { + Ok(ids) => ids, + Err(e) => { + tracing::warn!(error = %e, "installation sync: failed to list app installations"); + return; + } + }; + + if installation_ids.is_empty() { + tracing::debug!("installation sync: no installations to reconcile"); + return; + } + + let stale_after = ChronoDuration::seconds(config.installation_sync_stale_after_secs as i64); + let min_interval = ChronoDuration::seconds(config.installation_sync_min_interval_secs as i64); + + let mut synced = 0usize; + for installation_id in installation_ids { + if synced >= max_per_sweep { + tracing::info!( + limit = max_per_sweep, + "installation sync: reached per-sweep cap, deferring remaining installations" + ); + break; + } + + let installation_id_i64 = installation_id as i64; + let state = match github_installation_sync_state::find_by_installation_id( + pool, + installation_id_i64, + ) + .await + { + Ok(state) => state, + Err(e) => { + tracing::warn!( + error = %e, + installation_id, + "installation sync: failed to read sync state, skipping" + ); + continue; + } + }; + + if !needs_sync(state.as_ref(), stale_after, min_interval) { + continue; + } + + if let Err(e) = + github_installation_sync_state::mark_sync_started(pool, installation_id_i64).await + { + tracing::warn!( + error = %e, + installation_id, + "installation sync: failed to record sync_started, skipping" + ); + continue; + } + + let outcome = reconcile_installation(pool, github_client, installation_id).await; + let (status, err) = match &outcome { + Ok(_) => ("success", None), + Err(e) => ("failed", Some(e.clone())), + }; + + if let Err(e) = github_installation_sync_state::mark_sync_completed( + pool, + installation_id_i64, + status, + err.as_deref(), + ) + .await + { + tracing::warn!( + error = %e, + installation_id, + "installation sync: failed to record sync_completed" + ); + } + + match outcome { + Ok(repo_count) => { + tracing::info!( + installation_id, + repo_count, + "installation sync: reconciled installation" + ); + } + Err(e) => { + tracing::warn!( + installation_id, + error = %e, + "installation sync: failed to reconcile installation" + ); + } + } + synced += 1; + } +} + +fn needs_sync( + state: Option<&github_installation_sync_state::InstallationSyncState>, + stale_after: ChronoDuration, + min_interval: ChronoDuration, +) -> bool { + let now = chrono::Utc::now(); + + let state = match state { + None => return true, + Some(s) => s, + }; + + // Throttle: don't double-sync if we just ran (or just started). + if let Some(started) = state.last_sync_started_at { + if now - started < min_interval { + return false; + } + } + if let Some(completed) = state.last_sync_completed_at { + if now - completed < min_interval { + // Recently completed; only re-sync if previous attempt failed AND throttle window + // already passed (handled above). + return state.last_sync_status.as_deref() == Some("failed") + && now - completed >= min_interval; + } + } + + // Eligibility: either no webhook ever, webhook old enough, or last sync failed. + let webhook_stale = match state.webhook_seen_at { + None => true, + Some(seen) => now - seen >= stale_after, + }; + let failed = state.last_sync_status.as_deref() == Some("failed"); + + webhook_stale || failed +} + +async fn reconcile_installation( + pool: &PgPool, + github_client: &dyn GitHubAppClient, + installation_id: u64, +) -> Result { + let repos = match github_client + .list_installation_repositories(installation_id) + .await + { + Ok(repos) => repos, + Err(GitHubClientError::NotFound(msg)) => { + // Installation no longer accessible (deleted/suspended). Treat as + // empty so we clear stale rows. + tracing::warn!(installation_id, error = %msg, "installation not found, clearing rows"); + Vec::new() + } + Err(e) => return Err(e.to_string()), + }; + + upsert_and_prune(pool, installation_id as i64, &repos) + .await + .map_err(|e| format!("db error: {e}")) +} + +async fn upsert_and_prune( + pool: &PgPool, + installation_id: i64, + repos: &[InstallationRepoInfo], +) -> Result { + let mut keep: HashSet = HashSet::with_capacity(repos.len()); + + for repo in repos { + repository::upsert(pool, repo.id, &repo.owner, &repo.name, installation_id).await?; + keep.insert(repo.id); + } + + // Anything still claimed by this installation but not in the latest set + // has been removed — drop the installation linkage. + let current_repos = repository::list_github_ids_for_installation(pool, installation_id).await?; + for github_id in current_repos { + if !keep.contains(&github_id) { + let _ = + repository::clear_installation_for_repo(pool, github_id, installation_id).await?; + } + } + + Ok(repos.len()) +} + +/// Convenience wrapper used by the dispatcher loop. +pub fn interval(config: &WorkerConfig) -> tokio::time::Interval { + tokio::time::interval(StdDuration::from_secs( + config.installation_sync_interval_secs, + )) +} + +#[cfg(test)] +mod tests { + use super::*; + use boardflow_db::queries::github_installation_sync_state::InstallationSyncState; + use chrono::Utc; + + fn state( + webhook_seen_at: Option>, + last_sync_started_at: Option>, + last_sync_completed_at: Option>, + last_sync_status: Option<&str>, + ) -> InstallationSyncState { + InstallationSyncState { + installation_id: 1, + webhook_seen_at, + last_sync_started_at, + last_sync_completed_at, + last_sync_status: last_sync_status.map(String::from), + last_error: None, + updated_at: Utc::now(), + } + } + + #[test] + fn unknown_installation_needs_sync() { + let stale_after = ChronoDuration::hours(24); + let min_interval = ChronoDuration::hours(1); + assert!(needs_sync(None, stale_after, min_interval)); + } + + #[test] + fn fresh_webhook_no_sync() { + let stale_after = ChronoDuration::hours(24); + let min_interval = ChronoDuration::hours(1); + let s = state( + Some(Utc::now() - ChronoDuration::minutes(10)), + None, + None, + None, + ); + assert!(!needs_sync(Some(&s), stale_after, min_interval)); + } + + #[test] + fn stale_webhook_needs_sync() { + let stale_after = ChronoDuration::hours(24); + let min_interval = ChronoDuration::hours(1); + let s = state( + Some(Utc::now() - ChronoDuration::hours(48)), + None, + None, + None, + ); + assert!(needs_sync(Some(&s), stale_after, min_interval)); + } + + #[test] + fn recent_sync_throttles_even_if_webhook_stale() { + let stale_after = ChronoDuration::hours(24); + let min_interval = ChronoDuration::hours(1); + let s = state( + Some(Utc::now() - ChronoDuration::hours(48)), + Some(Utc::now() - ChronoDuration::minutes(5)), + Some(Utc::now() - ChronoDuration::minutes(5)), + Some("success"), + ); + assert!(!needs_sync(Some(&s), stale_after, min_interval)); + } + + #[test] + fn failed_sync_retries_once_throttle_clears() { + let stale_after = ChronoDuration::hours(24); + let min_interval = ChronoDuration::hours(1); + // Failed and last_sync was > min_interval ago → retry + let s = state( + Some(Utc::now() - ChronoDuration::hours(48)), + Some(Utc::now() - ChronoDuration::hours(2)), + Some(Utc::now() - ChronoDuration::hours(2)), + Some("failed"), + ); + assert!(needs_sync(Some(&s), stale_after, min_interval)); + } +} diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 4495878..f25177c 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -6,5 +6,6 @@ pub mod comment_body; pub mod config; pub mod dispatcher; pub mod handlers; +pub mod installation_sync; pub use config::WorkerConfig; diff --git a/crates/worker/src/main.rs b/crates/worker/src/main.rs index 39c257d..60ff05c 100644 --- a/crates/worker/src/main.rs +++ b/crates/worker/src/main.rs @@ -3,7 +3,7 @@ use boardflow_github::{GitHubAppClient, OctocrabGitHubAppClient}; use secrecy::SecretString; use boardflow_worker::config::WorkerConfig; -use boardflow_worker::dispatcher; +use boardflow_worker::{dispatcher, installation_sync}; #[tokio::main] async fn main() { @@ -81,6 +81,9 @@ async fn main() { )); cache_cleanup_interval.tick().await; // 初回tickを消化 + let mut installation_sync_interval = installation_sync::interval(&config); + installation_sync_interval.tick().await; // 初回tickを消化 + let shutdown = tokio::signal::ctrl_c(); tokio::pin!(shutdown); @@ -103,6 +106,11 @@ async fn main() { _ = cache_cleanup_interval.tick() => { dispatcher::sweep_expired_cache(&pool).await; } + _ = installation_sync_interval.tick() => { + if let Some(client) = github_client.as_deref() { + installation_sync::sync_stale_installations(&pool, &config, client).await; + } + } } } diff --git a/crates/worker/tests/create_issue_test.rs b/crates/worker/tests/create_issue_test.rs index 81d6176..4582431 100644 --- a/crates/worker/tests/create_issue_test.rs +++ b/crates/worker/tests/create_issue_test.rs @@ -115,6 +115,10 @@ fn make_config() -> boardflow_worker::WorkerConfig { poll_interval_secs: 2, timeout_sweep_interval_secs: 60, cache_cleanup_interval_secs: 3600, + installation_sync_interval_secs: 1800, + installation_sync_stale_after_secs: 86_400, + installation_sync_min_interval_secs: 3600, + installation_sync_max_per_sweep: 50, github_app_id: None, github_private_key_pem: None, app_domain: "https://test.boardflow.example.com".into(), diff --git a/crates/worker/tests/dashboard_comment_test.rs b/crates/worker/tests/dashboard_comment_test.rs index 2518726..89fa618 100644 --- a/crates/worker/tests/dashboard_comment_test.rs +++ b/crates/worker/tests/dashboard_comment_test.rs @@ -128,6 +128,10 @@ fn make_config() -> boardflow_worker::WorkerConfig { poll_interval_secs: 2, timeout_sweep_interval_secs: 60, cache_cleanup_interval_secs: 3600, + installation_sync_interval_secs: 1800, + installation_sync_stale_after_secs: 86_400, + installation_sync_min_interval_secs: 3600, + installation_sync_max_per_sweep: 50, github_app_id: None, github_private_key_pem: None, app_domain: "https://test.boardflow.example.com".into(), diff --git a/crates/worker/tests/installation_sync_test.rs b/crates/worker/tests/installation_sync_test.rs new file mode 100644 index 0000000..c37ebd3 --- /dev/null +++ b/crates/worker/tests/installation_sync_test.rs @@ -0,0 +1,286 @@ +//! Issue #105: integration tests for the worker installation reconciler. + +use std::sync::Mutex; + +use async_trait::async_trait; +use boardflow_db::queries::{github_installation_sync_state, repository as repo_queries}; +use boardflow_github::{ + CreatedComment, CreatedIssue, GitHubAppClient, GitHubClientError, InstallationRepoInfo, + IssueInfo, +}; +use boardflow_worker::installation_sync; +use secrecy::SecretString; +use serial_test::serial; +use sqlx::PgPool; + +mod common { + pub fn rand_i64() -> i64 { + rand::random::() as i64 + 10_000_000_000 + } +} + +async fn setup_pool() -> Option { + let database_url = match std::env::var("DATABASE_URL") { + Ok(url) => url, + Err(_) => { + eprintln!("Skipping test: DATABASE_URL not set"); + return None; + } + }; + let pool = PgPool::connect(&database_url).await.unwrap(); + sqlx::migrate!("../db/migrations").run(&pool).await.unwrap(); + Some(pool) +} + +fn make_config() -> boardflow_worker::WorkerConfig { + boardflow_worker::WorkerConfig { + db: boardflow_config::DatabaseConfig { + database_url: String::new(), + }, + s3: boardflow_config::S3Config { + endpoint: None, + access_key: None, + secret_key: None, + staging_bucket: "test".into(), + final_bucket: "test".into(), + }, + poll_interval_secs: 2, + timeout_sweep_interval_secs: 60, + cache_cleanup_interval_secs: 3600, + installation_sync_interval_secs: 1800, + installation_sync_stale_after_secs: 86_400, + installation_sync_min_interval_secs: 3600, + installation_sync_max_per_sweep: 50, + github_app_id: None, + github_private_key_pem: None, + app_domain: "https://test.example.com".into(), + } +} + +#[derive(Default)] +struct MockApp { + installations: Vec, + repos: std::collections::HashMap>, + fail_for: Option, + calls: Mutex>, +} + +impl MockApp { + fn new(installations: Vec) -> Self { + Self { + installations, + ..Default::default() + } + } + fn with_repos(mut self, installation_id: u64, repos: Vec) -> Self { + self.repos.insert(installation_id, repos); + self + } + fn with_fail(mut self, installation_id: u64) -> Self { + self.fail_for = Some(installation_id); + self + } +} + +#[async_trait] +impl GitHubAppClient for MockApp { + async fn get_installation_token(&self, _: u64) -> Result { + Ok(SecretString::from("mock".to_string())) + } + async fn create_issue( + &self, + _: u64, + _: &str, + _: &str, + _: &str, + _: &str, + ) -> Result { + unimplemented!() + } + async fn get_issue( + &self, + _: u64, + _: &str, + _: &str, + _: u64, + ) -> Result { + unimplemented!() + } + async fn create_comment( + &self, + _: u64, + _: &str, + _: &str, + _: u64, + _: &str, + ) -> Result { + unimplemented!() + } + async fn update_comment( + &self, + _: u64, + _: &str, + _: &str, + _: u64, + _: &str, + ) -> Result<(), GitHubClientError> { + unimplemented!() + } + + async fn list_installation_ids(&self) -> Result, GitHubClientError> { + Ok(self.installations.clone()) + } + + async fn list_installation_repositories( + &self, + installation_id: u64, + ) -> Result, GitHubClientError> { + self.calls.lock().unwrap().push(installation_id); + if self.fail_for == Some(installation_id) { + return Err(GitHubClientError::Api("boom".into())); + } + Ok(self + .repos + .get(&installation_id) + .cloned() + .unwrap_or_default()) + } +} + +#[tokio::test] +#[serial] +async fn syncs_unknown_installation_and_upserts_repos() { + let Some(pool) = setup_pool().await else { + return; + }; + let installation_id = common::rand_i64() as u64; + let repo_id = common::rand_i64(); + + let mock = MockApp::new(vec![installation_id]).with_repos( + installation_id, + vec![InstallationRepoInfo { + id: repo_id, + owner: "octo".into(), + name: "demo".into(), + }], + ); + let config = make_config(); + + installation_sync::sync_stale_installations(&pool, &config, &mock).await; + + let repo = repo_queries::find_by_github_id(&pool, repo_id) + .await + .unwrap(); + assert!(repo.is_some(), "repo should be upserted"); + let repo = repo.unwrap(); + assert_eq!(repo.owner, "octo"); + assert_eq!(repo.name, "demo"); + assert_eq!(repo.installation_id, installation_id as i64); + + let state = + github_installation_sync_state::find_by_installation_id(&pool, installation_id as i64) + .await + .unwrap() + .expect("sync_state row should exist"); + assert_eq!(state.last_sync_status.as_deref(), Some("success")); + assert!(state.last_sync_completed_at.is_some()); +} + +#[tokio::test] +#[serial] +async fn skips_installation_with_recent_webhook() { + let Some(pool) = setup_pool().await else { + return; + }; + let installation_id = common::rand_i64() as u64; + + // Recent webhook seen → should be skipped (and never call list_installation_repositories) + github_installation_sync_state::upsert_webhook_seen(&pool, installation_id as i64) + .await + .unwrap(); + + let mock = MockApp::new(vec![installation_id]); + let config = make_config(); + + installation_sync::sync_stale_installations(&pool, &config, &mock).await; + + let calls = mock.calls.lock().unwrap(); + assert!( + calls.is_empty(), + "list_installation_repositories should not be called for fresh webhook installations" + ); +} + +#[tokio::test] +#[serial] +async fn records_failure_when_repo_fetch_errors() { + let Some(pool) = setup_pool().await else { + return; + }; + let installation_id = common::rand_i64() as u64; + + let mock = MockApp::new(vec![installation_id]).with_fail(installation_id); + let config = make_config(); + + installation_sync::sync_stale_installations(&pool, &config, &mock).await; + + let state = + github_installation_sync_state::find_by_installation_id(&pool, installation_id as i64) + .await + .unwrap() + .expect("sync_state row should exist"); + assert_eq!(state.last_sync_status.as_deref(), Some("failed")); + assert!(state.last_error.is_some()); +} + +#[tokio::test] +#[serial] +async fn clears_installation_link_for_removed_repos() { + let Some(pool) = setup_pool().await else { + return; + }; + let installation_id = common::rand_i64() as u64; + let kept_repo = common::rand_i64(); + let removed_repo = common::rand_i64(); + + // Seed two repos belonging to this installation + repo_queries::upsert(&pool, kept_repo, "octo", "kept", installation_id as i64) + .await + .unwrap(); + repo_queries::upsert( + &pool, + removed_repo, + "octo", + "removed", + installation_id as i64, + ) + .await + .unwrap(); + + let mock = MockApp::new(vec![installation_id]).with_repos( + installation_id, + vec![InstallationRepoInfo { + id: kept_repo, + owner: "octo".into(), + name: "kept".into(), + }], + ); + let config = make_config(); + + installation_sync::sync_stale_installations(&pool, &config, &mock).await; + + let kept = repo_queries::find_by_github_id(&pool, kept_repo) + .await + .unwrap() + .unwrap(); + assert_eq!(kept.installation_id, installation_id as i64); + + let removed = repo_queries::find_by_github_id(&pool, removed_repo) + .await + .unwrap() + .unwrap(); + assert_eq!( + removed.installation_id, 0, + "removed repo should have its installation linkage cleared" + ); +} diff --git a/crates/worker/tests/run_result_comment_test.rs b/crates/worker/tests/run_result_comment_test.rs index 42bb336..ceb44d7 100644 --- a/crates/worker/tests/run_result_comment_test.rs +++ b/crates/worker/tests/run_result_comment_test.rs @@ -118,6 +118,10 @@ fn make_config() -> boardflow_worker::WorkerConfig { poll_interval_secs: 2, timeout_sweep_interval_secs: 60, cache_cleanup_interval_secs: 3600, + installation_sync_interval_secs: 1800, + installation_sync_stale_after_secs: 86_400, + installation_sync_min_interval_secs: 3600, + installation_sync_max_per_sweep: 50, github_app_id: None, github_private_key_pem: None, app_domain: "https://test.boardflow.example.com".into(),