diff --git a/CLAUDE.md b/CLAUDE.md index 3734954..c5781d9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -29,7 +29,23 @@ moments-worker — ingestion daemon binary (runs migrations, connects as mom ### Key Design Decisions - **Raw payload storage**: upstream JSON is stored verbatim in `events.payload` (JSONB). The `reshape()` function in `moments-core/src/presentation.rs` transforms payloads into `TimelineItem` at request time — no re-ingestion needed to change presentation. -- **Public/private gate**: `events.public` boolean controls API visibility. Only `public = true` rows are served. +- **Public/private gate**: `events.public` boolean controls API visibility. Only + `public = true` rows are served on the detail endpoints (`events`, `projects`, + `activity/summary`, `languages/repos`); the count endpoints (`activity/daily`, + `activity/hourly`, `sources`, `languages/daily`) pass `include_private = true`, + so private work shows up as volume without leaking repo names or messages. +- **Visibility reconciliation**: `public` is stamped at ingest from whatever the + forge reported then, and every poller is incremental — so nothing would ever + revisit a repo that later flipped public ↔ private. The github and gitea + sources therefore run a reconciliation pass before ingesting (`reconcile_visibility` + in `github_repo.rs` / `gitea.rs`): they re-read current visibility and + `UPDATE events SET public` for the whole history of each repo. It keys off + `events.repo`, a stored generated column (migration 0006) that derives the repo + from the payload — the same expression four read queries used to each carry + their own copy of. Repos already hidden are skipped (they can't leak, and + staying hidden is the safe direction), and a 404 counts as private: with the + user's own token, a repo still in reach answers 200 even when private. + Rate limits and transient errors never flip anything. - **Wire types are hand-maintained**: `ui/src/api/client.ts` mirrors Rust entity types manually. - **Migrations**: run automatically on worker startup via `sqlx::migrate!`. The API binary never runs migrations. @@ -77,7 +93,7 @@ UTC + explicit field widths so SSR and client hydration match byte-for-byte. ## Database -PostgreSQL with three migrations in `crates/moments-data/migrations/`. Two roles: `moments_rw` (worker, full access) and `moments_ro` (API, SELECT-only). +PostgreSQL with migrations in `crates/moments-data/migrations/`. Two roles: `moments_rw` (worker, full access) and `moments_ro` (API, SELECT-only). ## API Endpoints diff --git a/crates/moments-core/src/lib.rs b/crates/moments-core/src/lib.rs index 691af0c..bd43a4e 100644 --- a/crates/moments-core/src/lib.rs +++ b/crates/moments-core/src/lib.rs @@ -8,7 +8,7 @@ use async_trait::async_trait; use chrono::NaiveDate; use moments_entities::{ DailyCount, Event, EventQuery, HourlyAvg, LanguageDailyCount, ProjectSummary, RepoLanguage, - RepoPeriodCount, Source, SourceSummary, SummaryBucket, + RepoPeriodCount, RepoVisibility, Source, SourceSummary, SummaryBucket, }; #[derive(Debug, thiserror::Error)] @@ -70,4 +70,31 @@ pub trait EventWriter: Send + Sync { source: moments_entities::Source, keep_ids: &[String], ) -> Result; + + /// Every repo `source` holds events for, with its current exposure — + /// `public` is true when *any* event for that repo is still being + /// served. Reads, but lives on the write port: it exists only to tell a + /// poller which repos its reconciliation pass has to answer for. + /// + /// `host` filters on the `_host` stamped into each payload, for sources + /// (gitea) where one `Source` can span several forges. Rows predating + /// that stamp are treated as belonging to the queried host. + async fn repo_visibility( + &self, + source: Source, + host: Option<&str>, + ) -> Result, StoreError>; + + /// Re-stamp `events.public` for the given repos of `source` from what + /// upstream reports now. Returns the number of event rows actually + /// flipped — rows already carrying the right value are left alone. + /// + /// `events.public` is otherwise decided once, when a row is ingested, + /// and every poller is incremental; without this a repo that turned + /// private upstream would keep serving its history indefinitely. + async fn set_repo_visibility( + &self, + source: Source, + repos: &[RepoVisibility], + ) -> Result; } diff --git a/crates/moments-data/migrations/0006_event_repo.sql b/crates/moments-data/migrations/0006_event_repo.sql new file mode 100644 index 0000000..564edd6 --- /dev/null +++ b/crates/moments-data/migrations/0006_event_repo.sql @@ -0,0 +1,53 @@ +-- Materialise the repo each event belongs to. +-- +-- Two problems, one column. +-- +-- 1. Four read queries (list_events, list_projects, activity_summary, +-- language_daily_counts) each carried their own copy of the payload -> +-- repo CASE. A payload shape learned in one copy never reached the +-- others: /search/issues items, which carry neither `repo.name` nor +-- `repository.full_name`, resolved to NULL in all of them. +-- +-- 2. `events.public` is stamped once, at ingest, from the visibility the +-- poller saw at the time — and every poller is incremental, so history +-- is never re-fetched. A repo flipped public -> private upstream kept +-- serving its old commit messages and repo name forever. Reconciling +-- that needs a repo key the worker can UPDATE against. +-- +-- STORED rather than VIRTUAL: the reconciliation UPDATE and the language +-- visibility gate both filter on it, so it has to be indexable. +-- +-- Every branch is IMMUTABLE (jsonb accessors, COALESCE, regex substring), +-- which is what a generated column requires. + +ALTER TABLE events + ADD COLUMN repo TEXT GENERATED ALWAYS AS ( + CASE source + WHEN 'github' THEN COALESCE( + -- events API + payload->'repo'->>'name', + -- /search/commits + payload->'repository'->>'full_name', + -- per-repo commit enumeration (stamped by the poller) + payload->>'_repo', + -- /search/issues: only an api URL to go on. The regex + -- yields NULL rather than a mangled string when the URL + -- shape is anything else. + substring( + payload->>'repository_url' + from '^https://api\.github\.com/repos/(.+)$' + ) + ) + WHEN 'gitea' THEN COALESCE( + payload->'repo'->>'full_name', + payload->'repo'->>'name' + ) + WHEN 'hg' THEN payload->>'_repo' + WHEN 'bugzilla' THEN payload->>'product' + ELSE NULL + END + ) STORED; + +-- Serves the reconciliation UPDATE (source, repo), the exposure rollup it +-- reads first (source, repo, public), and the EXISTS gate on repo_languages. +CREATE INDEX events_source_repo_public ON events (source, repo, public); diff --git a/crates/moments-data/src/gitea.rs b/crates/moments-data/src/gitea.rs index 91aff9d..e932621 100644 --- a/crates/moments-data/src/gitea.rs +++ b/crates/moments-data/src/gitea.rs @@ -15,15 +15,23 @@ use std::sync::Arc; use async_trait::async_trait; use chrono::{DateTime, Utc}; use moments_core::{EventSource, EventWriter, PollerStateStore, SourceError}; -use moments_entities::{Event, RepoLanguage, Source}; +use moments_entities::{Event, RepoLanguage, RepoVisibility, Source}; use reqwest::{Client, header}; use serde_json::Value; -use tracing::debug; +use tracing::{debug, info, warn}; + +use crate::{Probe, probe_verdict}; const SOURCE_NAME: &str = "gitea"; const USER_AGENT: &str = concat!("moments/", env!("CARGO_PKG_VERSION"), " (+https://rob.tn)"); const MAX_BACKFILL_PAGES: u32 = 20; +/// Bookkeeping key and cadence for the visibility reconciliation pass. The +/// feed itself is polled every few minutes; re-asking the forge about every +/// repo we have history for is a once-a-day job. +const VISIBILITY_STATE_KEY: &str = "gitea:visibility"; +const VISIBILITY_INTERVAL_SECS: i64 = 86_400; + #[derive(Clone, Debug)] pub struct GiteaConfig { /// e.g. `git.lair.cafe`. Used to construct URLs the API doesn't return @@ -190,6 +198,108 @@ impl GiteaSource { Ok((total, repos)) } + /// Re-stamp `events.public` for every gitea repo we hold events for. + /// + /// The feed carries `is_private` per item, but after the first run only + /// page 1 is fetched — so a repo flipped to private upstream would have + /// its recent events relabelled while everything older kept its old + /// `public = true`. Gitea has no bulk "all repos I can see" endpoint + /// that covers org repos we merely contributed to, so each known repo is + /// asked directly, once a day rather than once a tick. + /// + /// Repos already hidden are skipped: they can't leak, and leaving them + /// hidden is the safe direction to err in. + async fn reconcile_visibility(&self) -> Result<(), SourceError> { + let prior = self.state.load(VISIBILITY_STATE_KEY).await?; + if let Some(last) = prior.as_ref().and_then(|s| s.last_modified) { + if (Utc::now() - last).num_seconds() < VISIBILITY_INTERVAL_SECS { + return Ok(()); + } + } + + let known = self + .writer + .repo_visibility(Source::Gitea, Some(&self.config.host)) + .await?; + let mut updates: Vec = Vec::new(); + let mut unprobed = 0usize; + + for entry in &known { + if !entry.public { + continue; + } + match self.probe_repo_visibility(&entry.repo).await { + Probe::Public => {} + Probe::Private => { + warn!(repo = %entry.repo, "repo no longer visible upstream; hiding its history"); + updates.push(RepoVisibility { + repo: entry.repo.clone(), + public: false, + }); + } + Probe::Unknown => unprobed += 1, + Probe::RateLimited => { + unprobed += 1; + break; + } + } + } + + let flipped = self + .writer + .set_repo_visibility(Source::Gitea, &updates) + .await?; + if flipped > 0 { + info!( + repos = updates.len(), + events = flipped, + "gitea visibility reconciled" + ); + } + if unprobed > 0 { + warn!(unprobed, "repos left unreconciled this pass"); + } + self.state + .save(VISIBILITY_STATE_KEY, None, Some(Utc::now())) + .await?; + Ok(()) + } + + /// Ask gitea what one repo's visibility is now. See [`probe_verdict`] + /// for how each status is read. + async fn probe_repo_visibility(&self, repo: &str) -> Probe { + let url = format!("https://{}/api/v1/repos/{}", self.config.host, repo); + let resp = match self.apply_headers(self.client.get(&url)).send().await { + Ok(r) => r, + Err(e) => { + warn!(repo = %repo, error = %e, "visibility probe failed; leaving unchanged"); + return Probe::Unknown; + } + }; + let status = resp.status().as_u16(); + if let Some(verdict) = probe_verdict(status) { + if verdict == Probe::Unknown { + warn!(repo = %repo, status, "visibility probe failed; leaving unchanged"); + } + return verdict; + } + match resp.json::().await { + // Absent `private` is read as private: an unrecognised body is + // not grounds for publishing. + Ok(v) => { + if v.get("private").and_then(Value::as_bool).unwrap_or(true) { + Probe::Private + } else { + Probe::Public + } + } + Err(e) => { + warn!(repo = %repo, error = %e, "visibility probe unparseable; leaving unchanged"); + Probe::Unknown + } + } + } + /// Fetch language breakdowns for the given repos via the Gitea REST API. async fn fetch_languages(&self, repos: &HashSet) -> Result { let mut total = 0usize; @@ -238,6 +348,13 @@ impl EventSource for GiteaSource { async fn poll(&self) -> Result { let mut all_repos = HashSet::new(); + // Ahead of ingestion: relabelling repos that turned private is the + // part that must not be what a mid-poll failure skips. Self-throttled + // to once a day. + if let Err(e) = self.reconcile_visibility().await { + warn!(error = %e, "visibility reconciliation failed; continuing"); + } + // Poll user's own activity feed (existing behavior). let user_url = self.user_feed_base_url(); let (mut total, repos) = self.poll_feed(SOURCE_NAME, &user_url, false).await?; diff --git a/crates/moments-data/src/github_repo.rs b/crates/moments-data/src/github_repo.rs index 5700ce8..984f651 100644 --- a/crates/moments-data/src/github_repo.rs +++ b/crates/moments-data/src/github_repo.rs @@ -19,17 +19,19 @@ //! `github_search`, so duplicates are resolved via idempotent upsert //! (the same commit reached via two branches just upserts twice). -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::sync::Arc; use async_trait::async_trait; use chrono::{DateTime, Utc}; use moments_core::{EventSource, EventWriter, PollerStateStore, SourceError}; -use moments_entities::{Event, RepoLanguage, Source}; +use moments_entities::{Event, RepoLanguage, RepoVisibility, Source}; use percent_encoding::{AsciiSet, CONTROLS, utf8_percent_encode}; use reqwest::{Client, header}; use serde_json::Value; -use tracing::{debug, warn}; +use tracing::{debug, info, warn}; + +use crate::{Probe, probe_verdict}; /// Encode characters that have meaning in a URL query — branch names can /// contain `/`, `#`, `?`, etc. Whitelisting is too fragile; encode anything @@ -59,6 +61,10 @@ pub struct GithubRepoConfig { pub user: String, pub token: Option, pub per_page: u32, + /// Ceiling on per-repo visibility probes in one reconciliation pass. + /// Only repos we are still exposing but discovery didn't return cost a + /// request, so this is a rate-limit backstop rather than a routine cap. + pub max_visibility_probes: u32, } impl Default for GithubRepoConfig { @@ -67,6 +73,7 @@ impl Default for GithubRepoConfig { user: "grenade".into(), token: None, per_page: 100, + max_visibility_probes: 250, } } } @@ -250,6 +257,130 @@ impl GithubRepoSource { Ok(repos) } + /// Re-stamp `events.public` for every github repo we hold events for. + /// + /// Visibility is read from upstream once, when a row is ingested, and + /// every github poller is incremental — the events feed is capped at 90 + /// days, search at its top-1000 window, this one at a per-branch `since` + /// cursor. So a repo flipped to private upstream would go on serving its + /// old commit messages and its name indefinitely. Discovery already + /// re-reads `private`/`isPrivate` for every accessible repo on each + /// poll; this applies that answer to the history we already hold. + /// + /// Repos discovery didn't return are probed individually, but only while + /// we're still exposing them: one already hidden can't leak, so it isn't + /// worth a request (and staying hidden is the safe direction to err in). + /// + /// All three github sources write `source = 'github'`, so this single + /// pass covers events ingested by any of them. + async fn reconcile_visibility(&self, discovered: &[Repo]) -> Result<(), SourceError> { + if discovered.is_empty() { + // No token, or discovery came back empty. Acting on that would + // hide the entire github timeline. + debug!("no repos discovered; skipping visibility reconciliation"); + return Ok(()); + } + let discovered: HashMap<&str, bool> = discovered + .iter() + .map(|r| (r.full_name.as_str(), !r.private)) + .collect(); + + let known = self.writer.repo_visibility(Source::Github, None).await?; + let mut updates: Vec = Vec::with_capacity(known.len()); + let mut probes: u32 = 0; + let mut unprobed: usize = 0; + let mut rate_limited = false; + + for entry in &known { + if let Some(&public) = discovered.get(entry.repo.as_str()) { + updates.push(RepoVisibility { + repo: entry.repo.clone(), + public, + }); + continue; + } + if !entry.public { + continue; + } + if rate_limited || probes >= self.config.max_visibility_probes { + unprobed += 1; + continue; + } + probes += 1; + match self.probe_repo_visibility(&entry.repo).await { + Probe::Public => {} + Probe::Private => { + warn!(repo = %entry.repo, "repo no longer visible upstream; hiding its history"); + updates.push(RepoVisibility { + repo: entry.repo.clone(), + public: false, + }); + } + Probe::Unknown => unprobed += 1, + Probe::RateLimited => { + rate_limited = true; + unprobed += 1; + } + } + } + + let flipped = self + .writer + .set_repo_visibility(Source::Github, &updates) + .await?; + if flipped > 0 { + info!( + repos = updates.len(), + events = flipped, + "github visibility reconciled" + ); + } else { + debug!(repos = updates.len(), "github visibility already current"); + } + if unprobed > 0 { + warn!( + unprobed, + probes, rate_limited, "repos left unreconciled this pass; retrying next poll" + ); + } + Ok(()) + } + + /// Ask GitHub what a single repo's visibility is now. See + /// [`probe_verdict`] for how each status is read. + async fn probe_repo_visibility(&self, full_name: &str) -> Probe { + let url = format!("https://api.github.com/repos/{full_name}"); + let resp = match self.apply_headers(self.client.get(&url)).send().await { + Ok(r) => r, + Err(e) => { + warn!(repo = %full_name, error = %e, "visibility probe failed; leaving unchanged"); + return Probe::Unknown; + } + }; + let status = resp.status().as_u16(); + if let Some(verdict) = probe_verdict(status) { + if verdict == Probe::Unknown { + warn!(repo = %full_name, status, "visibility probe failed; leaving unchanged"); + } + return verdict; + } + match resp.json::().await { + // Absent `private` is read as private: an unrecognised body is + // not grounds for publishing. + Ok(v) => { + if v.get("private").and_then(Value::as_bool).unwrap_or(true) { + Probe::Private + } else { + Probe::Public + } + } + Err(e) => { + warn!(repo = %full_name, error = %e, "visibility probe unparseable; leaving unchanged"); + Probe::Unknown + } + } + } + /// Branch discovery via GraphQL, filtered to branches whose HEAD /// commit was authored by the user. Skips the long tail of /// upstream-contributor branches in large forks (e.g. azure-docs). @@ -668,6 +799,13 @@ impl EventSource for GithubRepoSource { let repos = self.discover_repos().await?; debug!(repos = repos.len(), "discovered github repos"); + // Before scanning: the scan loop bails out on rate limits, and + // relabelling repos that turned private is the part that must not be + // the thing we skip. + if let Err(e) = self.reconcile_visibility(&repos).await { + warn!(error = %e, "visibility reconciliation failed; continuing"); + } + let mut total = 0usize; for repo in &repos { match self.scan_repo(repo).await { diff --git a/crates/moments-data/src/hg.rs b/crates/moments-data/src/hg.rs index 750af15..cd00cff 100644 --- a/crates/moments-data/src/hg.rs +++ b/crates/moments-data/src/hg.rs @@ -254,6 +254,20 @@ mod tests { ) -> Result { Ok(0) } + async fn repo_visibility( + &self, + _source: moments_entities::Source, + _host: Option<&str>, + ) -> Result, moments_core::StoreError> { + Ok(vec![]) + } + async fn set_repo_visibility( + &self, + _source: moments_entities::Source, + _repos: &[moments_entities::RepoVisibility], + ) -> Result { + Ok(0) + } } struct NoopState; #[async_trait] diff --git a/crates/moments-data/src/lib.rs b/crates/moments-data/src/lib.rs index 364b58a..cf725b9 100644 --- a/crates/moments-data/src/lib.rs +++ b/crates/moments-data/src/lib.rs @@ -12,7 +12,7 @@ use chrono::{DateTime, Utc}; use moments_core::{EventReader, EventWriter, PollerState, PollerStateStore, StoreError}; use moments_entities::{ DailyCount, Event, EventQuery, HourlyAvg, LanguageDailyCount, ProjectSummary, RepoLanguage, - RepoPeriodCount, Source, SourceSummary, SummaryBucket, + RepoPeriodCount, RepoVisibility, Source, SourceSummary, SummaryBucket, }; use sqlx::Row; use sqlx::postgres::{PgPool, PgPoolOptions}; @@ -43,6 +43,40 @@ fn map_err(e: E) -> StoreError { StoreError::Database(e.to_string()) } +/// What a single per-repo visibility probe concluded, during the +/// reconciliation pass that keeps `events.public` honest as repos flip +/// visibility upstream. Shared by the github and gitea sources. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum Probe { + Public, + /// Private, or gone/transferred beyond our reach — both mean stop + /// serving it. + Private, + /// Couldn't tell (transient error). Leave the repo as it is. + Unknown, + /// Rate limited; every further probe this pass would answer the same. + RateLimited, +} + +/// Verdict for a probe response status, or `None` when the body decides +/// (a 200 carries the repo's `private` flag). +/// +/// A 404 is deliberately `Private`, not `Unknown`: these probes run with the +/// user's own token, so a repo still within reach answers 200 even when it +/// is private — a 404 means deleted, transferred, or otherwise gone, none of +/// which we should keep publishing history for. Reconciliation runs in both +/// directions, so a spurious 404 that hides a repo is undone by the next +/// successful pass. Everything else leaves the repo exactly as it is: no +/// rate limit or outage should be able to blank the timeline. +pub(crate) fn probe_verdict(status: u16) -> Option { + match status { + 200 => None, + 403 | 429 => Some(Probe::RateLimited), + 404 => Some(Probe::Private), + _ => Some(Probe::Unknown), + } +} + #[async_trait] impl EventReader for PgStore { async fn list_events(&self, query: &EventQuery) -> Result, StoreError> { @@ -59,20 +93,7 @@ impl EventReader for PgStore { AND ($2::timestamptz IS NULL OR occurred_at < $2) AND ($3::text[] IS NULL OR source = ANY($3)) AND ($4::bool OR public = true) - AND ($6::text IS NULL OR (CASE source - WHEN 'github' THEN COALESCE( - payload->'repo'->>'name', - payload->'repository'->>'full_name', - payload->>'_repo' - ) - WHEN 'gitea' THEN COALESCE( - payload->'repo'->>'full_name', - payload->'repo'->>'name' - ) - WHEN 'hg' THEN payload->>'_repo' - WHEN 'bugzilla' THEN payload->>'product' - ELSE NULL - END) = $6) + AND ($6::text IS NULL OR repo = $6) ORDER BY occurred_at DESC LIMIT $5 "#, @@ -149,21 +170,7 @@ impl EventReader for PgStore { MIN(occurred_at) AS first_activity, MAX(occurred_at) AS last_activity FROM ( - SELECT source, occurred_at, - CASE source - WHEN 'github' THEN COALESCE( - payload->'repo'->>'name', - payload->'repository'->>'full_name', - payload->>'_repo' - ) - WHEN 'gitea' THEN COALESCE( - payload->'repo'->>'full_name', - payload->'repo'->>'name' - ) - WHEN 'hg' THEN payload->>'_repo' - WHEN 'bugzilla' THEN payload->>'product' - ELSE NULL - END AS repo, + SELECT source, occurred_at, repo, CASE source WHEN 'github' THEN 'github.com' WHEN 'gitea' THEN COALESCE(payload->>'_host', 'git.lair.cafe') @@ -275,18 +282,7 @@ impl EventReader for PgStore { AND e.action IN ('Commit', 'PushEvent', 'commit_repo') JOIN repo_languages rl ON rl.source = e.source - AND rl.repo = CASE e.source - WHEN 'github' THEN COALESCE( - e.payload->'repo'->>'name', - e.payload->'repository'->>'full_name', - e.payload->>'_repo' - ) - WHEN 'gitea' THEN COALESCE( - e.payload->'repo'->>'full_name', - e.payload->'repo'->>'name' - ) - ELSE NULL - END + AND rl.repo = e.repo JOIN LATERAL ( SELECT SUM(bytes)::float AS total FROM repo_languages r2 @@ -336,21 +332,7 @@ impl EventReader for PgStore { SELECT date_trunc($3, occurred_at AT TIME ZONE 'UTC')::date AS period, source, repo, COUNT(*)::bigint AS count FROM ( - SELECT source, occurred_at, - CASE source - WHEN 'github' THEN COALESCE( - payload->'repo'->>'name', - payload->'repository'->>'full_name', - payload->>'_repo' - ) - WHEN 'gitea' THEN COALESCE( - payload->'repo'->>'full_name', - payload->'repo'->>'name' - ) - WHEN 'hg' THEN payload->>'_repo' - WHEN 'bugzilla' THEN payload->>'product' - ELSE NULL - END AS repo + SELECT source, occurred_at, repo FROM events WHERE occurred_at >= ($1::date::timestamp AT TIME ZONE 'UTC') AND occurred_at < (($2::date + 1)::timestamp AT TIME ZONE 'UTC') @@ -434,16 +416,27 @@ impl EventReader for PgStore { } async fn repo_languages(&self) -> Result, StoreError> { + // `repo_languages` is populated for every repo the worker discovers, + // private ones included — this response carries repo names, so it + // needs the same gate the timeline has. Rather than a second + // visibility column to keep in sync, derive it: a repo's languages + // are exposed exactly when at least one of its events still is. let rows = sqlx::query( r#" - SELECT source, repo, language, bytes, - COALESCE(color, + SELECT rl.source, rl.repo, rl.language, rl.bytes, + COALESCE(rl.color, (SELECT color FROM repo_languages r2 - WHERE r2.language = repo_languages.language AND r2.color IS NOT NULL + WHERE r2.language = rl.language AND r2.color IS NOT NULL LIMIT 1) ) AS color - FROM repo_languages - ORDER BY repo, bytes DESC + FROM repo_languages rl + WHERE EXISTS ( + SELECT 1 FROM events e + WHERE e.source = rl.source + AND e.repo = rl.repo + AND e.public = true + ) + ORDER BY rl.repo, rl.bytes DESC "#, ) .fetch_all(&self.pool) @@ -587,6 +580,69 @@ impl EventWriter for PgStore { Ok(n as usize) } + async fn repo_visibility( + &self, + source: Source, + host: Option<&str>, + ) -> Result, StoreError> { + let rows = sqlx::query( + r#" + SELECT repo, bool_or(public) AS public + FROM events + WHERE source = $1 + AND repo IS NOT NULL AND repo <> '' + AND ($2::text IS NULL OR COALESCE(payload->>'_host', $2) = $2) + GROUP BY repo + ORDER BY repo + "#, + ) + .bind(source.as_str()) + .bind(host) + .fetch_all(&self.pool) + .await + .map_err(map_err)?; + + rows.into_iter() + .map(|r| { + Ok(RepoVisibility { + repo: r.try_get("repo").map_err(map_err)?, + public: r.try_get("public").map_err(map_err)?, + }) + }) + .collect() + } + + async fn set_repo_visibility( + &self, + source: Source, + repos: &[RepoVisibility], + ) -> Result { + if repos.is_empty() { + return Ok(0); + } + let names: Vec<&str> = repos.iter().map(|r| r.repo.as_str()).collect(); + let public: Vec = repos.iter().map(|r| r.public).collect(); + + let n = sqlx::query( + r#" + UPDATE events e + SET public = v.public + FROM unnest($2::text[], $3::bool[]) AS v(repo, public) + WHERE e.source = $1 + AND e.repo = v.repo + AND e.public IS DISTINCT FROM v.public + "#, + ) + .bind(source.as_str()) + .bind(&names) + .bind(&public) + .execute(&self.pool) + .await + .map_err(map_err)? + .rows_affected(); + Ok(n) + } + async fn upsert_repo_languages(&self, languages: &[RepoLanguage]) -> Result { if languages.is_empty() { return Ok(0); @@ -620,3 +676,31 @@ impl EventWriter for PgStore { Ok(count) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn probe_verdict_reads_body_on_ok() { + assert_eq!(probe_verdict(200), None); + } + + #[test] + fn probe_verdict_hides_unreachable_repos() { + assert_eq!(probe_verdict(404), Some(Probe::Private)); + } + + #[test] + fn probe_verdict_never_flips_on_rate_limit_or_outage() { + assert_eq!(probe_verdict(403), Some(Probe::RateLimited)); + assert_eq!(probe_verdict(429), Some(Probe::RateLimited)); + for status in [500, 502, 503, 401, 301] { + assert_eq!( + probe_verdict(status), + Some(Probe::Unknown), + "status {status} must leave visibility unchanged" + ); + } + } +} diff --git a/crates/moments-entities/src/lib.rs b/crates/moments-entities/src/lib.rs index b72aa99..ba51304 100644 --- a/crates/moments-entities/src/lib.rs +++ b/crates/moments-entities/src/lib.rs @@ -71,7 +71,8 @@ pub struct EventQuery { pub from: Option>, pub to: Option>, pub sources: Option>, - /// Filter to events matching a specific repo (matched against payload). + /// Filter to events matching a specific repo, against the `repo` column + /// the DB derives from each payload. pub repo: Option, /// When false (default), only `public = true` rows are returned. The API /// pins this to false today; a future authenticated path can flip it. @@ -175,6 +176,15 @@ pub struct LanguageDailyCount { pub commits: i64, } +/// Current exposure of one repo. Used in both directions by the worker's +/// visibility reconciliation: read back from the store as "is any event for +/// this repo still public", and written as "this is what upstream says now". +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RepoVisibility { + pub repo: String, + pub public: bool, +} + /// Per-repo language breakdown from the forge. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RepoLanguage { diff --git a/readme.md b/readme.md index 10a9708..4ef0300 100644 --- a/readme.md +++ b/readme.md @@ -81,6 +81,16 @@ DATABASE_URL=postgres://localhost/moments cargo run -p moments-api migrations live in `crates/moments-data/migrations/` and run automatically on worker startup. the api connects as `moments_ro` and never runs migrations — the worker (as `moments_rw`) is the schema owner. +private work is counted but never described: `events.public` gates the detail +endpoints (`events`, `projects`, `activity/summary`, `languages/repos`) while the +count endpoints (`activity/daily`, `activity/hourly`, `sources`, `languages/daily`) +include everything, so a private repo shows up as volume on the contribution graph +without leaking its name or commit messages. because that flag is stamped at ingest +and the pollers are all incremental, the github and gitea sources re-read current +repo visibility on each poll and rewrite `events.public` across the affected repo's +whole history — a repo flipped to private upstream stops being described, one +flipped back to public reappears. + ## deployment deployment is driven by Gitea Actions, not an operator workstation: