kasl-server 0.23.0

use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use kasl_server::{app, config, demo, import, provision};
use sqlx::postgres::PgPoolOptions;
use tokio::net::TcpListener;
use tracing_subscriber::EnvFilter;

/// Team server for kasl.
///
/// Runs the server when given no subcommand, which is what a container's
/// entrypoint does and what every existing deployment expects.
#[derive(Debug, Parser)]
#[command(name = "kasl-server", version, about)]
struct Cli {
    #[command(subcommand)]
    command: Option<Command>,
}

#[derive(Debug, Subcommand)]
enum Command {
    /// Import an employee's local kasl history from their SQLite database.
    Import(ImportArgs),
    /// Create the first administrator, or reset an existing one's password.
    Admin(AdminArgs),
    /// Write the installation's data to a file, or to stdout.
    Backup(BackupArgs),
    /// Load a backup into an empty installation.
    Restore(RestoreArgs),
}

#[derive(Debug, clap::Args)]
struct BackupArgs {
    /// Where to write. Omit for stdout, which is what a cron job piping into
    /// gzip or a remote copy wants.
    #[arg(long, value_name = "PATH")]
    out: Option<std::path::PathBuf>,
}

#[derive(Debug, clap::Args)]
struct RestoreArgs {
    /// The backup to read. Omit for stdin.
    #[arg(long, value_name = "PATH")]
    from: Option<std::path::PathBuf>,
}

#[derive(Debug, clap::Args)]
struct AdminArgs {
    /// Email the administrator signs in with.
    #[arg(long, value_name = "EMAIL")]
    email: String,
    /// Their password. At least 8 characters.
    #[arg(long, value_name = "PASSWORD")]
    password: String,
}

#[derive(Debug, clap::Args)]
struct ImportArgs {
    /// Path to the agent's database file (kasl's own `kasl.db`).
    #[arg(long, value_name = "PATH")]
    db: std::path::PathBuf,
    /// Email of the user to import into. The account must already exist.
    #[arg(long, value_name = "EMAIL")]
    user: String,
    /// UTC offset the history was recorded in, e.g. `-03:00`.
    ///
    /// Required, with no default: kasl stores bare wall-clock time, so nothing
    /// in the file says which offset it was. A wrong guess here silently shifts
    /// a year of someone's hours, so the answer comes from the person who knows
    /// (ADR 0006).
    ///
    /// `allow_hyphen_values` because the common case starts with one: without
    /// it `--timezone -03:00` is read as an unknown flag `-0`, and the operator
    /// is refused for typing exactly what the documentation shows.
    #[arg(long, value_name = "OFFSET", value_parser = parse_offset, allow_hyphen_values = true)]
    timezone: chrono::FixedOffset,
    /// Import only days on or after this date (`YYYY-MM-DD`).
    ///
    /// With `--until`, this is how an employee who moved between time zones is
    /// imported correctly: one run per stretch, each with its own offset.
    #[arg(long, value_name = "DATE")]
    since: Option<chrono::NaiveDate>,
    /// Import only days on or before this date (`YYYY-MM-DD`).
    #[arg(long, value_name = "DATE")]
    until: Option<chrono::NaiveDate>,
    /// Read and report what would be imported, without writing anything.
    #[arg(long)]
    dry_run: bool,
}

/// Parses `-03:00`, `+05:30`, or `Z`.
fn parse_offset(raw: &str) -> Result<chrono::FixedOffset, String> {
    // Parsed by borrowing a full timestamp's machinery: the offset alone has no
    // parser in chrono, and hand-rolling one invites the classic sign mistake.
    chrono::DateTime::parse_from_rfc3339(&format!("2000-01-01T00:00:00{}", if raw == "Z" { "Z".to_string() } else { raw.to_string() }))
        .map(|time| *time.offset())
        .map_err(|_| format!("not a UTC offset: {raw} (expected something like -03:00, +05:30 or Z)"))
}

