diff --git a/Cargo.lock b/Cargo.lock index 5f675b6a0..b7297d609 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5855,11 +5855,13 @@ dependencies = [ name = "knotmirror-db" version = "0.0.1" dependencies = [ + "chrono", "jacquard-common", "sqlx", "thiserror 2.0.18", "tokio", "tracing", + "url", ] [[package]] @@ -5879,6 +5881,8 @@ name = "knotmirror-index" version = "0.0.1" dependencies = [ "bobbin-runtime", + "chrono", + "gix", "jacquard-common", "knot-git", "knot-langs", @@ -5889,6 +5893,7 @@ dependencies = [ "reqwest 0.13.5", "scc", "serde_json", + "tempfile", "thiserror 2.0.18", "tokio", "tokio-util", @@ -5974,8 +5979,10 @@ version = "0.0.1" dependencies = [ "anyhow", "axum", + "base64 0.22.1", "bobbin-knot-proxy", "bobbin-runtime", + "bs58", "chrono", "deadpool-redis", "fixedbitset", @@ -5989,6 +5996,7 @@ dependencies = [ "jacquard-axum", "jacquard-common", "jacquard-identity", + "k256", "knot-cob", "knot-cobs", "knot-git", @@ -5998,10 +6006,12 @@ dependencies = [ "knot-types", "knotmirror-db", "knotmirror-git", + "knotmirror-types", "lexicons", "line-numbers", "metrics", "metrics-exporter-prometheus", + "moka", "quick_cache", "reqwest 0.13.5", "rustc-hash", diff --git a/knotmirror/crates/knotmirror-db/.sqlx/query-77c15f179c796a3a8d1d6e3049797238098885401886cc6aab8d386405c1370c.json b/knotmirror/crates/knotmirror-db/.sqlx/query-77c15f179c796a3a8d1d6e3049797238098885401886cc6aab8d386405c1370c.json new file mode 100644 index 000000000..f399f70b0 --- /dev/null +++ b/knotmirror/crates/knotmirror-db/.sqlx/query-77c15f179c796a3a8d1d6e3049797238098885401886cc6aab8d386405c1370c.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "delete from commit_counts where month < $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Date" + ] + }, + "nullable": [] + }, + "hash": "77c15f179c796a3a8d1d6e3049797238098885401886cc6aab8d386405c1370c" +} diff --git a/knotmirror/crates/knotmirror-db/.sqlx/query-886c1448ba85e0aba97c0017c24120962bc21fe3a3303e88bcb7d071d817c253.json b/knotmirror/crates/knotmirror-db/.sqlx/query-886c1448ba85e0aba97c0017c24120962bc21fe3a3303e88bcb7d071d817c253.json new file mode 100644 index 000000000..66f4ff775 --- /dev/null +++ b/knotmirror/crates/knotmirror-db/.sqlx/query-886c1448ba85e0aba97c0017c24120962bc21fe3a3303e88bcb7d071d817c253.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "update repos\n set activity_synced_at = $2, activity_tips = $3, activity_error = null,\n activity_retry_count = 0, activity_retry_after_us = null\n where did = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Timestamptz", + "TextArray" + ] + }, + "nullable": [] + }, + "hash": "886c1448ba85e0aba97c0017c24120962bc21fe3a3303e88bcb7d071d817c253" +} diff --git a/knotmirror/crates/knotmirror-db/.sqlx/query-c0899b71414aaa2fb7885289cb34f120f79ba4de8b635035ea61883ee1ff4595.json b/knotmirror/crates/knotmirror-db/.sqlx/query-c0899b71414aaa2fb7885289cb34f120f79ba4de8b635035ea61883ee1ff4595.json new file mode 100644 index 000000000..e1cb984f7 --- /dev/null +++ b/knotmirror/crates/knotmirror-db/.sqlx/query-c0899b71414aaa2fb7885289cb34f120f79ba4de8b635035ea61883ee1ff4595.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "update repos\n set activity_error = $2, activity_retry_count = activity_retry_count + 1,\n activity_retry_after_us = $3\n where did = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "c0899b71414aaa2fb7885289cb34f120f79ba4de8b635035ea61883ee1ff4595" +} diff --git a/knotmirror/crates/knotmirror-db/.sqlx/query-c15b02adc2ecf9a6f65e30af9ac43c3d663f73924e887886e15aa971c79f5053.json b/knotmirror/crates/knotmirror-db/.sqlx/query-c15b02adc2ecf9a6f65e30af9ac43c3d663f73924e887886e15aa971c79f5053.json new file mode 100644 index 000000000..3c7f124ff --- /dev/null +++ b/knotmirror/crates/knotmirror-db/.sqlx/query-c15b02adc2ecf9a6f65e30af9ac43c3d663f73924e887886e15aa971c79f5053.json @@ -0,0 +1,25 @@ +{ + "db_name": "PostgreSQL", + "query": "with input as (\n select distinct on (oid) oid, committer, month\n from unnest($1::bytea[], $2::text[], $3::date[]) as t(oid, committer, month)\n ), inserted as (\n insert into seen_commits (oid, month)\n select oid, month from input\n on conflict (oid) do nothing\n returning oid\n ), counted as (\n insert into commit_counts (committer, month, commits)\n select committer, month, count(*)\n from input join inserted using (oid)\n group by committer, month\n on conflict (committer, month) do update\n set commits = commit_counts.commits + excluded.commits\n )\n select count(*) as \"count!\" from inserted", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count!", + "type_info": "Int8", + "origin": "Expression" + } + ], + "parameters": { + "Left": [ + "ByteaArray", + "TextArray", + "DateArray" + ] + }, + "nullable": [ + null + ] + }, + "hash": "c15b02adc2ecf9a6f65e30af9ac43c3d663f73924e887886e15aa971c79f5053" +} diff --git a/knotmirror/crates/knotmirror-db/.sqlx/query-d2272d8970ee58876b35266ab2ba3997759e6abe4478ea7020bbe9cf133168bc.json b/knotmirror/crates/knotmirror-db/.sqlx/query-d2272d8970ee58876b35266ab2ba3997759e6abe4478ea7020bbe9cf133168bc.json new file mode 100644 index 000000000..adaca0657 --- /dev/null +++ b/knotmirror/crates/knotmirror-db/.sqlx/query-d2272d8970ee58876b35266ab2ba3997759e6abe4478ea7020bbe9cf133168bc.json @@ -0,0 +1,65 @@ +{ + "db_name": "PostgreSQL", + "query": "select did, updated_at, activity_tips, activity_retry_count\n from repos\n where deleted_at is null\n and state = 'active'\n and activity_synced_at is distinct from updated_at\n and (activity_retry_after_us is null or activity_retry_after_us <= $1)\n order by activity_retry_after_us nulls first, updated_at\n limit $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "did", + "type_info": "Text", + "origin": { + "Table": { + "table": "repos", + "name": "did" + } + } + }, + { + "ordinal": 1, + "name": "updated_at", + "type_info": "Timestamptz", + "origin": { + "Table": { + "table": "repos", + "name": "updated_at" + } + } + }, + { + "ordinal": 2, + "name": "activity_tips", + "type_info": "TextArray", + "origin": { + "Table": { + "table": "repos", + "name": "activity_tips" + } + } + }, + { + "ordinal": 3, + "name": "activity_retry_count", + "type_info": "Int4", + "origin": { + "Table": { + "table": "repos", + "name": "activity_retry_count" + } + } + } + ], + "parameters": { + "Left": [ + "Int8", + "Int8" + ] + }, + "nullable": [ + false, + false, + false, + false + ] + }, + "hash": "d2272d8970ee58876b35266ab2ba3997759e6abe4478ea7020bbe9cf133168bc" +} diff --git a/knotmirror/crates/knotmirror-db/.sqlx/query-d947bb488a5ec5f7f30a17d4faf90746f85e8fc0d88ed641ec9f901ad3fcacfa.json b/knotmirror/crates/knotmirror-db/.sqlx/query-d947bb488a5ec5f7f30a17d4faf90746f85e8fc0d88ed641ec9f901ad3fcacfa.json new file mode 100644 index 000000000..86d4bf4fb --- /dev/null +++ b/knotmirror/crates/knotmirror-db/.sqlx/query-d947bb488a5ec5f7f30a17d4faf90746f85e8fc0d88ed641ec9f901ad3fcacfa.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "delete from seen_commits where month < $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Date" + ] + }, + "nullable": [] + }, + "hash": "d947bb488a5ec5f7f30a17d4faf90746f85e8fc0d88ed641ec9f901ad3fcacfa" +} diff --git a/knotmirror/crates/knotmirror-db/.sqlx/query-e680cd589b5830abdbf1eaad8be5f0ea4a5d3323784cd4f261efb99aa02679fc.json b/knotmirror/crates/knotmirror-db/.sqlx/query-e680cd589b5830abdbf1eaad8be5f0ea4a5d3323784cd4f261efb99aa02679fc.json new file mode 100644 index 000000000..20d1ae9da --- /dev/null +++ b/knotmirror/crates/knotmirror-db/.sqlx/query-e680cd589b5830abdbf1eaad8be5f0ea4a5d3323784cd4f261efb99aa02679fc.json @@ -0,0 +1,38 @@ +{ + "db_name": "PostgreSQL", + "query": "select month, sum(commits)::bigint as \"commits!\"\n from commit_counts\n where committer = any($1) and month >= $2 and month < $3\n group by month\n order by month desc\n limit $4", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "month", + "type_info": "Date", + "origin": { + "Table": { + "table": "commit_counts", + "name": "month" + } + } + }, + { + "ordinal": 1, + "name": "commits!", + "type_info": "Int8", + "origin": "Expression" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Date", + "Date", + "Int8" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "e680cd589b5830abdbf1eaad8be5f0ea4a5d3323784cd4f261efb99aa02679fc" +} diff --git a/knotmirror/crates/knotmirror-db/Cargo.toml b/knotmirror/crates/knotmirror-db/Cargo.toml index 5f601ce09..59f0a185c 100644 --- a/knotmirror/crates/knotmirror-db/Cargo.toml +++ b/knotmirror/crates/knotmirror-db/Cargo.toml @@ -6,12 +6,14 @@ license.workspace = true rust-version.workspace = true [dependencies] +chrono = { workspace = true } jacquard-common = { workspace = true } sqlx = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } [dev-dependencies] +url = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } [lints] diff --git a/knotmirror/crates/knotmirror-db/migrations/20261029000000_commit_activity.sql b/knotmirror/crates/knotmirror-db/migrations/20261029000000_commit_activity.sql new file mode 100644 index 000000000..1b5dc5b60 --- /dev/null +++ b/knotmirror/crates/knotmirror-db/migrations/20261029000000_commit_activity.sql @@ -0,0 +1,27 @@ +alter table repos + add column activity_synced_at timestamptz, + add column activity_tips text[] not null default '{}', + add column activity_error text, + add column activity_retry_count integer not null default 0, + add column activity_retry_after_us bigint; + +-- only mark_synced moves updated_at on an active repo, and a null activity_synced_at is the backfill +create index if not exists idx_repos_activity_queue + on repos (activity_retry_after_us nulls first, updated_at) + where deleted_at is null and state = 'active' + and activity_synced_at is distinct from updated_at; + +-- one row per oid across every repo, which keeps forks from counting a commit twice +create table if not exists seen_commits ( + oid bytea primary key, + month date not null +); + +create index if not exists idx_seen_commits_month on seen_commits (month); + +create table if not exists commit_counts ( + committer text not null, + month date not null, + commits bigint not null check (commits >= 0), + primary key (committer, month) +); diff --git a/knotmirror/crates/knotmirror-db/src/lib.rs b/knotmirror/crates/knotmirror-db/src/lib.rs index 7bd5ac484..88e18681c 100644 --- a/knotmirror/crates/knotmirror-db/src/lib.rs +++ b/knotmirror/crates/knotmirror-db/src/lib.rs @@ -1,5 +1,6 @@ use std::borrow::Cow; +use chrono::{DateTime, Datelike as _, Months, NaiveDate, Utc}; use jacquard_common::deps::bytes::Bytes; use jacquard_common::deps::smol_str::SmolStr; use jacquard_common::types::did::Did; @@ -88,6 +89,58 @@ pub struct IndexedLanguagesPage { pub next: Option, } +/// also the most months `getCommitStats` accepts +pub const ACTIVITY_MONTHS: u32 = 24; + +/// utc months, `before` excluded +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct MonthWindow { + pub since: NaiveDate, + pub before: NaiveDate, +} + +impl MonthWindow { + /// the last `months` months, counting the one `now_us` is in + pub fn ending_at(now_us: i64, months: u32) -> Self { + let current = month_start( + DateTime::from_timestamp_micros(now_us) + .unwrap_or_default() + .date_naive(), + ); + Self { + since: current - Months::new(months.saturating_sub(1)), + before: current + Months::new(1), + } + } + + pub fn month_of(&self, seconds: i64) -> Option { + let date = DateTime::from_timestamp(seconds, 0)?.date_naive(); + (self.since..self.before) + .contains(&date) + .then(|| month_start(date)) + } +} + +fn month_start(date: NaiveDate) -> NaiveDate { + date.with_day(1).expect("every month has a first day") +} + +#[derive(Debug)] +pub struct ActivityRow { + pub did: Did, + pub synced_at: DateTime, + /// branch tips already counted + pub tips: Vec, + pub retry_count: i32, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CommitActivity { + pub oid: Vec, + pub committer: String, + pub month: NaiveDate, +} + #[derive(Debug)] pub struct RepoRow { pub did: Did, @@ -530,4 +583,154 @@ impl Db { rows.into_iter().map(|r| (r.language, r.size)).collect(), )) } + + pub async fn pending_activity( + &self, + now_us: i64, + limit: i64, + ) -> Result, DbError> { + let rows = sqlx::query!( + "select did, updated_at, activity_tips, activity_retry_count + from repos + where deleted_at is null + and state = 'active' + and activity_synced_at is distinct from updated_at + and (activity_retry_after_us is null or activity_retry_after_us <= $1) + order by activity_retry_after_us nulls first, updated_at + limit $2", + now_us, + limit, + ) + .fetch_all(&self.0) + .await?; + Ok(rows + .into_iter() + .filter_map(|r| { + let did = Did::new_owned(&r.did) + .inspect_err(|e| warn!(did = %r.did, error = %e, "unparseable did in repos")) + .ok()?; + Some(ActivityRow { + did, + synced_at: r.updated_at, + tips: r.activity_tips, + retry_count: r.activity_retry_count, + }) + }) + .collect()) + } + + /// returns how many commits no repo had shown before + pub async fn save_commit_activity(&self, commits: &[CommitActivity]) -> Result { + let (oids, committers, months): (Vec<&[u8]>, Vec<&str>, Vec) = commits + .iter() + .map(|c| (c.oid.as_slice(), c.committer.as_str(), c.month)) + .collect(); + let inserted = sqlx::query_scalar!( + r#"with input as ( + select distinct on (oid) oid, committer, month + from unnest($1::bytea[], $2::text[], $3::date[]) as t(oid, committer, month) + ), inserted as ( + insert into seen_commits (oid, month) + select oid, month from input + on conflict (oid) do nothing + returning oid + ), counted as ( + insert into commit_counts (committer, month, commits) + select committer, month, count(*) + from input join inserted using (oid) + group by committer, month + on conflict (committer, month) do update + set commits = commit_counts.commits + excluded.commits + ) + select count(*) as "count!" from inserted"#, + &oids as &[&[u8]], + &committers as &[&str], + &months, + ) + .fetch_one(&self.0) + .await?; + Ok(inserted as u64) + } + + pub async fn finish_activity( + &self, + did: &Did, + synced_at: DateTime, + tips: &[String], + ) -> Result<(), DbError> { + sqlx::query!( + "update repos + set activity_synced_at = $2, activity_tips = $3, activity_error = null, + activity_retry_count = 0, activity_retry_after_us = null + where did = $1", + did.as_str(), + synced_at, + tips, + ) + .execute(&self.0) + .await?; + Ok(()) + } + + pub async fn mark_activity_failed( + &self, + did: &Did, + error: &str, + retry_after_us: i64, + ) -> Result<(), DbError> { + let error = if error.contains('\0') { + Cow::Owned(error.replace('\0', "")) + } else { + Cow::Borrowed(error) + }; + sqlx::query!( + "update repos + set activity_error = $2, activity_retry_count = activity_retry_count + 1, + activity_retry_after_us = $3 + where did = $1", + did.as_str(), + &*error, + retry_after_us, + ) + .execute(&self.0) + .await?; + warn!(%did, retry_after_us, error = %error, "counting commits failed, backing off"); + Ok(()) + } + + pub async fn prune_commit_activity(&self, before: NaiveDate) -> Result<(), DbError> { + let mut tx = self.0.begin().await?; + sqlx::query!("delete from seen_commits where month < $1", before) + .execute(&mut *tx) + .await?; + sqlx::query!("delete from commit_counts where month < $1", before) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } + + pub async fn commit_counts( + &self, + committers: &[String], + since: NaiveDate, + before: NaiveDate, + limit: i64, + ) -> Result, DbError> { + let rows = sqlx::query!( + r#"select month, sum(commits)::bigint as "commits!" + from commit_counts + where committer = any($1) and month >= $2 and month < $3 + group by month + order by month desc + limit $4"#, + committers, + since, + before, + limit, + ) + .fetch_all(&self.0) + .await?; + Ok(rows.into_iter().map(|r| (r.month, r.commits)).collect()) + } } diff --git a/knotmirror/crates/knotmirror-db/tests/commit_activity.rs b/knotmirror/crates/knotmirror-db/tests/commit_activity.rs new file mode 100644 index 000000000..ce4518af5 --- /dev/null +++ b/knotmirror/crates/knotmirror-db/tests/commit_activity.rs @@ -0,0 +1,182 @@ +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use chrono::{Months, NaiveDate}; +use jacquard_common::deps::bytes::Bytes; +use jacquard_common::types::did::Did; +use knotmirror_db::{CommitActivity, Db, GitRefs, MonthWindow, RefsUpsert}; + +const ACTOR: &str = "did:plc:actor"; + +async fn test_db() -> Option { + let Ok(url) = std::env::var("KNOTMIRROR_TEST_DATABASE_URL") else { + return None; + }; + static NEXT: AtomicUsize = AtomicUsize::new(0); + let name = format!( + "commit_activity_{}_{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_micros(), + NEXT.fetch_add(1, Ordering::Relaxed) + ); + let admin = Db::connect(&url, 1).await.unwrap(); + sqlx::query(sqlx::AssertSqlSafe(format!("create database {name}"))) + .execute(admin.pool()) + .await + .unwrap(); + let mut url = url::Url::parse(&url).unwrap(); + url.set_path(&name); + let db = Db::connect(url.as_str(), 4).await.unwrap(); + db.migrate().await.unwrap(); + Some(db) +} + +fn commit(oid: u8, committer: &str, month: NaiveDate) -> CommitActivity { + CommitActivity { + oid: vec![oid; 20], + committer: committer.to_owned(), + month, + } +} + +async fn synced_repo(db: &Db, did: &Did) { + db.upsert_refs(RefsUpsert { + did, + refs_cid: "bafyrefs", + refs: &GitRefs { + default_branch: "main".into(), + refs: Bytes::new(), + }, + }) + .await + .unwrap(); + sync_pending(db, did).await; +} + +async fn sync_pending(db: &Db, did: &Did) { + let row = db + .pending_repos(i64::MAX, 10) + .await + .unwrap() + .into_iter() + .find(|row| &row.did == did) + .unwrap(); + db.mark_synced(&row, None).await.unwrap(); +} + +#[tokio::test] +async fn a_commit_counts_once_however_often_it_is_seen() { + let Some(db) = test_db().await else { return }; + let month = NaiveDate::from_ymd_opt(2026, 9, 1).unwrap(); + let commits = [ + commit(0xaa, ACTOR, month), + commit(0xbb, "person@example.com", month), + commit(0xcc, "person@example.com", month), + ]; + assert_eq!(db.save_commit_activity(&commits).await.unwrap(), 3); + assert_eq!(db.save_commit_activity(&commits).await.unwrap(), 0); + + let committers = [ACTOR.to_owned(), "person@example.com".to_owned()]; + let counts = db + .commit_counts(&committers, month, month + Months::new(1), 24) + .await + .unwrap(); + assert_eq!(counts, [(month, 3)]); + let counts = db + .commit_counts(&committers[1..], month, month + Months::new(1), 24) + .await + .unwrap(); + assert_eq!(counts, [(month, 2)]); +} + +#[tokio::test] +async fn a_synced_repo_is_counted_until_its_next_sync() { + let Some(db) = test_db().await else { return }; + let did = Did::new_owned("did:plc:repo").unwrap(); + synced_repo(&db, &did).await; + + let pending = db.pending_activity(i64::MAX, 10).await.unwrap(); + assert_eq!(pending.len(), 1, "an already mirrored repo is backfilled"); + assert!(pending[0].tips.is_empty()); + + let tips = ["d".repeat(40)]; + db.finish_activity(&did, pending[0].synced_at, &tips) + .await + .unwrap(); + assert!(db.pending_activity(i64::MAX, 10).await.unwrap().is_empty()); + + db.bump_rev(&did, "3lrev").await.unwrap(); + sync_pending(&db, &did).await; + let pending = db.pending_activity(i64::MAX, 10).await.unwrap(); + assert_eq!(pending.len(), 1, "a new sync queues the repo again"); + assert_eq!(pending[0].tips, tips); +} + +#[tokio::test] +async fn a_failed_count_waits_out_its_backoff() { + let Some(db) = test_db().await else { return }; + let did = Did::new_owned("did:plc:repo").unwrap(); + synced_repo(&db, &did).await; + db.mark_activity_failed(&did, "boom", 1_000).await.unwrap(); + assert!(db.pending_activity(999, 10).await.unwrap().is_empty()); + let pending = db.pending_activity(1_000, 10).await.unwrap(); + assert_eq!(pending[0].retry_count, 1); +} + +#[tokio::test] +async fn pruning_drops_months_before_the_window() { + let Some(db) = test_db().await else { return }; + let current = NaiveDate::from_ymd_opt(2026, 9, 1).unwrap(); + let old = current - Months::new(24); + db.save_commit_activity(&[commit(0xee, ACTOR, old), commit(0xff, ACTOR, current)]) + .await + .unwrap(); + db.prune_commit_activity(current - Months::new(23)) + .await + .unwrap(); + + for query in [ + "select count(*) from seen_commits", + "select count(*) from commit_counts", + ] { + let rows: i64 = sqlx::query_scalar(query) + .fetch_one(db.pool()) + .await + .unwrap(); + assert_eq!(rows, 1, "{query}"); + } +} + +#[test] +fn the_window_ends_with_the_current_utc_month() { + let at = |s: &str| { + chrono::DateTime::parse_from_rfc3339(s) + .unwrap() + .timestamp_micros() + }; + let day = |y, m, d| NaiveDate::from_ymd_opt(y, m, d).unwrap(); + let window = MonthWindow::ending_at(at("2026-03-31T23:30:00-05:00"), 24); + assert_eq!(window.since, day(2024, 5, 1)); + assert_eq!(window.before, day(2026, 5, 1)); + assert_eq!( + MonthWindow::ending_at(at("2026-03-01T00:00:00Z"), 1), + MonthWindow { + since: day(2026, 3, 1), + before: day(2026, 4, 1) + } + ); + + let seconds = |s: &str| at(s) / 1_000_000; + assert_eq!( + window.month_of(seconds("2026-04-30T23:59:59Z")), + Some(day(2026, 4, 1)) + ); + assert_eq!(window.month_of(seconds("2026-05-01T00:00:00Z")), None); + assert_eq!( + window.month_of(seconds("2024-05-01T00:00:00Z")), + Some(day(2024, 5, 1)) + ); + assert_eq!(window.month_of(seconds("2024-04-30T23:59:59Z")), None); +} diff --git a/knotmirror/crates/knotmirror-index/Cargo.toml b/knotmirror/crates/knotmirror-index/Cargo.toml index 28398b14a..f6efed523 100644 --- a/knotmirror/crates/knotmirror-index/Cargo.toml +++ b/knotmirror/crates/knotmirror-index/Cargo.toml @@ -14,6 +14,7 @@ knot-git = { workspace = true } knot-langs = { workspace = true } knot-types = { workspace = true } +gix = { workspace = true } jacquard-common = { workspace = true } reqwest = { workspace = true } scc = { workspace = true } @@ -23,5 +24,9 @@ tokio = { workspace = true } tokio-util = { workspace = true } tracing = { workspace = true } +[dev-dependencies] +chrono = { workspace = true } +tempfile = { workspace = true } + [lints] workspace = true diff --git a/knotmirror/crates/knotmirror-index/src/activity.rs b/knotmirror/crates/knotmirror-index/src/activity.rs new file mode 100644 index 000000000..bf6756ea5 --- /dev/null +++ b/knotmirror/crates/knotmirror-index/src/activity.rs @@ -0,0 +1,332 @@ +use std::sync::Arc; +use std::time::Duration; + +use bobbin_runtime::Clock; +use gix::ObjectId; +use jacquard_common::types::did::Did; +use knotmirror_db::{ACTIVITY_MONTHS, ActivityRow, CommitActivity, Db, DbError, MonthWindow}; +use knotmirror_git::layout::Layout; +use knotmirror_types::{backoff_us, now_us}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info}; + +use crate::IndexError; + +const BATCH_SIZE: i64 = 64; +const SAVE_CHUNK: usize = 1_000; +const MAX_COMMITTER_LEN: usize = 2_048; +const PRUNE_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60); + +pub struct ActivityRuntime { + pub db: Db, + pub clock: Arc, + pub cancel: CancellationToken, + pub layout: Layout, +} + +pub async fn run(poll_interval: Duration, rt: ActivityRuntime) -> Result<(), IndexError> { + info!("commit activity worker started"); + let mut next_prune = rt.clock.now_instant(); + + while !rt.cancel.is_cancelled() { + if rt.clock.now_instant() >= next_prune { + let window = MonthWindow::ending_at(now_us(&*rt.clock), ACTIVITY_MONTHS); + rt.db.prune_commit_activity(window.since).await?; + next_prune = rt.clock.now_instant() + PRUNE_INTERVAL; + } + + let rows = rt + .db + .pending_activity(now_us(&*rt.clock), BATCH_SIZE) + .await?; + if rows.is_empty() { + tokio::select! { + () = rt.cancel.cancelled() => break, + () = rt.clock.sleep(poll_interval) => {} + } + } + for row in rows { + if rt.cancel.is_cancelled() { + break; + } + count_repo(&rt, row).await; + } + } + + info!("commit activity worker stopped"); + Ok(()) +} + +async fn count_repo(rt: &ActivityRuntime, row: ActivityRow) { + let path = rt.layout.repo_path(&row.did); + let window = MonthWindow::ending_at(now_us(&*rt.clock), ACTIVITY_MONTHS); + let counted = row.tips.clone(); + let scanned = tokio::task::spawn_blocking(move || { + let repo = gix::open(path).map_err(|e| e.to_string())?; + scan(&repo, &counted, window) + }) + .await + .unwrap_or_else(|_| Err("commit scan panicked".to_owned())); + + let write = match scanned { + Ok(scan) => save(&rt.db, &row, scan).await, + Err(e) => { + rt.db + .mark_activity_failed( + &row.did, + &e, + now_us(&*rt.clock) + backoff_us(row.retry_count), + ) + .await + } + }; + if let Err(e) = write { + error!(did = %row.did, error = %e, "recording commit activity failed"); + } +} + +async fn save(db: &Db, row: &ActivityRow, scan: Scan) -> Result<(), DbError> { + let mut new = 0; + for chunk in scan.commits.chunks(SAVE_CHUNK) { + new += db.save_commit_activity(chunk).await?; + } + db.finish_activity(&row.did, row.synced_at, &scan.tips) + .await?; + debug!(did = %row.did, commits = scan.commits.len(), new, "counted commits"); + Ok(()) +} + +#[derive(Debug)] +pub struct Scan { + pub tips: Vec, + pub commits: Vec, +} + +/// a counted tip that no longer exists is ignored, which walks its history again +pub fn scan( + repo: &gix::Repository, + counted: &[String], + window: MonthWindow, +) -> Result { + let mut tips = Vec::new(); + for reference in repo + .references() + .map_err(|e| e.to_string())? + .prefixed("refs/heads/") + .map_err(|e| e.to_string())? + { + tips.push( + reference + .map_err(|e| e.to_string())? + .into_fully_peeled_id() + .map_err(|e| e.to_string())? + .detach(), + ); + } + tips.sort_unstable(); + tips.dedup(); + + let hidden = counted + .iter() + .filter_map(|tip| ObjectId::from_hex(tip.as_bytes()).ok()) + .filter(|tip| repo.has_object(tip)); + let mut commits = Vec::new(); + if !tips.is_empty() { + let walk = repo + .rev_walk(tips.iter().copied()) + .with_hidden(hidden) + .all() + .map_err(|e| e.to_string())?; + for info in walk { + let info = info.map_err(|e| e.to_string())?; + let commit = info.object().map_err(|e| e.to_string())?; + let committer = commit.committer().map_err(|e| e.to_string())?; + let Some(month) = committer + .time() + .ok() + .and_then(|time| window.month_of(time.seconds)) + else { + continue; + }; + let Some(committer) = normalize_committer(&committer.email.to_string()) else { + continue; + }; + commits.push(CommitActivity { + oid: info.id.as_bytes().to_vec(), + committer, + month, + }); + } + } + + Ok(Scan { + tips: tips.iter().map(ToString::to_string).collect(), + commits, + }) +} + +/// dids keep their case and emails get lowercased +fn normalize_committer(email: &str) -> Option { + let email = email.trim(); + if email.is_empty() || email.len() > MAX_COMMITTER_LEN { + return None; + } + Some(match Did::new(email) { + Ok(did) => did.as_str().to_owned(), + Err(_) => email.to_lowercase(), + }) +} + +#[cfg(test)] +mod tests { + use std::path::Path; + use std::process::Command; + + use chrono::{Days, Months, NaiveDate, Utc}; + + use super::*; + + fn git(dir: &Path, env: &[(&str, &str)], args: &[&str]) { + let out = Command::new("git") + .args(args) + .current_dir(dir) + .env("GIT_CONFIG_NOSYSTEM", "1") + .env("GIT_CONFIG_GLOBAL", "/dev/null") + .envs(env.iter().copied()) + .output() + .expect("run git"); + assert!(out.status.success(), "git {args:?}: {out:?}"); + } + + fn commit(dir: &Path, file: &str, committer: &str, at: NaiveDate) { + std::fs::write(dir.join(file), file).unwrap(); + git(dir, &[], &["add", file]); + let date = at.and_hms_opt(12, 0, 0).unwrap().and_utc().to_rfc3339(); + git( + dir, + &[ + ("GIT_AUTHOR_NAME", "author"), + ("GIT_AUTHOR_EMAIL", "not-the-committer@example.com"), + ("GIT_AUTHOR_DATE", &date), + ("GIT_COMMITTER_NAME", "committer"), + ("GIT_COMMITTER_EMAIL", committer), + ("GIT_COMMITTER_DATE", &date), + ], + &["commit", "-q", "-m", file], + ); + } + + struct Fixture { + dir: tempfile::TempDir, + current: NaiveDate, + window: MonthWindow, + } + + impl Fixture { + fn new() -> Self { + let dir = tempfile::tempdir().unwrap(); + let now = Utc::now(); + let window = MonthWindow::ending_at(now.timestamp_micros(), ACTIVITY_MONTHS); + let current = window.before - Months::new(1); + let day = |months| current - Months::new(months) + Days::new(1); + let path = dir.path(); + git(path, &[], &["init", "-q", "-b", "main"]); + commit(path, "base", "did:plc:alice", day(2)); + git(path, &[], &["branch", "feature"]); + commit(path, "main", "did:plc:alice", day(0)); + git(path, &[], &["checkout", "-q", "feature"]); + commit(path, "feature", "did:plc:bob", day(1)); + git(path, &[], &["checkout", "-q", "main"]); + Self { + dir, + current, + window, + } + } + + fn scan(&self, counted: &[String]) -> Scan { + scan(&gix::open(self.dir.path()).unwrap(), counted, self.window).unwrap() + } + + fn commit(&self, file: &str, committer: &str) { + commit( + self.dir.path(), + file, + committer, + self.current + Days::new(2), + ); + } + } + + fn counted(scan: &Scan) -> Vec<(String, NaiveDate)> { + let mut counted: Vec<_> = scan + .commits + .iter() + .map(|c| (c.committer.clone(), c.month)) + .collect(); + counted.sort(); + counted + } + + #[test] + fn every_branch_counts_under_its_committer_and_utc_month() { + let fixture = Fixture::new(); + let scan = fixture.scan(&[]); + let month = |months| fixture.current - Months::new(months); + assert_eq!( + counted(&scan), + [ + ("did:plc:alice".to_owned(), month(2)), + ("did:plc:alice".to_owned(), month(0)), + ("did:plc:bob".to_owned(), month(1)), + ] + ); + assert_eq!(scan.tips.len(), 2); + } + + #[test] + fn counted_tips_hide_their_history() { + let fixture = Fixture::new(); + let before = fixture.scan(&[]); + fixture.commit("new", "did:plc:carol"); + let scan = fixture.scan(&before.tips); + assert_eq!( + counted(&scan), + [("did:plc:carol".to_owned(), fixture.current)] + ); + } + + #[test] + fn a_vanished_counted_tip_walks_everything_again() { + let fixture = Fixture::new(); + let scan = fixture.scan(&["f".repeat(40), "not hex".to_owned()]); + assert_eq!(scan.commits.len(), 3); + } + + #[test] + fn a_branch_sharing_a_tip_is_walked_once() { + let fixture = Fixture::new(); + git(fixture.dir.path(), &[], &["branch", "same-as-main", "main"]); + let scan = fixture.scan(&[]); + assert_eq!(scan.commits.len(), 3); + assert_eq!(scan.tips.len(), 2); + } + + #[test] + fn emails_are_lowercased_and_old_commits_skipped() { + let fixture = Fixture::new(); + fixture.commit("upper", "Person@Example.COM"); + commit( + fixture.dir.path(), + "ancient", + "person@example.com", + fixture.window.since - Days::new(1), + ); + let scan = fixture.scan(&[]); + assert!( + counted(&scan).contains(&("person@example.com".to_owned(), fixture.current)), + "{scan:?}" + ); + assert_eq!(scan.commits.len(), 4); + } +} diff --git a/knotmirror/crates/knotmirror-index/src/lib.rs b/knotmirror/crates/knotmirror-index/src/lib.rs index ea85034b8..008878a36 100644 --- a/knotmirror/crates/knotmirror-index/src/lib.rs +++ b/knotmirror/crates/knotmirror-index/src/lib.rs @@ -12,6 +12,8 @@ use tokio::sync::{Notify, Semaphore}; use tokio_util::sync::CancellationToken; use tracing::{Instrument as _, debug, error, info, warn}; +pub mod activity; + const BATCH_SIZE: i64 = 64; #[derive(Clone, Debug)] @@ -142,13 +144,6 @@ impl Job { ); } - // TODO: index user commit activity (skip when repo is fork) - // from previous index being `` and updated HEAD being ``, - // 1. discard activity indicies between .. (all commits reachable from - // previous and not from head) - // 2. index user activity between .. (all commits reachable from head and - // not from previous) - let langs = match commit { Some(commit) => { index_languages(self.layout.repo_path(&row.did), commit, self.langs_timeout).await diff --git a/knotmirror/crates/knotmirror-xrpc/Cargo.toml b/knotmirror/crates/knotmirror-xrpc/Cargo.toml index 3df8ea996..37fb5506d 100644 --- a/knotmirror/crates/knotmirror-xrpc/Cargo.toml +++ b/knotmirror/crates/knotmirror-xrpc/Cargo.toml @@ -10,6 +10,7 @@ bobbin-knot-proxy = { workspace = true } bobbin-runtime = { workspace = true } knotmirror-db = { workspace = true } knotmirror-git = { workspace = true } +knotmirror-types = { workspace = true } gix-receive-pack = { workspace = true } knot-cob = { workspace = true } knot-cobs = { workspace = true } @@ -21,6 +22,8 @@ tangled-axum = { workspace = true } anyhow = { workspace = true } axum = { workspace = true } +base64 = { workspace = true } +bs58 = { workspace = true } chrono = { workspace = true } deadpool-redis = { workspace = true } jacquard-axum = { workspace = true } @@ -36,6 +39,8 @@ line-numbers = "0.4.0" rustc-hash = "2.1.2" fixedbitset = "0.5" itertools = { workspace = true } +k256 = { workspace = true } +moka = { workspace = true } gix = { version = "0.84", features = ["parallel", "blob-diff", "blame", "merge", "sha1", "sha256", "revision", "tree-editor", "worktree-stream"] } gix-archive = { workspace = true } gix-pack = { workspace = true } diff --git a/knotmirror/crates/knotmirror-xrpc/src/committers.rs b/knotmirror/crates/knotmirror-xrpc/src/committers.rs new file mode 100644 index 000000000..d4b4b5e12 --- /dev/null +++ b/knotmirror/crates/knotmirror-xrpc/src/committers.rs @@ -0,0 +1,116 @@ +use std::sync::Arc; +use std::time::Duration; + +use bobbin_runtime::Clock; +use jacquard_common::types::did::Did; +use lexicons::sh_tangled::identity::resolve_committers::ResolveCommittersOutput; +use serde_json::json; +use tracing::warn; + +use crate::signer::ServiceSigner; + +const RESOLVE_COMMITTERS: &str = "sh.tangled.identity.resolveCommitters"; +const TIMEOUT: Duration = Duration::from_secs(1); +const MAX_RESPONSE_BYTES: usize = 64 * 1024; +const MAX_COMMITTERS: usize = 251; +const MAX_COMMITTER_LEN: usize = 320; +const CACHE_CAPACITY: u64 = 4_096; +const CACHE_FRESHNESS: Duration = Duration::from_secs(30); + +pub struct Deliberi { + pub url: String, + pub did: Did, +} + +/// falls back to only the actor's did when deliberi is unset or fails +pub(crate) struct Committers { + deliberi: Option<(Deliberi, Arc)>, + cache: moka::future::Cache>, +} + +impl Committers { + pub(crate) fn new(deliberi: Option<(Deliberi, Arc)>) -> Self { + Self { + deliberi, + cache: moka::future::Cache::builder() + .max_capacity(CACHE_CAPACITY) + .time_to_live(CACHE_FRESHNESS) + .build(), + } + } + + pub(crate) async fn resolve( + &self, + http: &reqwest::Client, + clock: &dyn Clock, + actor: &Did, + ) -> Arc<[String]> { + let Some((deliberi, signer)) = &self.deliberi else { + return Arc::from([actor.as_str().to_owned()]); + }; + let now_secs = clock.now_unix_micros().raw() / 1_000_000; + let fetch = fetch(http, deliberi, signer, actor, now_secs); + match self.cache.try_get_with(actor.clone(), fetch).await { + Ok(committers) => committers, + Err(e) => { + warn!(%actor, error = %e, "resolving verified committer emails failed"); + Arc::from([actor.as_str().to_owned()]) + } + } + } +} + +async fn fetch( + http: &reqwest::Client, + deliberi: &Deliberi, + signer: &ServiceSigner, + actor: &Did, + now_secs: u64, +) -> Result, String> { + let mut response = http + .post(format!( + "{}/xrpc/{RESOLVE_COMMITTERS}", + deliberi.url.trim_end_matches('/') + )) + .bearer_auth(signer.token(deliberi.did.as_str(), RESOLVE_COMMITTERS, now_secs)) + .json(&json!({ "actor": actor.as_str() })) + .timeout(TIMEOUT) + .send() + .await + .map_err(|e| e.to_string())?; + if !response.status().is_success() { + return Err(format!("deliberi returned {}", response.status())); + } + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await.map_err(|e| e.to_string())? { + if body.len() + chunk.len() > MAX_RESPONSE_BYTES { + return Err("deliberi response is too large".to_owned()); + } + body.extend_from_slice(&chunk); + } + let output: ResolveCommittersOutput = + serde_json::from_slice(&body).map_err(|e| e.to_string())?; + Ok(committers(actor, &output.committers).into()) +} + +/// lowercased because the activity index stores emails lowercased +fn committers(actor: &Did, resolved: &[impl AsRef]) -> Vec { + let mut out = vec![actor.as_str().to_owned()]; + for value in resolved { + let value = value.as_ref().trim().to_lowercase(); + if value.is_empty() || value.len() > MAX_COMMITTER_LEN || value.contains('\0') { + continue; + } + let usable = match Did::new(value.as_str()) { + Ok(did) => did.as_str() == actor.as_str(), + Err(_) => value.contains('@'), + }; + if usable && !out.contains(&value) { + out.push(value); + if out.len() == MAX_COMMITTERS { + break; + } + } + } + out +} diff --git a/knotmirror/crates/knotmirror-xrpc/src/lib.rs b/knotmirror/crates/knotmirror-xrpc/src/lib.rs index 543a9a560..2e289bc23 100644 --- a/knotmirror/crates/knotmirror-xrpc/src/lib.rs +++ b/knotmirror/crates/knotmirror-xrpc/src/lib.rs @@ -1,4 +1,5 @@ mod archive; +mod committers; mod did_ext; mod diff; mod error; @@ -8,6 +9,7 @@ mod last_commit; mod merge; pub mod metrics; mod routes; +mod signer; mod sniff; use std::{path::PathBuf, sync::Arc}; @@ -23,6 +25,10 @@ use jacquard_common::service_auth::ServiceAuthClaims; use jacquard_identity::JacquardResolver; use tangled_axum::atproto::{AtprotoService, ClaimsPolicy, Verified}; +use crate::committers::Committers; +pub use crate::committers::Deliberi; +pub use crate::signer::ServiceSigner; + pub(crate) type Directory = JacquardResolver; #[derive(Clone)] @@ -38,6 +44,8 @@ pub struct AppState { pub(crate) redis: Option, pub(crate) max_blob_bytes: u64, pub(crate) max_archive_bytes: u64, + pub(crate) signer: Option>, + pub(crate) committers: Arc, } impl AppState { @@ -65,8 +73,19 @@ impl AppState { redis, max_blob_bytes, max_archive_bytes, + signer: None, + committers: Arc::new(Committers::new(None)), } } + + pub fn with_signer(mut self, signer: ServiceSigner, deliberi: Option) -> Self { + let signer = Arc::new(signer); + self.committers = Arc::new(Committers::new( + deliberi.map(|deliberi| (deliberi, signer.clone())), + )); + self.signer = Some(signer); + self + } } impl AtprotoService for AppState { @@ -91,7 +110,9 @@ impl AtprotoService for AppState { pub fn router(state: AppState) -> Router { use routes::*; Router::new() + .route("/.well-known/did.json", get(activity::did_document)) .route("/xrpc/sh.tangled.git.temp2.getBlame", get(git::get_blame)) + .route("/xrpc/sh.tangled.git.temp2.getCommitStats", get(activity::get_commit_stats)) .route("/xrpc/sh.tangled.git.temp2.getDiff", get(git::get_diff)) .route("/xrpc/sh.tangled.git.temp2.getInterdiff", get(git::get_interdiff)) .route("/xrpc/sh.tangled.git.temp2.mergeCheck", get(git::merge_check)) diff --git a/knotmirror/crates/knotmirror-xrpc/src/routes/activity.rs b/knotmirror/crates/knotmirror-xrpc/src/routes/activity.rs new file mode 100644 index 000000000..525ff7b22 --- /dev/null +++ b/knotmirror/crates/knotmirror-xrpc/src/routes/activity.rs @@ -0,0 +1,322 @@ +use axum::Json; +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::{IntoResponse as _, Response}; +use jacquard_axum::{ExtractXrpc, XrpcResponse}; +use jacquard_common::types::string::Datetime; +use knotmirror_db::{ACTIVITY_MONTHS, MonthWindow}; +use knotmirror_types::now_us; +use lexicons::sh_tangled::git::temp2::get_commit_stats; + +use crate::{AppState, error::XrpcError}; + +const DEFAULT_MONTHS: i64 = 7; + +pub(crate) async fn get_commit_stats( + State(state): State, + ExtractXrpc(args): ExtractXrpc, +) -> Result, XrpcError> { + let months = args.months.unwrap_or(DEFAULT_MONTHS); + let Some(months) = u32::try_from(months) + .ok() + .filter(|months| (1..=ACTIVITY_MONTHS).contains(months)) + else { + return Err(XrpcError::InvalidRequest(format!( + "months must be between 1 and {ACTIVITY_MONTHS}" + ))); + }; + let window = MonthWindow::ending_at(now_us(&*state.clock), months); + let committers = state + .committers + .resolve(&state.http, &*state.clock, &args.actor) + .await; + let counts = state + .db + .commit_counts(&committers, window.since, window.before, i64::from(months)) + .await + .map_err(|e| XrpcError::Internal(e.to_string()))?; + Ok(XrpcResponse(get_commit_stats::GetCommitStatsOutput { + months: counts + .into_iter() + .map(|(month, commits)| get_commit_stats::Month { + commits, + start: Datetime::new(month.and_time(Default::default()).and_utc().fixed_offset()), + extra_data: Default::default(), + }) + .collect(), + extra_data: Default::default(), + })) +} + +pub(crate) async fn did_document(State(state): State) -> Response { + match &state.signer { + Some(signer) => Json(signer.did_document()).into_response(), + None => StatusCode::NOT_FOUND.into_response(), + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::time::{SystemTime, UNIX_EPOCH}; + + use axum::extract::State; + use axum::http::{HeaderMap, StatusCode}; + use axum::routing::post; + use bobbin_runtime::{ReqwestHttp, SystemClock}; + use chrono::{Months, Utc}; + use jacquard_axum::service_auth::ServiceAuthConfig; + use jacquard_common::service_auth::{PublicKey, parse_jwt, verify_signature}; + use jacquard_common::types::did::Did; + use jacquard_identity::JacquardResolver; + use knotmirror_db::{ACTIVITY_MONTHS, CommitActivity, Db, MonthWindow}; + use serde_json::{Value, json}; + + use crate::{AppState, Deliberi, KnotPolicy, ServiceSigner}; + + const ACTOR: &str = "did:plc:profileactor"; + const MIRROR: &str = "did:web:mirror.example"; + const DELIBERI: &str = "did:web:deliberi.example"; + + fn signer() -> ServiceSigner { + let mut key = vec![0x81, 0x26]; + key.extend_from_slice(&[7; 32]); + let encoded = format!("z{}", bs58::encode(key).into_string()); + ServiceSigner::new(Did::new_owned(MIRROR).unwrap(), &encoded).unwrap() + } + + async fn test_db() -> Option { + let Ok(url) = std::env::var("KNOTMIRROR_TEST_DATABASE_URL") else { + return None; + }; + static NEXT: AtomicUsize = AtomicUsize::new(0); + let name = format!( + "commit_stats_{}_{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_micros(), + NEXT.fetch_add(1, Ordering::Relaxed) + ); + let admin = Db::connect(&url, 1).await.unwrap(); + sqlx::query(sqlx::AssertSqlSafe(format!("create database {name}"))) + .execute(admin.pool()) + .await + .unwrap(); + let mut url = reqwest::Url::parse(&url).unwrap(); + url.set_path(&name); + let db = Db::connect(url.as_str(), 4).await.unwrap(); + db.migrate().await.unwrap(); + Some(db) + } + + #[derive(Clone)] + struct FakeDeliberi { + public_key: String, + calls: Arc, + status: StatusCode, + } + + async fn resolve_committers( + State(fake): State, + headers: HeaderMap, + body: axum::Json, + ) -> (StatusCode, axum::Json) { + fake.calls.fetch_add(1, Ordering::SeqCst); + let token = headers["authorization"] + .to_str() + .unwrap() + .strip_prefix("Bearer ") + .unwrap(); + let jwt = parse_jwt(token).unwrap(); + let multikey = bs58::decode(fake.public_key.strip_prefix('z').unwrap()) + .into_vec() + .unwrap(); + let key = + PublicKey::from_k256_bytes(multikey.strip_prefix(&[0xe7, 0x01]).unwrap()).unwrap(); + verify_signature(&jwt, &key).unwrap(); + let claims = jwt.claims(); + assert_eq!(claims.iss.as_str(), MIRROR); + assert_eq!(claims.aud.as_str(), DELIBERI); + assert_eq!( + claims.lxm.as_ref().map(|lxm| lxm.as_str()), + Some("sh.tangled.identity.resolveCommitters") + ); + assert_eq!(body["actor"], ACTOR); + let committers = json!({"committers": [ + ACTOR, + "Person@Example.com", + "person@example.com", + "not-an-email", + "did:plc:someoneelse", + ]}); + (fake.status, axum::Json(committers)) + } + + async fn serve(router: axum::Router) -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { axum::serve(listener, router).await }); + url + } + + struct Mirror { + url: String, + deliberi_calls: Arc, + _repos: tempfile::TempDir, + } + + async fn mirror(db: Db, deliberi_status: StatusCode) -> Mirror { + let signer = signer(); + let fake = FakeDeliberi { + public_key: signer.did_document()["verificationMethod"][0]["publicKeyMultibase"] + .as_str() + .unwrap() + .to_owned(), + calls: Arc::default(), + status: deliberi_status, + }; + let deliberi_calls = fake.calls.clone(); + let deliberi = serve( + axum::Router::new() + .route( + "/xrpc/sh.tangled.identity.resolveCommitters", + post(resolve_committers), + ) + .with_state(fake), + ) + .await; + + let repos = tempfile::tempdir().unwrap(); + let http = reqwest::Client::new(); + let directory = Arc::new(JacquardResolver::new( + ReqwestHttp::new(http.clone()), + Default::default(), + )); + let state = AppState::new( + db, + Arc::new(repos.path().to_owned()), + http, + ServiceAuthConfig::new(Did::raw(MIRROR.into()), directory), + KnotPolicy { + allow_private: false, + require_https: true, + }, + Arc::new(SystemClock::new()), + None, + 1 << 20, + 1 << 20, + ) + .with_signer( + signer, + Some(Deliberi { + url: deliberi, + did: Did::new_owned(DELIBERI).unwrap(), + }), + ); + Mirror { + url: serve(crate::router(state)).await, + deliberi_calls, + _repos: repos, + } + } + + impl Mirror { + async fn stats(&self, params: &[(&str, &str)]) -> (StatusCode, Value) { + let url = reqwest::Url::parse_with_params( + &format!("{}/xrpc/sh.tangled.git.temp2.getCommitStats", self.url), + params, + ) + .unwrap(); + let response = reqwest::get(url).await.unwrap(); + let status = response.status(); + (status, response.json().await.unwrap_or(Value::Null)) + } + } + + async fn seed(db: &Db) -> [String; 2] { + let window = MonthWindow::ending_at(Utc::now().timestamp_micros(), ACTIVITY_MONTHS); + let current = window.before - Months::new(1); + let commit = |oid: u8, committer: &str, months_back| CommitActivity { + oid: vec![oid; 20], + committer: committer.to_owned(), + month: current - Months::new(months_back), + }; + db.save_commit_activity(&[ + commit(1, ACTOR, 0), + commit(2, "person@example.com", 0), + commit(3, "person@example.com", 1), + commit(4, "person@example.com", 3), + commit(5, "someone@else.example", 0), + ]) + .await + .unwrap(); + let start = + |months_back| format!("{}T00:00:00.000000Z", current - Months::new(months_back)); + [start(0), start(1)] + } + + #[tokio::test] + async fn stats_count_the_did_and_verified_emails() { + let Some(db) = test_db().await else { return }; + let [this_month, last_month] = seed(&db).await; + let mirror = mirror(db, StatusCode::OK).await; + + let (status, body) = mirror.stats(&[("actor", ACTOR), ("months", "2")]).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body, + json!({"months": [ + {"start": this_month, "commits": 2}, + {"start": last_month, "commits": 1}, + ]}) + ); + + let (status, body) = mirror.stats(&[("actor", ACTOR)]).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body["months"].as_array().unwrap().len(), 3); + assert_eq!(mirror.deliberi_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn stats_fall_back_to_the_did_when_deliberi_fails() { + let Some(db) = test_db().await else { return }; + let [this_month, _] = seed(&db).await; + let mirror = mirror(db, StatusCode::INTERNAL_SERVER_ERROR).await; + let (status, body) = mirror.stats(&[("actor", ACTOR), ("months", "2")]).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!( + body, + json!({"months": [{"start": this_month, "commits": 1}]}) + ); + } + + #[tokio::test] + async fn stats_reject_bad_parameters() { + let Some(db) = test_db().await else { return }; + let mirror = mirror(db, StatusCode::OK).await; + for params in [ + [("actor", ACTOR), ("months", "0")], + [("actor", ACTOR), ("months", "25")], + [("actor", "not-a-did"), ("months", "7")], + ] { + let (status, body) = mirror.stats(¶ms).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{params:?}: {body}"); + } + } + + #[tokio::test] + async fn the_did_document_publishes_the_signing_key() { + let Some(db) = test_db().await else { return }; + let mirror = mirror(db, StatusCode::OK).await; + let body: Value = reqwest::get(format!("{}/.well-known/did.json", mirror.url)) + .await + .unwrap() + .json() + .await + .unwrap(); + assert_eq!(body, signer().did_document()); + assert_eq!(body["id"], MIRROR); + } +} diff --git a/knotmirror/crates/knotmirror-xrpc/src/routes/mod.rs b/knotmirror/crates/knotmirror-xrpc/src/routes/mod.rs index c092ab210..7d03da7fa 100644 --- a/knotmirror/crates/knotmirror-xrpc/src/routes/mod.rs +++ b/knotmirror/crates/knotmirror-xrpc/src/routes/mod.rs @@ -1,3 +1,4 @@ +pub(crate) mod activity; pub(crate) mod cob; pub mod git; pub(crate) mod reads; diff --git a/knotmirror/crates/knotmirror-xrpc/src/signer.rs b/knotmirror/crates/knotmirror-xrpc/src/signer.rs new file mode 100644 index 000000000..9ba3d9c79 --- /dev/null +++ b/knotmirror/crates/knotmirror-xrpc/src/signer.rs @@ -0,0 +1,70 @@ +use std::sync::atomic::{AtomicU64, Ordering}; + +use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; +use jacquard_common::types::did::Did; +use k256::ecdsa::{Signature, SigningKey, signature::Signer as _}; +use serde_json::{Value, json}; + +const SECP256K1_PRIVATE: [u8; 2] = [0x81, 0x26]; +const SECP256K1_PUBLIC: [u8; 2] = [0xe7, 0x01]; +const TOKEN_TTL_SECS: u64 = 60; + +pub struct ServiceSigner { + did: Did, + key: SigningKey, + issued: AtomicU64, +} + +impl ServiceSigner { + /// `encoded` is `z` plus base58 of 0x8126 and the 32 key bytes, the atproto multibase form + pub fn new(did: Did, encoded: &str) -> Result { + let bytes = encoded + .strip_prefix('z') + .and_then(|b58| bs58::decode(b58).into_vec().ok()) + .ok_or("service private key is not multibase base58btc")?; + let raw = bytes + .strip_prefix(&SECP256K1_PRIVATE) + .ok_or("service private key is not a secp256k1 key")?; + let key = SigningKey::from_slice(raw).map_err(|e| format!("service private key: {e}"))?; + Ok(Self { + did, + key, + issued: AtomicU64::new(0), + }) + } + + pub(crate) fn token(&self, audience: &str, lxm: &str, now_secs: u64) -> String { + let jti = format!("{now_secs}-{}", self.issued.fetch_add(1, Ordering::Relaxed)); + let claims = json!({ + "iss": self.did.as_str(), + "aud": audience, + "lxm": lxm, + "iat": now_secs, + "exp": now_secs + TOKEN_TTL_SECS, + "jti": jti, + }); + let input = format!( + "{}.{}", + URL_SAFE_NO_PAD.encode(br#"{"alg":"ES256K","typ":"JWT"}"#), + URL_SAFE_NO_PAD.encode(claims.to_string()), + ); + let signature: Signature = self.key.sign(input.as_bytes()); + format!("{input}.{}", URL_SAFE_NO_PAD.encode(signature.to_bytes())) + } + + pub(crate) fn did_document(&self) -> Value { + let did = self.did.as_str(); + let mut key = SECP256K1_PUBLIC.to_vec(); + key.extend_from_slice(self.key.verifying_key().to_encoded_point(true).as_bytes()); + json!({ + "@context": ["https://www.w3.org/ns/did/v1"], + "id": did, + "verificationMethod": [{ + "id": format!("{did}#atproto"), + "type": "Multikey", + "controller": did, + "publicKeyMultibase": format!("z{}", bs58::encode(key).into_string()), + }], + }) + } +} diff --git a/knotmirror/crates/knotmirror/src/config.rs b/knotmirror/crates/knotmirror/src/config.rs index 6a230eedc..03f3d2288 100644 --- a/knotmirror/crates/knotmirror/src/config.rs +++ b/knotmirror/crates/knotmirror/src/config.rs @@ -17,6 +17,8 @@ pub struct MirrorConfig { pub service_auth: ServiceAuthConfig, #[config(nested)] pub identity: IdentityConfig, + #[config(nested)] + pub deliberi: DeliberiConfig, #[config(nested)] pub redis: RedisConfig, @@ -106,6 +108,22 @@ pub struct ServiceAuthConfig { /// This service's own DID, e.g. `did:web:mirror.tangled.network`. #[config(env = "KNOTMIRROR_SERVICE_DID")] pub did: Did, + + /// multibase secp256k1 private key (`z...`) for outgoing service auth and `/.well-known/did.json`. + /// needed to call deliberi. + #[config(env = "KNOTMIRROR_SERVICE_PRIVATE_KEY")] + pub private_key: Option, +} + +#[derive(Debug, Config)] +pub struct DeliberiConfig { + /// deliberi base url for verified emails. unset counts only commits whose committer email is the did. + #[config(env = "KNOTMIRROR_DELIBERI_URL")] + pub url: Option, + + /// audience of the service auth sent to deliberi. + #[config(env = "KNOTMIRROR_DELIBERI_DID")] + pub did: Option, } #[derive(Debug, Config)] diff --git a/knotmirror/crates/knotmirror/src/main.rs b/knotmirror/crates/knotmirror/src/main.rs index 69c4d4ed9..4a7632da1 100644 --- a/knotmirror/crates/knotmirror/src/main.rs +++ b/knotmirror/crates/knotmirror/src/main.rs @@ -25,7 +25,7 @@ use config::MirrorConfig; use knotmirror_git::layout::Layout; use knotmirror_ingest::{IngestRuntime, run as run_ingest}; use knotmirror_sync::{BackfillWorker, Directory, SyncRuntime}; -use knotmirror_xrpc::{AppState, KnotPolicy, router}; +use knotmirror_xrpc::{AppState, Deliberi, KnotPolicy, ServiceSigner, router}; #[derive(Parser)] #[command(name = "knotmirror", about = "KnotMirror service")] @@ -192,6 +192,21 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { let directory = directory(&http, &cfg.identity.plc_url)?; let service_auth = ServiceAuthConfig::new(cfg.service_auth.did.clone(), directory.clone()) .disable_replay_protection(); + let signer = cfg + .service_auth + .private_key + .as_deref() + .map(|key| ServiceSigner::new(cfg.service_auth.did.clone(), key)) + .transpose() + .map_err(|e| anyhow!(e))?; + let deliberi = match (cfg.deliberi.url, cfg.deliberi.did) { + (Some(url), Some(did)) => Some(Deliberi { url, did }), + (None, None) => None, + _ => return Err(anyhow!("deliberi url and did must be set together")), + }; + if signer.is_none() && deliberi.is_some() { + return Err(anyhow!("calling deliberi needs a service private key")); + } let redis = cfg .redis .url @@ -245,9 +260,15 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { clock: clock.clone(), cancel: cancel.clone(), http: http.clone(), - layout, + layout: layout.clone(), notify: index_notify, }; + let activity_runtime = knotmirror_index::activity::ActivityRuntime { + db: db.clone(), + clock: clock.clone(), + cancel: cancel.clone(), + layout, + }; let maintenance = cfg.maintenance.enabled.then(|| { knotmirror_maintenance::MaintenanceWorker::new( knotmirror_maintenance::MaintenanceRuntime { @@ -291,10 +312,15 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { ("ingest", res.map_err(anyhow::Error::from)) }); tasks.spawn(async move { ("sync", worker.run().await.map_err(anyhow::Error::from)) }); + let activity_poll = index_cfg.poll_interval; tasks.spawn(async move { let res = knotmirror_index::run(index_cfg, index_runtime).await; ("index", res.map_err(anyhow::Error::from)) }); + tasks.spawn(async move { + let res = knotmirror_index::activity::run(activity_poll, activity_runtime).await; + ("activity", res.map_err(anyhow::Error::from)) + }); if let Some(maintenance) = maintenance { tasks.spawn(async move { maintenance.run().await; @@ -304,7 +330,7 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { let max_blob_bytes = cfg.repo.max_blob_bytes; let max_archive_bytes = cfg.repo.max_archive_bytes; - let state = AppState::new( + let mut state = AppState::new( db.clone(), Arc::new(cfg.repo.scan_path), http, @@ -318,6 +344,9 @@ async fn serve(cfg: MirrorConfig) -> anyhow::Result<()> { max_blob_bytes, max_archive_bytes, ); + if let Some(signer) = signer { + state = state.with_signer(signer, deliberi); + } let app = router(state); // TODO: add debug routes diff --git a/nix/Cargo.nix b/nix/Cargo.nix index c45f0de6a..7f22fcea1 100644 --- a/nix/Cargo.nix +++ b/nix/Cargo.nix @@ -21599,6 +21599,11 @@ rec { libName = "knotmirror_db"; license = "MIT"; dependencies = [ + { + name = "chrono"; + packageId = "chrono"; + features = [ "serde" ]; + } { name = "jacquard-common"; packageId = "jacquard-common"; @@ -21624,6 +21629,11 @@ rec { packageId = "tokio"; features = [ "macros" "rt-multi-thread" "time" "signal" "io-util" "net" "sync" "macros" "rt-multi-thread" ]; } + { + name = "url"; + packageId = "url"; + features = [ "serde" ]; + } ]; }; @@ -21675,6 +21685,11 @@ rec { name = "bobbin-runtime"; packageId = "bobbin-runtime"; } + { + name = "gix"; + packageId = "gix"; + features = [ "parallel" "revision" "blob-diff" "blame" "merge" "worktree-archive" "tree-editor" "sha1" "sha256" ]; + } { name = "jacquard-common"; packageId = "jacquard-common"; @@ -21737,6 +21752,17 @@ rec { packageId = "tracing"; } ]; + devDependencies = [ + { + name = "chrono"; + packageId = "chrono"; + features = [ "serde" ]; + } + { + name = "tempfile"; + packageId = "tempfile"; + } + ]; }; "knotmirror-ingest" = rec { @@ -22021,6 +22047,10 @@ rec { packageId = "axum"; features = [ "macros" ]; } + { + name = "base64"; + packageId = "base64 0.22.1"; + } { name = "bobbin-knot-proxy"; packageId = "bobbin-knot-proxy"; @@ -22029,6 +22059,10 @@ rec { name = "bobbin-runtime"; packageId = "bobbin-runtime"; } + { + name = "bs58"; + packageId = "bs58"; + } { name = "chrono"; packageId = "chrono"; @@ -22088,6 +22122,11 @@ rec { packageId = "jacquard-identity"; features = [ "cache" ]; } + { + name = "k256"; + packageId = "k256"; + features = [ "ecdsa" ]; + } { name = "knot-cob"; packageId = "knot-cob"; @@ -22116,6 +22155,10 @@ rec { name = "knotmirror-git"; packageId = "knotmirror-git"; } + { + name = "knotmirror-types"; + packageId = "knotmirror-types"; + } { name = "lexicons"; packageId = "lexicons"; @@ -22133,6 +22176,11 @@ rec { packageId = "metrics-exporter-prometheus"; usesDefaultFeatures = false; } + { + name = "moka"; + packageId = "moka"; + features = [ "sync" "future" ]; + } { name = "quick_cache"; packageId = "quick_cache"; diff --git a/nix/modules/knotmirror.nix b/nix/modules/knotmirror.nix index 302817389..ed8b7c915 100644 --- a/nix/modules/knotmirror.nix +++ b/nix/modules/knotmirror.nix @@ -63,6 +63,24 @@ in description = "atproto PLC directory"; }; + deliberiUrl = mkOption { + type = types.str; + default = ""; + description = "base URL of deliberi, asked for an actor's verified emails when counting their commits"; + }; + + deliberiDid = mkOption { + type = types.str; + default = ""; + description = "deliberi's service DID, the audience of the service auth sent to it"; + }; + + environmentFile = mkOption { + type = types.nullOr types.path; + default = null; + description = "file with secret env vars such as KNOTMIRROR_SERVICE_PRIVATE_KEY"; + }; + legacy = { enable = mkOption { type = types.bool; @@ -170,6 +188,17 @@ in }; config = mkIf cfg.enable (mkMerge [ { + assertions = [ + { + assertion = (cfg.deliberiUrl == "") == (cfg.deliberiDid == ""); + message = "services.tangled.knotmirror deliberiUrl and deliberiDid must be set together"; + } + { + assertion = cfg.deliberiUrl == "" || cfg.environmentFile != null; + message = "services.tangled.knotmirror environmentFile is required when deliberi is configured"; + } + ]; + environment.systemPackages = [ pkgs.git cfg.package @@ -182,16 +211,22 @@ in serviceConfig = { LogsDirectory = "knotmirror"; StateDirectory = "knotmirror"; - Environment = [ - "KNOTMIRROR_BIND=${cfg.listenAddr}" - "KNOTMIRROR_DATABASE_URL=${cfg.dbUrl}" - "KNOTMIRROR_HYDRANT_URL=${cfg.hydrantUrl}" - "KNOTMIRROR_SCAN_PATH=/var/lib/knotmirror/repos" - "KNOTMIRROR_PLC_URL=${cfg.atpPlcUrl}" - "KNOTMIRROR_SERVICE_DID=did:web:${cfg.hostname}" - "KNOTMIRROR_LOG_ENDPOINT=${cfg.logging.endpoint}" - "KNOTMIRROR_LOG_SERVICE_NAME=${cfg.logging.serviceName}" - ]; + EnvironmentFile = mkIf (cfg.environmentFile != null) cfg.environmentFile; + Environment = + [ + "KNOTMIRROR_BIND=${cfg.listenAddr}" + "KNOTMIRROR_DATABASE_URL=${cfg.dbUrl}" + "KNOTMIRROR_HYDRANT_URL=${cfg.hydrantUrl}" + "KNOTMIRROR_SCAN_PATH=/var/lib/knotmirror/repos" + "KNOTMIRROR_PLC_URL=${cfg.atpPlcUrl}" + "KNOTMIRROR_SERVICE_DID=did:web:${cfg.hostname}" + "KNOTMIRROR_LOG_ENDPOINT=${cfg.logging.endpoint}" + "KNOTMIRROR_LOG_SERVICE_NAME=${cfg.logging.serviceName}" + ] + ++ optionals (cfg.deliberiUrl != "") [ + "KNOTMIRROR_DELIBERI_URL=${cfg.deliberiUrl}" + "KNOTMIRROR_DELIBERI_DID=${cfg.deliberiDid}" + ]; ExecStart = "${getExe cfg.package} serve"; Restart = "always"; };