fix: reconcile repo visibility instead of trusting the ingest-time flag

`events.public` was decided once, when a row was ingested, from whatever
the forge reported at that moment — and every poller is incremental (the
github events feed caps at 90 days, search at its top-1000 window, the
per-repo scanner at a `since` cursor, the gitea feed at page 1 after the
first run). Nothing ever revisited a repo, so flipping one to private
upstream only relabelled whatever activity happened afterwards: its
history kept serving commit messages, issue titles and the repo name
indefinitely. The reverse flip was equally frozen.

The github and gitea sources now run a reconciliation pass before
ingesting. github reuses its existing repo discovery, which already
re-reads `private`/`isPrivate` for everything reachable, and only spends
a request on repos we're still exposing that discovery didn't return;
gitea has no equivalent bulk endpoint, so it asks per repo, once a day
rather than once a tick. Repos already hidden are skipped — they can't
leak, and staying hidden is the safe direction to err in. A 404 counts
as private (with the user's own token, a repo still in reach answers 200
even when private, so 404 means gone or transferred), while rate limits
and transient errors flip nothing; the pass runs in both directions, so
a spurious hide is undone by the next successful poll.

That needs a repo key the worker can UPDATE against, hence `events.repo`
— a stored generated column, and now the single definition of the
payload -> repo mapping that list_events, list_projects,
activity_summary and language_daily_counts each carried their own copy
of. Consolidating them fixes an attribution gap along the way:
/search/issues items carry neither `repo.name` nor
`repository.full_name`, only `repository_url`, so every issue and PR
backfilled through search resolved to NULL in all four queries. Those
events now attach to their repo, which both makes them reconcilable and
means they show up in /projects and /activity/summary.

Also closes a leak that predates the flip problem: `/v1/languages/repos`
had no visibility gate at all, and repo_languages is populated for every
repo the worker discovers, private ones included. Repo names were on the
wire (and baked into the prerendered HTML via the dehydrated query
cache) regardless of what `events.public` said. Rather than a second
visibility column to keep in sync, the response now derives it — a
repo's languages are exposed exactly when at least one of its events is.

Verified against postgres 16: the generated column extracts every
payload shape the four sources produce (and NULLs a non-github
`repository_url`), the reconciliation UPDATE flips all three github
event shapes for a repo in one statement and is a no-op on re-run, the
gitea host filter excludes other hosts while treating rows predating the
`_host` stamp as local, and /v1/languages/repos drops a repo once its
events go private.

Closes #6

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<usize, StoreError>;
+
+    /// 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<Vec<RepoVisibility>, 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<u64, StoreError>;
 }

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);

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<RepoVisibility> = 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::<Value>().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<String>) -> Result<usize, SourceError> {
         let mut total = 0usize;
@@ -238,6 +348,13 @@ impl EventSource for GiteaSource {
     async fn poll(&self) -> Result<usize, SourceError> {
         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?;

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<String>,
     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<RepoVisibility> = 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::<Value>().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 {

crates/moments-data/src/hg.rs

@@ -254,6 +254,20 @@ mod tests {
         ) -> Result<usize, moments_core::StoreError> {
             Ok(0)
         }
+        async fn repo_visibility(
+            &self,
+            _source: moments_entities::Source,
+            _host: Option<&str>,
+        ) -> Result<Vec<moments_entities::RepoVisibility>, moments_core::StoreError> {
+            Ok(vec![])
+        }
+        async fn set_repo_visibility(
+            &self,
+            _source: moments_entities::Source,
+            _repos: &[moments_entities::RepoVisibility],
+        ) -> Result<u64, moments_core::StoreError> {
+            Ok(0)
+        }
     }
     struct NoopState;
     #[async_trait]

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: std::fmt::Display>(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<Probe> {
+    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<Vec<Event>, 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<Vec<RepoLanguage>, 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<Vec<RepoVisibility>, 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<u64, StoreError> {
+        if repos.is_empty() {
+            return Ok(0);
+        }
+        let names: Vec<&str> = repos.iter().map(|r| r.repo.as_str()).collect();
+        let public: Vec<bool> = 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<usize, StoreError> {
         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"
+            );
+        }
+    }
+}

crates/moments-entities/src/lib.rs

@@ -71,7 +71,8 @@ pub struct EventQuery {
     pub from: Option<DateTime<Utc>>,
     pub to: Option<DateTime<Utc>>,
     pub sources: Option<Vec<Source>>,
-    /// 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<String>,
     /// 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 {