use crate::api::kasl_server::{AGENT_TOKEN_PROMPT, AGENT_TOKEN_SECRET, KaslServer, UploadError, normalize_url};
use crate::db::server_outbox::ServerOutbox;
use crate::db::server_pulse::ServerPulse;
use crate::db::workdays::Workdays;
use crate::libs::config::{Config, KaslServerConfig};
use crate::libs::daemon;
use crate::libs::day_delivery::{Delivered, deliver, record_single};
use crate::libs::day_upload::build_day_upload;
use crate::libs::messages::Message;
use crate::libs::pulse::{self, format_ago};
use crate::libs::secret::Secret;
use crate::{msg_error_anyhow, msg_info, msg_print, msg_success, msg_warning};
use anyhow::{Context, Result};
use chrono::{Duration, Local, NaiveDate, Utc};
use clap::{Args, Subcommand};
use dialoguer::{Input, Password, theme::ColorfulTheme};
use reqwest::StatusCode;
#[derive(Debug, Args)]
pub struct ServerArgs {
#[command(subcommand)]
command: ServerCommand,
}
#[derive(Debug, Subcommand)]
enum ServerCommand {
#[command(about = "Connect this machine to a kasl-server")]
Connect(ConnectArgs),
#[command(about = "Show the current connection to a kasl-server")]
Status,
#[command(about = "Send a day's work to the connected kasl-server")]
Push(PushArgs),
#[command(about = "Send every day still waiting to reach the server")]
Flush,
#[command(about = "Show the days still waiting to reach the server")]
Queue,
#[command(about = "Show what the connected kasl-server stores about you")]
Manifest,
#[command(about = "Queue every recorded day in a date range and send them")]
Backfill(BackfillArgs),
#[command(about = "Turn the live pulse on or off, or show how it is going")]
Pulse(PulseArgs),
#[command(about = "Forget the connection and the stored agent token")]
Disconnect,
}
#[derive(Debug, Args)]
pub struct BackfillArgs {
#[arg(long, value_name = "YYYY-MM-DD")]
from: Option<NaiveDate>,
#[arg(long, value_name = "YYYY-MM-DD")]
to: Option<NaiveDate>,
}
#[derive(Debug, Args)]
pub struct PulseArgs {
#[command(subcommand)]
command: Option<PulseCommand>,
}
#[derive(Debug, Subcommand)]
enum PulseCommand {
#[command(about = "Start telling the server whether you are working, on a break, or not in a day")]
Enable,
#[command(about = "Stop telling the server whether you are working right now")]
Disable,
}
#[derive(Debug, Args)]
pub struct PushArgs {
#[arg(long, short, help = "Send the last day instead of today")]
last: bool,
#[arg(long, value_name = "YYYY-MM-DD", conflicts_with = "last")]
date: Option<NaiveDate>,
}
#[derive(Debug, Args)]
pub struct ConnectArgs {
#[arg(long, value_name = "URL")]
url: Option<String>,
#[arg(long, value_name = "PATH")]
ca_certificate: Option<String>,
}
pub async fn cmd(args: ServerArgs) -> Result<()> {
match args.command {
ServerCommand::Connect(args) => connect(args).await,
ServerCommand::Status => status().await,
ServerCommand::Push(args) => push(args).await,
ServerCommand::Flush => flush().await,
ServerCommand::Queue => queue(),
ServerCommand::Manifest => manifest().await,
ServerCommand::Backfill(args) => backfill(args).await,
ServerCommand::Pulse(args) => match args.command {
Some(PulseCommand::Enable) => pulse_enable().await,
Some(PulseCommand::Disable) => pulse_disable(),
None => pulse_show(),
},
ServerCommand::Disconnect => disconnect(),
}
}
async fn connect(args: ConnectArgs) -> Result<()> {
crate::libs::prompt::ensure_interactive("`kasl server connect` needs a terminal to ask for the agent token")?;
let mut config = Config::read().unwrap_or_default();
let url = match args.url {
Some(url) => normalize_url(&url),
None => {
let entered: String = Input::with_theme(&ColorfulTheme::default())
.with_prompt(Message::PromptKaslServerUrl.to_string())
.with_initial_text(config.kasl_server.as_ref().map(|s| s.url.clone()).unwrap_or_default())
.interact_text()?;
normalize_url(&entered)
}
};
if !url.starts_with("http://") && !url.starts_with("https://") {
return Err(msg_error_anyhow!(Message::KaslServerUrlNeedsScheme(url)));
}
let candidate = KaslServerConfig {
url: url.clone(),
ca_certificate: args
.ca_certificate
.or_else(|| config.kasl_server.as_ref().and_then(|s| s.ca_certificate.clone())),
pulse: keeps_pulse(config.kasl_server.as_ref(), &url),
};
let client = KaslServer::new(&candidate)?;
let health = client.health().await?;
msg_info!(Message::KaslServerReached {
url: url.clone(),
version: health.version.clone(),
});
if health.database != "ok" {
msg_warning!(Message::KaslServerDatabaseUnhealthy(health.database.clone()));
}
let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
let token: String = Password::with_theme(&ColorfulTheme::default())
.with_prompt(Message::PromptKaslServerToken.to_string())
.interact()?;
let token = token.trim().to_string();
if token.is_empty() {
return Err(msg_error_anyhow!(Message::KaslServerTokenEmpty));
}
let identity = client.identify(&token).await?;
secret
.store(&token)
.context("the token was accepted but could not be stored in the OS keyring")?;
config.kasl_server = Some(candidate);
config.save()?;
msg_success!(Message::KaslServerConnected {
user_name: identity.user_name,
agent_name: identity.agent_name,
});
Ok(())
}
async fn status() -> Result<()> {
let config = Config::read().unwrap_or_default();
let Some(server_config) = config.kasl_server else {
msg_print!(Message::KaslServerNotConnected);
return Ok(());
};
msg_info!(Message::KaslServerConfigured(server_config.url.clone()));
report_connection(&server_config).await?;
report_pulse(server_config.pulse)
}
async fn report_connection(server_config: &KaslServerConfig) -> Result<()> {
let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
let Some(token) = secret.try_get_cached() else {
msg_warning!(Message::KaslServerTokenMissing);
return Ok(());
};
let client = KaslServer::new(server_config)?;
match client.health().await {
Ok(health) => msg_info!(Message::KaslServerReached {
url: server_config.url.clone(),
version: health.version,
}),
Err(error) => {
msg_warning!(Message::KaslServerUnreachable(error.to_string()));
return Ok(());
}
}
match client.identify(&token).await {
Ok(identity) => {
msg_success!(Message::KaslServerConnected {
user_name: identity.user_name,
agent_name: identity.agent_name,
});
msg_info!(Message::KaslServerCompatibility {
server_version: identity.server_version,
api_version: identity.api_version,
});
}
Err(error) => msg_warning!(Message::KaslServerTokenRejected(error.to_string())),
}
Ok(())
}
fn report_pulse(enabled: bool) -> Result<()> {
let record = if enabled { ServerPulse::new()?.last()? } else { None };
for line in pulse::describe(enabled, record.as_ref(), Utc::now()) {
if line.warning {
msg_warning!(line.message);
} else {
msg_info!(line.message);
}
}
Ok(())
}
fn keeps_pulse(previous: Option<&KaslServerConfig>, url: &str) -> bool {
previous.is_some_and(|previous| previous.pulse && previous.url == url)
}
const FIRST_PULSE_WAIT: std::time::Duration = std::time::Duration::from_secs(12);
async fn pulse_enable() -> Result<()> {
let mut config = Config::read().unwrap_or_default();
let Some(server) = config.kasl_server.as_mut() else {
return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
};
if Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).try_get_cached().is_none() {
return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
}
ServerPulse::new()?.clear()?;
server.pulse = true;
let url = server.url.clone();
config.save()?;
msg_success!(Message::PulseEnabled(url));
if !daemon::is_running() {
msg_warning!(Message::PulseNoWatcher);
return Ok(());
}
let deadline = std::time::Instant::now() + FIRST_PULSE_WAIT;
loop {
if let Some(record) = ServerPulse::new()?.last()? {
match (record.error, record.state) {
(None, Some(state)) => msg_success!(Message::PulseFirstArrived(state.to_string())),
(Some(error), _) => msg_warning!(Message::PulseFirstFailed(error)),
(None, None) => msg_info!(Message::PulseFirstPending),
}
return Ok(());
}
if std::time::Instant::now() >= deadline {
msg_info!(Message::PulseFirstPending);
return Ok(());
}
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
}
}
fn pulse_disable() -> Result<()> {
let mut config = Config::read().unwrap_or_default();
let Some(server) = config.kasl_server.as_mut() else {
msg_print!(Message::KaslServerNotConnected);
return Ok(());
};
if !server.pulse {
msg_print!(Message::PulseAlreadyOff);
return Ok(());
}
server.pulse = false;
config.save()?;
let mut store = ServerPulse::new()?;
let stale_after = store.last()?.and_then(|record| record.stale_after_seconds).map(format_ago);
store.clear()?;
msg_success!(Message::PulseDisabled(stale_after));
Ok(())
}
fn pulse_show() -> Result<()> {
let config = Config::read().unwrap_or_default();
let Some(server) = config.kasl_server else {
msg_print!(Message::KaslServerNotConnected);
return Ok(());
};
report_pulse(server.pulse)
}
async fn manifest() -> Result<()> {
let (client, token) = connected_client()?;
let manifest = client.privacy(&token).await?;
msg_info!(Message::KaslServerPrivacyHeading(manifest.level));
msg_print!(Message::KaslServerPrivacySummary(manifest.summary));
msg_print!(Message::KaslServerPrivacyStoredHeading);
for stored in manifest.stored {
msg_print!(Message::KaslServerPrivacyStored {
what: stored.what,
detail: stored.detail,
});
}
msg_print!(Message::KaslServerPrivacyNeverHeading);
for line in manifest.never_collected {
msg_print!(Message::KaslServerPrivacyBullet(line));
}
msg_print!(Message::KaslServerPrivacyVisibleHeading);
for line in manifest.visible_to {
msg_print!(Message::KaslServerPrivacyBullet(line));
}
msg_print!(Message::KaslServerPrivacyRetention(manifest.retention));
msg_print!(Message::KaslServerPrivacyOnChange(manifest.on_change));
if let Some(updated_at) = manifest.updated_at {
msg_print!(Message::KaslServerPrivacyUpdatedAt(
updated_at.with_timezone(&Local).format("%Y-%m-%d %H:%M").to_string()
));
}
msg_print!(Message::KaslServerPrivacySetByAdmin);
Ok(())
}
async fn push(args: PushArgs) -> Result<()> {
let date = match args.date {
Some(date) => date,
None if args.last => (Local::now() - Duration::days(1)).date_naive(),
None => Local::now().date_naive(),
};
let config = Config::read().unwrap_or_default();
let Some(server_config) = config.kasl_server else {
return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
};
let Some(day) = build_day_upload(date)? else {
msg_print!(Message::KaslServerNoDayToPush(date.to_string()));
return Ok(());
};
let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
let Some(token) = secret.try_get_cached() else {
return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
};
let client = KaslServer::new(&server_config)?;
match client.upload_day(&token, &day).await {
Ok(accepted) => {
msg_success!(Message::KaslServerDayPushed {
date: accepted.date.to_string(),
pauses: accepted.pauses,
tasks: accepted.tasks,
});
if accepted.deleted_tasks > 0 {
msg_info!(Message::KaslServerTasksDeleted(accepted.deleted_tasks));
}
ServerOutbox::new()?.remove(date)?;
flush_with(&client, &token).await?;
Ok(())
}
Err(error) => {
let outcome = record_single(&mut ServerOutbox::new()?, date, &error)?;
if matches!(outcome, Delivered::Deferred { .. }) {
msg_info!(Message::KaslServerDayQueued(date.to_string()));
}
match error {
error @ UploadError::Rejected {
status: StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN,
..
} => Err(msg_error_anyhow!(Message::KaslServerPushTokenRejected(error.to_string()))),
error @ UploadError::Rejected { .. } => Err(msg_error_anyhow!(Message::KaslServerPushRejected(error.to_string()))),
error => Err(msg_error_anyhow!(Message::KaslServerPushRetryable(error.to_string()))),
}
}
}
}
async fn flush() -> Result<()> {
if ServerOutbox::new()?.count()? == 0 {
msg_print!(Message::KaslServerQueueEmpty);
return Ok(());
}
let (client, token) = connected_client()?;
flush_with(&client, &token).await
}
async fn flush_with(client: &KaslServer, token: &str) -> Result<()> {
let mut outbox = ServerOutbox::new()?;
let dates: Vec<NaiveDate> = outbox.pending()?.into_iter().map(|owed| owed.date).collect();
if dates.is_empty() {
return Ok(());
}
msg_info!(Message::KaslServerQueueSending(dates.len()));
let outcomes = deliver(client, token, &mut outbox, &dates).await?;
let (mut accepted, mut refused, mut deferred) = (0, 0, 0);
for outcome in &outcomes {
match outcome {
Delivered::Accepted {
date,
pauses,
tasks,
deleted_tasks,
} => {
accepted += 1;
msg_success!(Message::KaslServerDayPushed {
date: date.to_string(),
pauses: *pauses,
tasks: *tasks,
});
if *deleted_tasks > 0 {
msg_info!(Message::KaslServerTasksDeleted(*deleted_tasks));
}
}
Delivered::Refused { date, reason } => {
refused += 1;
msg_warning!(Message::KaslServerDayRefused {
date: date.to_string(),
reason: reason.clone(),
});
}
Delivered::Deferred { date, reason } => {
deferred += 1;
msg_warning!(Message::KaslServerDayDeferred {
date: date.to_string(),
reason: reason.clone(),
});
}
}
}
msg_print!(Message::KaslServerFlushSummary { accepted, refused, deferred });
Ok(())
}
fn queue() -> Result<()> {
let outbox = ServerOutbox::new()?;
let owed = outbox.pending()?;
if owed.is_empty() {
msg_print!(Message::KaslServerQueueEmpty);
return Ok(());
}
msg_info!(Message::KaslServerQueueOwed(owed.len() as i64));
for day in &owed {
msg_print!(Message::KaslServerQueueEntry {
date: day.date.to_string(),
attempts: day.attempts,
last_error: day.last_error.clone(),
});
}
Ok(())
}
async fn backfill(args: BackfillArgs) -> Result<()> {
let to = args.to.unwrap_or_else(|| Local::now().date_naive());
if let Some(from) = args.from
&& from > to
{
return Err(msg_error_anyhow!(Message::KaslServerBackfillOrderReversed));
}
let (client, token) = connected_client()?;
let dates = Workdays::new()?.recorded_dates(args.from, Some(to))?;
if dates.is_empty() {
match args.from {
Some(from) => msg_print!(Message::KaslServerBackfillNoDays {
from: from.to_string(),
to: to.to_string(),
}),
None => msg_print!(Message::KaslServerBackfillNothingRecorded),
}
return Ok(());
}
match args.from {
Some(from) => msg_info!(Message::KaslServerBackfillRange {
from: from.to_string(),
to: to.to_string(),
days: dates.len(),
}),
None => msg_info!(Message::KaslServerBackfillWholeHistory {
from: dates[0].to_string(),
to: to.to_string(),
days: dates.len(),
}),
}
let mut outbox = ServerOutbox::new()?;
for date in &dates {
outbox.enqueue(*date, "queued by backfill")?;
}
flush_with(&client, &token).await
}
fn connected_client() -> Result<(KaslServer, String)> {
let config = Config::read().unwrap_or_default();
let Some(server_config) = config.kasl_server else {
return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
};
let Some(token) = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).try_get_cached() else {
return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
};
Ok((KaslServer::new(&server_config)?, token))
}
fn disconnect() -> Result<()> {
let mut config = Config::read().unwrap_or_default();
if let Err(error) = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).delete() {
msg_warning!(Message::KaslServerTokenNotRemoved(error.to_string()));
}
if config.kasl_server.take().is_some() {
config.save()?;
ServerPulse::new()?.clear()?;
msg_success!(Message::KaslServerDisconnected);
} else {
msg_print!(Message::KaslServerNotConnected);
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn connection(url: &str, pulse: bool) -> KaslServerConfig {
KaslServerConfig {
url: url.to_string(),
ca_certificate: None,
pulse,
}
}
#[test]
fn reconnecting_to_the_same_server_keeps_the_pulse() {
assert!(keeps_pulse(Some(&connection("https://kasl.example.com", true)), "https://kasl.example.com"));
}
#[test]
fn another_server_does_not_inherit_the_consent() {
assert!(!keeps_pulse(Some(&connection("https://kasl.example.com", true)), "https://kasl.other.example"));
assert!(!keeps_pulse(Some(&connection("https://kasl.example.com", false)), "https://kasl.example.com"));
assert!(!keeps_pulse(None, "https://kasl.example.com"));
}
}