//! tally — the commit activity behind dev.permadeath.com.

mod burn;
mod commit_types;
mod config;
mod crawl;
mod day_shards;
mod db;
mod declared;
mod described;
mod didbot;
mod discovery;
mod github;
mod global_summary;
mod http;
mod land_gap;
mod llms;
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(())
}