Something went wrong. Try again.
unified development activity
Something went wrong. Try again.
13 kB · 421 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422//! tally — the commit activity behind dev.permadeath.com.
mod burn;mod commit_types;mod config;mod crawl;mod day_shards;mod db;mod described;mod didbot;mod discovery;mod github;mod global_summary;mod http;mod land_gap;mod model;mod og;mod project_sites;mod project_timelines;mod reach;mod rules;mod serve;mod shard;mod site;mod stats;mod tangled;mod work_inbox;
#[cfg(feature = "aws")]mod publish;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};use clap::{Args, Parser, Subcommand};
use crate::config::Config;use crate::crawl::CrawlOptions;use crate::db::Db;use crate::model::IdentityKind;
#[derive(Debug, Parser)]#[command( name = "tally", version, about = "Deduplicated commit activity for dev.permadeath.com")]struct Cli { #[command(subcommand)] command: Command,}
#[derive(Debug, Subcommand)]enum Command { /// Crawl Tangled, didbot and GitHub, then write the static site. Crawl(CrawlArgs), /// Serve an already-built site for local inspection. Serve(ServeArgs), /// Crawl on a timer and publish to S3 when configured. Daemon(DaemonArgs), /// Merge the shards' counted changes into the global summary. GlobalMerge(MergeArgs), /// Find every public Tangled repository once and write the list the /// shards of a global run read (TALLY_GLOBAL_DISCOVERY). GlobalDiscover, /// Print the didbot identities currently known to the database. Bots, /// Upload an existing build to S3. #[cfg(feature = "aws")] Publish(PublishArgs),}
#[derive(Debug, Clone, Args)]struct CrawlArgs { #[arg(long)] db: Option<PathBuf>, #[arg(long)] out: Option<PathBuf>, /// Nightly ecosystem pass: every public Tangled repository, no owned-first /// queue, no GitHub, no inbox. Writes the global summary and does not /// publish the site. #[arg(long, conflicts_with = "owned_only")] global: bool, /// Only scan repositories owned by the personal or bot identities. #[arg(long)] owned_only: bool, #[arg(long)] skip_global: bool, #[arg(long)] skip_github: bool, /// Ignore stored branch heads and re-read history. #[arg(long)] full: bool, /// Apply the global filters to the existing database and write a summary /// without walking commits. Requires --global. #[arg(long, requires = "global")] filter_only: bool, #[arg(long)] max_repos: Option<usize>, /// Crawl one shard of the global pass, as index/count (for example 2/8). /// Overrides TALLY_SHARD_INDEX and TALLY_SHARD_COUNT. #[arg(long, value_parser = shard::Shard::parse, requires = "global")] shard: Option<shard::Shard>, /// Start a shard with no database of its own from this single-task /// database, keeping only the shard's repositories. With publishing on, /// the single-task state object is the seed. #[arg(long, requires = "shard")] seed: Option<PathBuf>,}
impl CrawlArgs { fn apply(&self, config: &mut Config) { if let Some(db) = &self.db { config.db_path = db.clone(); } if let Some(out) = &self.out { config.out_dir = out.clone(); } if self.max_repos.is_some() { config.max_repos_per_run = self.max_repos; } if let Some(shard) = self.shard { config.set_shard(shard); } }
fn options(&self) -> CrawlOptions { CrawlOptions { global: self.global, owned_only: self.owned_only, skip_global: self.skip_global, skip_github: self.skip_github, full: self.full, filter_only: self.filter_only, max_repos: self.max_repos, } }}
#[derive(Debug, Args)]struct MergeArgs { /// Counted-change files to merge. Without them, each shard's file is /// downloaded from the state bucket. #[arg(long)] counts: Vec<PathBuf>, /// How many shards there are (default TALLY_SHARD_COUNT, or one per /// --counts file). #[arg(long)] shards: Option<usize>,}
#[derive(Debug, Args)]struct ServeArgs { #[arg(long, default_value = "dist")] dir: PathBuf, #[arg(long, default_value = "127.0.0.1:8787")] bind: String,}
#[derive(Debug, Args)]struct DaemonArgs { #[command(flatten)] crawl: CrawlArgs, /// Seconds between runs. #[arg(long, default_value_t = 3_600)] interval: u64, /// Run once and exit instead of sleeping. #[arg(long)] once: bool, /// Optional address for the health endpoint (for example 0.0.0.0:8080). #[arg(long, env = "TALLY_HEALTH_BIND")] health_bind: Option<String>,}
#[cfg(feature = "aws")]#[derive(Debug, Args)]struct PublishArgs { #[arg(long, default_value = "dist")] dir: PathBuf, #[arg(long)] db: Option<PathBuf>,}
#[tokio::main]async fn main() -> Result<()> { tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), ) .with_target(false) .init();
match Cli::parse().command { Command::Crawl(args) => { let mut config = Config::from_env()?; args.apply(&mut config); crawl_once(&config, &args.options(), args.seed.as_deref()).await?; } Command::Serve(args) => serve::run(args.dir, &args.bind).await?, Command::Daemon(args) => { let mut config = Config::from_env()?; args.crawl.apply(&mut config); daemon(&config, &args).await?; } Command::GlobalMerge(args) => { let config = Config::from_env()?; global_merge(&config, &args).await?; } Command::GlobalDiscover => discovery::run(&Config::from_env()?).await?, Command::Bots => print_bots().await?, #[cfg(feature = "aws")] Command::Publish(args) => { let mut config = Config::from_env()?; if let Some(db) = &args.db { config.db_path = db.clone(); } publish::publish(&config, &args.dir).await?; } } Ok(())}
async fn crawl_once(config: &Config, options: &CrawlOptions, seed: Option<&Path>) -> Result<()> { // A seed downloaded for this run is removed once used; a --seed file is // the caller's. #[allow(unused_mut)] let (mut seed, mut downloaded) = (seed.map(Path::to_path_buf), false); #[cfg(feature = "aws")] if config.publish && let Some(restored) = publish::restore_state_if_missing(config).await? { (seed, downloaded) = (Some(restored), true); }
let db = match seed { Some(seed) if config.shard.is_sharded() && !config.db_path.exists() => { let started = std::time::Instant::now(); let db = Db::seed_shard(&config.db_path, &seed, config.shard).with_context(|| { format!( "seeding {} from {}", config.db_path.display(), seed.display() ) })?; tracing::info!( shard = %config.shard, elapsed_secs = started.elapsed().as_secs(), "seeded the shard's database from the single-task one" ); if downloaded { let _ = std::fs::remove_file(&seed); } db } _ => Db::open(&config.db_path) .with_context(|| format!("opening {}", config.db_path.display()))?, }; db.set_branch_scope(config.branch_scope)?; let report = crawl::run(config, &db, options).await?;
if options.global { #[cfg(not(feature = "aws"))] if config.publish { tracing::warn!( "TALLY_PUBLISH is set but this binary was built without the aws feature" ); } crawl::publish_global(config, &db, options.filter_only).await?; tracing::info!( shard = %config.shard, repos_total = report.repos_total, repos_queued = report.repos_queued, commits_seen = report.scan.commits_seen, commits_new = report.scan.commits_new, "global crawl complete" ); return Ok(()); }
let global = match global_summary::read(config).await { Ok(summary) => summary, Err(err) => { tracing::warn!(error = %err, "could not read the global summary; hiding the share"); None } }; // Token burn reads vibescrobble records, so it follows the activity card. if config.activity_url.is_some() { let since = stats::window_start(chrono::Utc::now()).and_time(chrono::NaiveTime::MIN); if let Err(err) = burn::refresh(config, &db, since.and_utc()).await { tracing::warn!(error = %err, "could not read scrobble usage"); } } let data = stats::build(config, &db, &report, global.as_ref())?; site::write(&config.out_dir, &data, config.og_layout.as_deref())?;
tracing::info!( repos_total = report.repos_total, repos_owned = report.repos_owned, queued = report.repos_queued, commits_seen = report.scan.commits_seen, commits_new = report.scan.commits_new, tangled_total = data.summary.tangled_total, me_tangled = data.summary.me_tangled, bots_tangled = data.summary.bots_tangled, share = format!("{:.2}%", data.summary.share_me), out = %config.out_dir.display(), "crawl complete" );
#[cfg(feature = "aws")] if config.publish { // The upload is the database file: fold the log in first, or the // restored copy would lack the run's last writes. db.flush_wal()?; publish::publish(config, &config.out_dir).await?; } #[cfg(not(feature = "aws"))] if config.publish { tracing::warn!("TALLY_PUBLISH is set but this binary was built without the aws feature"); }
Ok(())}
/// Download every shard's counted changes from the state bucket.#[cfg(feature = "aws")]async fn download_counts(config: &Config, shards: usize) -> Result<Vec<PathBuf>> { let mut paths = Vec::with_capacity(shards); for index in 0..shards { let key = shard::Shard::new(index, shards)?.counts_key(&config.base_state_key); let path = config .db_path .with_extension(format!("shard-{index}.counts.sqlite3")); publish::download_file(config, &key, &path).await?; paths.push(path); } Ok(paths)}
#[cfg(not(feature = "aws"))]async fn download_counts(_config: &Config, _shards: usize) -> Result<Vec<PathBuf>> { anyhow::bail!("pass --counts files; this build cannot read them from S3")}
/// Build the global summary from every shard's counted changes and write it.async fn global_merge(config: &Config, args: &MergeArgs) -> Result<()> { let shards = args.shards.unwrap_or(if args.counts.is_empty() { config.shard.count } else { args.counts.len() }); let paths = if args.counts.is_empty() { download_counts(config, shards).await? } else { args.counts.clone() }; let rules = rules::RuleSet::from_config(config)?; let summary = global_summary::from_counts(&paths, &rules, shards)?; summary.check_publishable( config.global_min_coverage_pct, config.global_min_default_ratio, )?; global_summary::write(config, &summary).await?; tracing::info!( shards, partial = summary.partial, generated_at = %summary.generated_at, days = summary.days.len(), changes = summary.total(None), rules = %rules.hash(), held = summary.held_for_review.as_ref().map_or(0, Vec::len), "merged the global shards" ); Ok(())}
async fn daemon(config: &Config, args: &DaemonArgs) -> Result<()> { if let Some(bind) = &args.health_bind { let dir = config.out_dir.clone(); let bind = bind.clone(); tokio::spawn(async move { if let Err(err) = serve::run(dir, &bind).await { tracing::error!(error = %err, "health server stopped"); } }); }
loop { match crawl_once(config, &args.crawl.options(), args.crawl.seed.as_deref()).await { Ok(()) => {} Err(err) => tracing::error!(error = %err, "crawl failed; keeping the previous site"), } if args.once { break; } tracing::info!(interval = args.interval, "sleeping until the next crawl"); tokio::time::sleep(std::time::Duration::from_secs(args.interval)).await; } Ok(())}
async fn print_bots() -> Result<()> { let config = Config::from_env()?; let db = Db::open(&config.db_path)?; let identities = db.list_identities()?; let bots: Vec<_> = identities .iter() .filter(|identity| identity.kind == IdentityKind::Bot) .collect(); println!( "{} bot identities ({} active)", bots.len(), bots.iter().filter(|identity| identity.active).count() ); for bot in bots.iter().filter(|identity| identity.active) { println!("{} {}", bot.did, bot.handle); } Ok(())}