#[tokio::main]
async fn main() -> Result<()> {
    let cli = Cli::parse();

    // Logs go to stderr, always. `kasl-server backup` writes the backup itself
    // to stdout, and the startup line about migrations landing in the middle of
    // it produced a file that looked fine and could not be parsed - found by
    // running the documented command against the published image, not by a
    // test. Keeping the two streams apart is also what every other CLI does.
    tracing_subscriber::fmt()
        .with_writer(std::io::stderr)
        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("kasl_server=info,tower_http=info")))
        .init();

    let config = config::Config::from_env()?;

    let pool = PgPoolOptions::new()
        .max_connections(10)
        .connect(&config.database_url)
        .await
        .context("failed to connect to PostgreSQL")?;
    let migrator = kasl_server::migrator();
    // Worth a line: on a fresh install this is where the schema appears, and
    // on an upgrade it is the first thing to check when something looks off.
    let target = migrator.migrations.last().map(|m| m.version).unwrap_or_default();
    migrator.run(&pool).await.context("failed to apply database migrations")?;
    tracing::info!(version = target, "database schema is up to date");

    match cli.command {
        Some(Command::Import(args)) => run_import(&pool, args).await,
        Some(Command::Admin(args)) => {
            provision::ensure_admin(&pool, &args.email, &args.password).await?;
            println!("admin {} is ready", args.email);
            Ok(())
        }
        Some(Command::Backup(args)) => run_backup(&pool, target, args).await,
        Some(Command::Restore(args)) => run_restore(&pool, target, args).await,
        None => serve(pool, config).await,
    }
}

async fn run_backup(pool: &sqlx::PgPool, schema_version: i64, args: BackupArgs) -> Result<()> {
    // Progress goes to stderr throughout: stdout may be the backup itself,
    // and a friendly line in the middle of it would corrupt the file.
    let summary = match &args.out {
        Some(path) => {
            let file = std::fs::File::create(path).with_context(|| format!("failed to create {}", path.display()))?;
            // `dump` flushes what it wrote; dropping the writer here is what
            // gets the last of it onto disk.
            let mut writer = std::io::BufWriter::new(file);
            let summary = kasl_server::backup::dump(pool, schema_version, &mut writer).await?;
            drop(writer);
            eprintln!("wrote {} rows from {} tables to {}", summary.rows, summary.tables, path.display());
            summary
        }
        None => {
            let stdout = std::io::stdout();
            let mut writer = std::io::BufWriter::new(stdout.lock());
            let summary = kasl_server::backup::dump(pool, schema_version, &mut writer).await?;
            drop(writer);
            eprintln!("wrote {} rows from {} tables", summary.rows, summary.tables);
            summary
        }
    };

    if summary.rows == 0 {
        eprintln!("note: this installation holds no data yet");
    }
    Ok(())
}

async fn run_restore(pool: &sqlx::PgPool, schema_version: i64, args: RestoreArgs) -> Result<()> {
    let summary = match &args.from {
        Some(path) => {
            let file = std::fs::File::open(path).with_context(|| format!("failed to open {}", path.display()))?;
            kasl_server::backup::load(pool, schema_version, std::io::BufReader::new(file)).await?
        }
        None => {
            let stdin = std::io::stdin();
            kasl_server::backup::load(pool, schema_version, stdin.lock()).await?
        }
    };

    println!("restored {} rows into {} tables", summary.rows, summary.tables);
    Ok(())
}

async fn run_import(pool: &sqlx::PgPool, args: ImportArgs) -> Result<()> {
    // Resolved before the file is read: a typo in the email should cost the
    // operator a second, not the time it takes to parse a year of history.
    let user_id = import::resolve_user(pool, &args.user).await?;

    let (days, summary) = import::read_agent_db(&args.db)?;
    println!(
        "read {} workdays, {} pauses, {} tasks from {}",
        summary.days,
        summary.pauses,
        summary.tasks,
        args.db.display()
    );

    let days = import::within(days, args.since, args.until);
    if args.since.is_some() || args.until.is_some() {
        println!("selected {} days in range", days.len());
    }
    if summary.skipped_deleted_tasks > 0 {
        println!("skipped {} tasks the employee had deleted", summary.skipped_deleted_tasks);
    }
    if summary.skipped_unreadable > 0 {
        println!("skipped {} rows whose timestamps could not be read", summary.skipped_unreadable);
    }

    if args.dry_run {
        println!("dry run: nothing was written");
        return Ok(());
    }

    let written = import::write_days(pool, user_id, &days, args.timezone).await?;
    // The offset is echoed because it is the one thing that cannot be checked
    // afterwards by looking at the data: every instant is now stated relative
    // to it, and a wrong one looks entirely plausible.
    println!("imported {written} days as {} at {}", args.user, args.timezone);

    Ok(())
}

