Skip to content
Sign in

flagon-io/g1t

public
0

g1t/services/repos/src/stats.rs

418 lines17,793 bytesCodeBlame
1//! What a repository's About reads from its files and history: its
2//! license, its security policy, its languages and its contributors.
3//!
4//! Worked out for the default branch's head in the background, never on
5//! the way to a page: a request that finds the kept answer behind the head
6//! (or none at all) answers with what is kept and starts the work in its
7//! `wait_until` (`lib.rs`, `refresh`). One run at a time per repository,
8//! by a lease in `repo_stats.started_ms`. What it reads is kept by hash in
9//! the store's cache (trees, blob sizes, history), so working out the next
10//! commit reads only what changed. A run that runs out of time keeps what
11//! it read, marked `partial`, and the next view starts another.
12
13use std::collections::{HashMap, HashSet};
14
15use g1t_contracts::about::{ABOUT_CONTRIBUTORS, Contributor, Contributors, Freshness, LanguageShare, Languages, License, WeekCommits};
16use g1t_contracts::accounts::{EmailOwner, EmailOwnersArgs};
17use g1t_contracts::repos::{FileEntry, MAX_LISTED_FILES, Repo};
18use g1t_contracts::time::rfc3339;
19use g1t_kit::now_ms;
20use serde::{Deserialize, Serialize};
21use worker::wasm_bindgen::JsValue;
22use worker::{D1Database, Fetcher, Result};
23
24use crate::contributors::{Authored, tally};
25use crate::languages::{self, Attributes};
26use crate::license;
27use crate::store::GitRepo;
28
29/// A run older than this is taken to have died.
30pub const LEASE_MS: u64 = 120_000;
31/// How long a run reads before keeping what it has.
32const BUDGET_MS: u64 = 22_000;
33/// The most commits of history read for contributors.
34pub const MAX_HISTORY: usize = 3_000;
35/// Commits read at a time.
36const LOG_PAGE: u32 = 200;
37/// The most blobs sized for languages.
38const MAX_SIZED: usize = 6_000;
39/// Blobs sized at once: each is a round trip to the store.
40const SIZE_BATCH: usize = 32;
41/// The most distinct addresses matched to accounts.
42const MAX_MATCHED: usize = 1_000;
43/// Directories never read: nothing in them ever counts.
44const NEVER_READ: [&str; 3] = ["node_modules", "bower_components", "jspm_packages"];
45/// Where a security policy may be.
46const POLICY_DIRS: [&str; 4] = ["", ".g1t/", ".github/", "docs/"];
47
48/// What `repo_stats` keeps for the About, without the full contributors.
49#[derive(Debug, Default, Deserialize)]
50pub struct Kept {
51 pub commit_hash: Option<String>,
52 pub computed_at: Option<String>,
53 pub started_ms: Option<f64>,
54 pub partial: f64,
55 pub license: Option<String>,
56 pub security_policy: Option<String>,
57 pub languages: Option<String>,
58 pub contributors_total: f64,
59 pub contributors_top: Option<String>,
60}
61
62impl Kept {
63 /// Whether a run is under way now.
64 pub fn running(&self, now: u64) -> bool {
65 self.started_ms.is_some_and(|started| now.saturating_sub(started as u64) < LEASE_MS)
66 }
67
68 /// Whether a new run should start: the head moved past what is kept,
69 /// or what is kept is partial, and none is under way.
70 pub fn wants_run(&self, head: Option<&str>, now: u64) -> bool {
71 let Some(head) = head else { return false };
72 let behind = self.commit_hash.as_deref() != Some(head) || self.partial > 0.0;
73 // A partial answer is tried again at most every lease.
74 let rested = self.partial == 0.0 || self.started_ms.is_none_or(|started| now.saturating_sub(started as u64) >= LEASE_MS);
75 behind && rested && !self.running(now)
76 }
77
78 pub fn freshness(&self, head: Option<String>) -> Freshness {
79 Freshness {
80 pending: head.is_some() && self.commit_hash.is_none(),
81 head,
82 commit: self.commit_hash.clone(),
83 computed_at: self.computed_at.clone(),
84 partial: self.partial > 0.0,
85 }
86 }
87
88 pub fn license(&self) -> Option<License> {
89 self.license.as_deref().and_then(|text| serde_json::from_str(text).ok())
90 }
91
92 pub fn languages(&self) -> Vec<LanguageShare> {
93 self.languages.as_deref().and_then(|text| serde_json::from_str(text).ok()).unwrap_or_default()
94 }
95
96 pub fn top_contributors(&self) -> Vec<Contributor> {
97 self.contributors_top.as_deref().and_then(|text| serde_json::from_str(text).ok()).unwrap_or_default()
98 }
99}
100
101/// What `repo_stats` keeps for a repository; the default when nothing is.
102pub async fn kept(db: &D1Database, repo_id: &str) -> Result<Kept> {
103 Ok(db
104 .prepare(
105 "SELECT commit_hash, computed_at, started_ms, partial, license, security_policy, languages, contributors_total, contributors_top
106 FROM repo_stats WHERE repo_id = ?",
107 )
108 .bind(&[repo_id.into()])?
109 .first::<Kept>(None)
110 .await?
111 .unwrap_or_default())
112}
113
114/// The whole contributors answer kept for a repository.
115#[derive(Debug, Default, Serialize, Deserialize)]
116struct KeptContributors {
117 commits: u32,
118 contributors: Vec<Contributor>,
119 weeks: Vec<WeekCommits>,
120}
121
122/// The Contributors page's answer, as kept.
123pub async fn contributors(db: &D1Database, repo_id: &str, head: Option<String>) -> Result<Contributors> {
124 #[derive(Deserialize)]
125 struct Row {
126 contributors: Option<String>,
127 }
128 let summary = kept(db, repo_id).await?;
129 let full: KeptContributors = db
130 .prepare("SELECT contributors FROM repo_stats WHERE repo_id = ?")
131 .bind(&[repo_id.into()])?
132 .first::<Row>(None)
133 .await?
134 .and_then(|row| row.contributors)
135 .and_then(|text| serde_json::from_str(&text).ok())
136 .unwrap_or_default();
137 Ok(Contributors {
138 freshness: summary.freshness(head),
139 total: summary.contributors_total as u32,
140 commits: full.commits,
141 contributors: full.contributors,
142 weeks: full.weeks,
143 })
144}
145
146/// The languages answer, as kept.
147pub fn languages_of(kept: &Kept, head: Option<String>) -> Languages {
148 Languages { freshness: kept.freshness(head), languages: kept.languages() }
149}
150
151/// Takes the lease for a run, unless one is under way. Whether it was taken.
152pub async fn claim(db: &D1Database, repo_id: &str, now: u64) -> Result<bool> {
153 let taken = db
154 .prepare(
155 "INSERT INTO repo_stats (repo_id, started_ms) VALUES (?1, ?2)
156 ON CONFLICT (repo_id) DO UPDATE SET started_ms = excluded.started_ms
157 WHERE repo_stats.started_ms IS NULL OR repo_stats.started_ms < ?3
158 RETURNING repo_id",
159 )
160 .bind(&[repo_id.into(), JsValue::from_f64(now as f64), JsValue::from_f64(now.saturating_sub(LEASE_MS) as f64)])?
161 .first::<serde_json::Value>(None)
162 .await?;
163 Ok(taken.is_some())
164}
165
166/// Everything one run worked out.
167#[derive(Debug, Default)]
168pub struct Worked {
169 pub commit: String,
170 pub partial: bool,
171 pub license: Option<License>,
172 pub security_policy: Option<String>,
173 pub languages: Vec<LanguageShare>,
174 pub contributors: Vec<Contributor>,
175 pub contributors_total: u32,
176 pub commits: u32,
177 pub weeks: Vec<WeekCommits>,
178}
179
180/// Keeps a run's answer and lets the lease go. A partial answer keeps its
181/// lease's start, so the next is not tried until the lease is over.
182pub async fn keep(db: &D1Database, repo_id: &str, worked: &Worked) -> Result<()> {
183 let top: Vec<Contributor> = worked
184 .contributors
185 .iter()
186 .take(ABOUT_CONTRIBUTORS)
187 .map(|contributor| Contributor { weeks: Vec::new(), ..contributor.clone() })
188 .collect();
189 let full = KeptContributors { commits: worked.commits, contributors: worked.contributors.clone(), weeks: worked.weeks.clone() };
190 db.prepare(
191 "UPDATE repo_stats SET commit_hash = ?2, computed_at = ?3, started_ms = CASE WHEN ?4 = 1 THEN started_ms ELSE NULL END,
192 partial = ?4, license = ?5, security_policy = ?6, languages = ?7, contributors_total = ?8, contributors_top = ?9,
193 contributors = ?10
194 WHERE repo_id = ?1",
195 )
196 .bind(&[
197 repo_id.into(),
198 worked.commit.as_str().into(),
199 rfc3339(now_ms()).into(),
200 JsValue::from_f64(if worked.partial { 1.0 } else { 0.0 }),
201 match &worked.license {
202 Some(license) => json(license).into(),
203 None => JsValue::NULL,
204 },
205 match &worked.security_policy {
206 Some(path) => path.as_str().into(),
207 None => JsValue::NULL,
208 },
209 json(&worked.languages).into(),
210 JsValue::from_f64(f64::from(worked.contributors_total)),
211 json(&top).into(),
212 json(&full).into(),
213 ])?
214 .run()
215 .await?;
216 Ok(())
217}
218
219/// Lets the lease go after a run that failed, so the next view tries again
220/// once the lease would have ended anyway.
221pub async fn release(db: &D1Database, repo_id: &str) {
222 let statement = db
223 .prepare("UPDATE repo_stats SET started_ms = ?2 WHERE repo_id = ?1")
224 .bind(&[repo_id.into(), JsValue::from_f64(now_ms().saturating_sub(LEASE_MS / 2) as f64)]);
225 if let Ok(statement) = statement
226 && let Err(error) = statement.run().await
227 {
228 worker::console_error!("the About's lease of {repo_id} was not let go: {error}");
229 }
230}
231
232fn json<T: Serialize>(value: &T) -> String {
233 serde_json::to_string(value).unwrap_or_else(|_| "null".to_owned())
234}
235
236/// The security policy among a repository's files, at the root or in
237/// `.g1t`, `.github` or `docs`.
238pub fn security_policy(files: &[FileEntry]) -> Option<String> {
239 POLICY_DIRS.iter().find_map(|dir| {
240 files
241 .iter()
242 .find(|file| {
243 file.path
244 .strip_prefix(dir)
245 .is_some_and(|name| !name.contains('/') && (name.eq_ignore_ascii_case("security.md") || name.eq_ignore_ascii_case("security.txt") && dir.is_empty()))
246 })
247 .map(|file| file.path.clone())
248 })
249}
250
251/// Works out the About of `repo` at `head` (a commit). `identity` matches
252/// authors' addresses to accounts.
253pub async fn work_out<R: GitRepo>(git: &R, head: &str, tree: &str, identity: Option<&Fetcher>) -> Result<Worked> {
254 let started = now_ms();
255 let out_of_time = move || now_ms().saturating_sub(started) > BUDGET_MS;
256 let never: Vec<String> = NEVER_READ.iter().map(|dir| (*dir).to_owned()).collect();
257 let files = async {
258 let (files, truncated) = crate::listing::changed(git, None, tree, &never, MAX_LISTED_FILES).await?;
259 let root: Vec<&FileEntry> = files.iter().filter(|file| !file.path.contains('/')).collect();
260 // The license: the best-named license file that reads as text.
261 let mut license = None;
262 for name in license::candidates(root.iter().map(|file| file.path.as_str())) {
263 let Some(hash) = root.iter().find(|file| file.path == name).and_then(|file| file.hash.clone()) else { continue };
264 if let Some(text) = git.read_blob(&hash).await?.and_then(|bytes| String::from_utf8(bytes).ok()) {
265 license = Some(license::detect(name, &text));
266 break;
267 }
268 }
269 let attributes = match root.iter().find(|file| file.path == ".gitattributes").and_then(|file| file.hash.clone()) {
270 Some(hash) => git
271 .read_blob(&hash)
272 .await?
273 .map(|bytes| Attributes::parse(&String::from_utf8_lossy(&bytes)))
274 .unwrap_or_default(),
275 None => Attributes::default(),
276 };
277 let policy = security_policy(&files);
278 // Each counted file's language, its blob sized once.
279 let mut counted: Vec<(&'static languages::Language, String)> = Vec::new();
280 for file in &files {
281 let Some(hash) = &file.hash else { continue };
282 if let Some(language) = languages::counted(&file.path, &attributes) {
283 counted.push((language, hash.clone()));
284 }
285 }
286 let mut partial = truncated || counted.len() > MAX_SIZED;
287 counted.truncate(MAX_SIZED);
288 let distinct: Vec<String> = counted.iter().map(|(_, hash)| hash.clone()).collect::<HashSet<_>>().into_iter().collect();
289 let mut sizes: HashMap<String, u64> = HashMap::with_capacity(distinct.len());
290 for batch in distinct.chunks(SIZE_BATCH) {
291 if out_of_time() {
292 partial = true;
293 break;
294 }
295 let found = futures_util::future::join_all(batch.iter().map(|hash| git.blob_size(hash))).await;
296 for (hash, size) in batch.iter().zip(found) {
297 if let Ok(Some(size)) = size {
298 sizes.insert(hash.clone(), size);
299 }
300 }
301 }
302 let shares = languages::shares(counted.iter().filter_map(|(language, hash)| Some((*language, *sizes.get(hash)?))));
303 Ok::<_, worker::Error>((license, policy, shares, partial))
304 };
305 let history = async {
306 let mut commits: Vec<Authored> = Vec::new();
307 let mut next = Some(head.to_owned());
308 let mut partial = false;
309 while let Some(from) = next.take() {
310 let page = git.log(&from, LOG_PAGE).await?;
311 let more = page.len() == LOG_PAGE as usize;
312 let after = page.last().and_then(|commit| commit.parents.first().cloned());
313 for commit in page {
314 commits.push(Authored { name: commit.author.name, email: commit.author.email, at: commit.authored_at });
315 }
316 if more && after.is_some() {
317 if commits.len() >= MAX_HISTORY || out_of_time() {
318 partial = true;
319 } else {
320 next = after;
321 }
322 }
323 }
324 commits.truncate(MAX_HISTORY);
325 Ok::<_, worker::Error>((commits, partial))
326 };
327 let (files, history) = futures_util::future::join(files, history).await;
328 let (license, security_policy, languages, files_partial) = files?;
329 let (commits, history_partial) = history?;
330 let owners = match identity {
331 Some(identity) => owners(identity, &commits).await,
332 None => HashMap::new(),
333 };
334 let (contributors, contributors_total, weeks) = tally(&commits, &owners);
335 Ok(Worked {
336 commit: head.to_owned(),
337 partial: files_partial || history_partial,
338 license,
339 security_policy,
340 languages,
341 contributors,
342 contributors_total,
343 commits: commits.len() as u32,
344 weeks,
345 })
346}
347
348/// The accounts behind the commits' addresses, the most frequent first. A
349/// failure matches none: the authors show by name.
350async fn owners(identity: &Fetcher, commits: &[Authored]) -> HashMap<String, EmailOwner> {
351 let mut counts: HashMap<String, usize> = HashMap::new();
352 for commit in commits.iter().filter(|commit| !crate::contributors::is_g1t(commit)) {
353 let email = commit.email.trim().to_ascii_lowercase();
354 if !email.is_empty() {
355 *counts.entry(email).or_default() += 1;
356 }
357 }
358 let mut emails: Vec<(String, usize)> = counts.into_iter().collect();
359 emails.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
360 emails.truncate(MAX_MATCHED);
361 let mut owners = HashMap::new();
362 for chunk in emails.chunks(200) {
363 let args = EmailOwnersArgs { emails: chunk.iter().map(|(email, _)| email.clone()).collect() };
364 match g1t_kit::call::<_, HashMap<String, EmailOwner>>(identity, "email_owners", &args).await {
365 Ok(found) => owners.extend(found),
366 Err(error) => worker::console_error!("contributors' accounts not matched: {error}"),
367 }
368 }
369 owners
370}
371
372/// Whether the repository is one whose About is worked out: not a pull
373/// request's working copy.
374pub fn has_about(repo: &Repo) -> bool {
375 repo.fork_of.is_none()
376}
377
378#[cfg(test)]
379mod tests {
380 use super::*;
381
382 fn file(path: &str) -> FileEntry {
383 FileEntry { path: path.into(), hash: Some("h".into()) }
384 }
385
386 #[test]
387 fn a_security_policy_is_found_where_it_may_be() {
388 assert_eq!(security_policy(&[file("src/SECURITY.md"), file(".github/SECURITY.md")]).as_deref(), Some(".github/SECURITY.md"));
389 assert_eq!(security_policy(&[file("docs/security.md"), file("SECURITY.md")]).as_deref(), Some("SECURITY.md"));
390 assert_eq!(security_policy(&[file(".g1t/SECURITY.md")]).as_deref(), Some(".g1t/SECURITY.md"));
391 assert_eq!(security_policy(&[file("SECURITY.txt")]).as_deref(), Some("SECURITY.txt"));
392 assert_eq!(security_policy(&[file("docs/guides/SECURITY.md"), file("README.md")]), None);
393 }
394
395 #[test]
396 fn a_run_starts_when_the_head_moved_and_none_is_under_way() {
397 let now = 1_000_000;
398 let none = Kept::default();
399 assert!(none.wants_run(Some("c2"), now));
400 assert!(!none.wants_run(None, now), "an empty repository has nothing to work out");
401 let current = Kept { commit_hash: Some("c2".into()), ..Kept::default() };
402 assert!(!current.wants_run(Some("c2"), now));
403 assert!(current.wants_run(Some("c3"), now));
404 let running = Kept { commit_hash: Some("c2".into()), started_ms: Some((now - 1_000) as f64), ..Kept::default() };
405 assert!(!running.wants_run(Some("c3"), now));
406 let died = Kept { commit_hash: Some("c2".into()), started_ms: Some((now - LEASE_MS - 1) as f64), ..Kept::default() };
407 assert!(died.wants_run(Some("c3"), now));
408 let partial = Kept { commit_hash: Some("c2".into()), partial: 1.0, started_ms: Some((now - 5_000) as f64), ..Kept::default() };
409 assert!(!partial.wants_run(Some("c2"), now), "a partial answer rests a lease before the next try");
410 }
411
412 #[test]
413 fn nothing_kept_is_pending() {
414 let freshness = Kept::default().freshness(Some("c1".into()));
415 assert!(freshness.pending);
416 assert!(!Kept::default().freshness(None).pending, "an empty repository is not waiting on anything");
417 }
418}