async fn serve(pool: sqlx::PgPool, config: config::Config) -> Result<()> {
    // The demo goes first: it needs the database empty, and everything below
    // adds to it. An agent from KASL_AGENTS lands on top of the fictional
    // team, which is how a real kasl gets pointed at a demo.
    match (config.demo, demo::status(&pool).await?) {
        (true, demo::Status::Empty) => {
            let seeded = demo::seed(&pool, chrono::Utc::now()).await?;
            tracing::info!(
                people = seeded.people,
                departments = seeded.departments,
                days = seeded.days,
                "seeded the demo team"
            );
            print_demo_logins();
            demo::keep_pulses_fresh(pool.clone());
        }
        (true, demo::Status::Demo) => {
            // A stand left running shows a team that stopped working the day
            // it was seeded - "no days recorded for 16 days" on every row,
            // which describes the data truthfully and the product falsely.
            // Regenerated rather than patched field by field: that was tried
            // twice (the pulse in v0.17.1, the calendar in v0.21) and each
            // time the next milestone brought another field to patch.
            match demo::history_is_stale(&pool, chrono::Utc::now()).await {
                Ok(true) => match demo::reseed(&pool, chrono::Utc::now()).await {
                    Ok(seeded) => tracing::info!(
                        people = seeded.people,
                        days = seeded.days,
                        "the demo's history had stopped reaching today; generated it again"
                    ),
                    Err(error) => tracing::warn!(%error, "failed to regenerate the demo"),
                },
                Ok(false) => {}
                Err(error) => tracing::warn!(%error, "failed to check whether the demo's history is current"),
            }

            print_demo_logins();
            // A demo seeded before the pulse existed has agents and no
            // pulses, and bumping the image does not re-seed: without this
            // its dashboard shows twelve rows of "unknown" and none of the
            // live column. Never overwrites a pulse that is already there.
            match demo::ensure_pulses(&pool).await {
                Ok(0) => {}
                Ok(given) => tracing::info!(given, "gave the demo's agents their pulses"),
                Err(error) => tracing::warn!(%error, "failed to give the demo's agents their pulses"),
            }
            // The same upgrade path for the calendar and the part-time rate:
            // a demo seeded before v0.21 has neither, and a dashboard of
            // twelve identical full norms is the one thing that version is
            // about, missing.
            match demo::ensure_calendar(&pool, chrono::Utc::now()).await {
                Ok(0) => {}
                Ok(written) => tracing::info!(written, "gave the demo its calendar and rates"),
                Err(error) => tracing::warn!(%error, "failed to give the demo its calendar"),
            }
            demo::keep_pulses_fresh(pool.clone());
        }
        (true, demo::Status::Populated { accounts }) => {
            // Refused rather than seeded alongside: a flag left in a file
            // after a trial would otherwise put twelve invented people on a
            // real team's dashboard, and nothing would say which are which.
            anyhow::bail!(
                "KASL_DEMO is set, but this database already holds {accounts} accounts that are not the demo's. \
                 Unset KASL_DEMO to start normally, or point the demo at an empty database."
            );
        }
        (false, demo::Status::Demo) => {
            // The flag is gone from the environment but the data is still
            // invented; the UI keeps saying so, because it reads the database.
            tracing::info!("this installation holds the demo team (KASL_DEMO is not set; the data stays labelled as a demo)");
        }
        (false, _) => {}
    }

    let seeds = provision::parse_seeds(&config.agents)?;
    provision::apply_seeds(&pool, &seeds).await?;

    // The bootstrap admin, same shape as KASL_AGENTS: a container has no other
    // way to be handed a first account.
    if let Some((email, password)) = provision::parse_admin(&config.admin)? {
        provision::ensure_admin(&pool, &email, &password).await?;
        tracing::info!(%email, "administrator from KASL_ADMIN is ready");
    } else if let Some(password) = provision::ensure_some_admin(&pool, &config.admin_email).await? {
        // The one moment this server prints a secret. An installation with no
        // administrator has no way in at all, and the alternative - refusing to
        // start until the operator sets one - turns the first run into a
        // documentation exercise and leaves that password in a file forever.
        //
        // Printed to stdout rather than logged: a log ships to wherever logs
        // go, and this belongs on the console of the person who just typed
        // `docker compose up`.
        println!("\n  An administrator account was created, because this installation had none:\n");
        println!("      email:    {}", config.admin_email);
        println!("      password: {password}\n");
        println!("  This is the only time it is shown. Sign in and change it.\n");
    }

    // Sessions that expired while the server was down are of no use to anyone.
    match kasl_server::session::sweep_expired(&pool).await {
        Ok(0) => {}
        Ok(swept) => tracing::info!(swept, "removed expired sessions"),
        Err(error) => tracing::warn!(%error, "failed to sweep expired sessions"),
    }

    // The sweep starts before the listener: an alert exists so that somebody
    // does not have to be looking, and the first thing a server that was down
    // overnight should do is notice what happened while it was.
    let webhooks = std::sync::Arc::new(config.webhooks.clone());
    kasl_server::alerts::run_sweeps(pool.clone(), webhooks.clone());

    // Named at startup, by label and kind and never by address: "is the
    // design channel wired up" should be answerable from the log, and the
    // log is not where a hook belongs.
    for destination in webhooks.destinations() {
        let events: Vec<&str> = destination.events.iter().map(|event| event.name()).collect();
        tracing::info!(
            destination = %destination.name,
            kind = ?destination.kind,
            target = %destination.shown_target(),
            events = %events.join(","),
            department = destination.department.as_deref().unwrap_or("everyone"),
            "webhook destination"
        );
    }
    // Always started, even with no destinations: a delivery queued before a
    // variable was removed still has to be given up on in writing.
    kasl_server::webhooks::run_dispatcher(pool.clone(), webhooks);

    let listener = TcpListener::bind(config.addr)
        .await
        .with_context(|| format!("failed to bind {}", config.addr))?;
    tracing::info!(version = env!("CARGO_PKG_VERSION"), addr = %config.addr, max_batch_days = config.max_batch_days, max_body_bytes = config.max_body_bytes, "kasl-server listening");

    axum::serve(listener, app::router_with(pool, &config))
        .with_graceful_shutdown(shutdown_signal())
        .await
        .context("server error")?;
    Ok(())
}

/// The accounts a visitor can try, one per role.
///
/// On stdout like the generated administrator's password, and for the same
/// reason: this belongs on the console of whoever just started the demo. Not
/// a secret - the password is in the README - so it is printed on every
/// start, not only the first.
fn print_demo_logins() {
    println!("\n  This is a demo: a fictional team, nothing here is real. Sign in as\n");
    for account in demo::showcase() {
        let role = serde_json::to_value(account.role)
            .ok()
            .and_then(|v| v.as_str().map(str::to_owned))
            .unwrap_or_default();
        println!("      {role:<9} {:<32} {}", account.email, account.display_name);
    }
    println!("\n  with the password `{}`. The same password opens every account.\n", demo::PASSWORD);
}

async fn shutdown_signal() {
    if let Err(error) = tokio::signal::ctrl_c().await {
        tracing::error!(%error, "failed to install the shutdown signal handler");
        return;
    }
    tracing::info!("shutting down");
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn reads_utc_offsets_in_the_forms_an_operator_would_type() {
        assert_eq!(parse_offset("-03:00").unwrap().local_minus_utc(), -3 * 3600);
        assert_eq!(parse_offset("+05:30").unwrap().local_minus_utc(), 5 * 3600 + 1800);
        assert_eq!(parse_offset("Z").unwrap().local_minus_utc(), 0);
    }

    #[test]
    fn refuses_something_that_is_not_an_offset() {
        // Notably a zone name: accepting it would imply DST handling this does
        // not do, and quietly importing a year at the wrong hour.
        assert!(parse_offset("America/Asuncion").is_err());
        assert!(parse_offset("-3").is_err());
        assert!(parse_offset("").is_err());
    }

    #[test]
    fn the_cli_still_runs_the_server_without_a_subcommand() {
        // The container entrypoint passes no arguments; that must keep meaning
        // "serve" now that subcommands exist.
        let cli = Cli::try_parse_from(["kasl-server"]).expect("no arguments must remain valid");
        assert!(cli.command.is_none());
    }

    #[test]
    fn the_import_requires_a_timezone() {
        let missing = Cli::try_parse_from(["kasl-server", "import", "--db", "kasl.db", "--user", "a@b.c"]);
        assert!(missing.is_err(), "an import without an offset must not start");

        let complete = Cli::try_parse_from(["kasl-server", "import", "--db", "kasl.db", "--user", "a@b.c", "--timezone", "-03:00"]);
        assert!(complete.is_ok(), "and with one it must");
    }
}