1use crate::api::kasl_server::{AGENT_TOKEN_PROMPT, AGENT_TOKEN_SECRET, KaslServer, UploadError, normalize_url};
24use crate::db::server_outbox::ServerOutbox;
25use crate::db::server_pulse::ServerPulse;
26use crate::db::workdays::Workdays;
27use crate::libs::config::{Config, KaslServerConfig};
28use crate::libs::daemon;
29use crate::libs::day_delivery::{Delivered, deliver, record_single};
30use crate::libs::day_upload::build_day_upload;
31use crate::libs::json;
32use crate::libs::messages::Message;
33use crate::libs::output::{self, Format};
34use crate::libs::pulse::{self, format_ago};
35use crate::libs::secret::Secret;
36use crate::{msg_error_anyhow, msg_info, msg_print, msg_success, msg_warning};
37use anyhow::{Context, Result, bail};
38use chrono::{Duration, Local, NaiveDate, Utc};
39use clap::{Args, Subcommand};
40use dialoguer::{Input, Password, theme::ColorfulTheme};
41use reqwest::StatusCode;
42
43#[derive(Debug, Args)]
45pub struct ServerArgs {
46 #[command(subcommand)]
47 command: ServerCommand,
48}
49
50#[derive(Debug, Subcommand)]
52enum ServerCommand {
53 #[command(about = "Connect this machine to a kasl-server")]
55 Connect(ConnectArgs),
56
57 #[command(about = "Show the current connection to a kasl-server")]
59 Status {
60 #[command(flatten)]
61 format: Format,
62 },
63
64 #[command(about = "Send a day's work to the connected kasl-server")]
66 Push(PushArgs),
67
68 #[command(about = "Send every day still waiting to reach the server")]
70 Flush,
71
72 #[command(about = "Show the days still waiting to reach the server")]
74 Queue {
75 #[command(flatten)]
76 format: Format,
77 },
78
79 #[command(about = "Show what the connected kasl-server stores about you")]
81 Manifest {
82 #[command(flatten)]
83 format: Format,
84 },
85
86 #[command(about = "Queue every recorded day in a date range and send them")]
88 Backfill(BackfillArgs),
89
90 #[command(about = "Turn the live pulse on or off, or show how it is going")]
92 Pulse(PulseArgs),
93
94 #[command(about = "Forget the connection and the stored agent token")]
96 Disconnect,
97}
98
99#[derive(Debug, Args)]
101pub struct BackfillArgs {
102 #[arg(long, value_name = "YYYY-MM-DD")]
104 from: Option<NaiveDate>,
105
106 #[arg(long, value_name = "YYYY-MM-DD")]
108 to: Option<NaiveDate>,
109}
110
111#[derive(Debug, Args)]
113pub struct PulseArgs {
114 #[command(subcommand)]
115 command: Option<PulseCommand>,
116
117 #[command(flatten)]
119 format: Format,
120}
121
122#[derive(Debug, Subcommand)]
125enum PulseCommand {
126 #[command(about = "Start telling the server whether you are working, on a break, or not in a day")]
128 Enable,
129
130 #[command(about = "Stop telling the server whether you are working right now")]
132 Disable,
133}
134
135#[derive(Debug, Args)]
137pub struct PushArgs {
138 #[arg(long, short, help = "Send the last day instead of today")]
140 last: bool,
141
142 #[arg(long, value_name = "YYYY-MM-DD", conflicts_with = "last")]
144 date: Option<NaiveDate>,
145}
146
147#[derive(Debug, Args)]
149pub struct ConnectArgs {
150 #[arg(long, value_name = "URL")]
152 url: Option<String>,
153
154 #[arg(long, value_name = "PATH")]
156 ca_certificate: Option<String>,
157}
158
159pub async fn cmd(args: ServerArgs) -> Result<()> {
161 match args.command {
162 ServerCommand::Connect(args) => connect(args).await,
163 ServerCommand::Status { format } => status(format.json).await,
164 ServerCommand::Push(args) => push(args).await,
165 ServerCommand::Flush => flush().await,
166 ServerCommand::Queue { format } => queue(format.json),
167 ServerCommand::Manifest { format } => manifest(format.json).await,
168 ServerCommand::Backfill(args) => backfill(args).await,
169 ServerCommand::Pulse(args) => match args.command {
170 Some(_) if args.format.json => bail!("--json applies to showing the pulse: `kasl server pulse --json`"),
172 Some(PulseCommand::Enable) => pulse_enable().await,
173 Some(PulseCommand::Disable) => pulse_disable(),
174 None => pulse_show(args.format.json),
175 },
176 ServerCommand::Disconnect => disconnect(),
177 }
178}
179
180async fn connect(args: ConnectArgs) -> Result<()> {
186 crate::libs::prompt::ensure_interactive("`kasl server connect` needs a terminal to ask for the agent token")?;
193
194 let mut config = Config::read().unwrap_or_default();
195
196 let url = match args.url {
197 Some(url) => normalize_url(&url),
198 None => {
199 let entered: String = Input::with_theme(&ColorfulTheme::default())
200 .with_prompt(Message::PromptKaslServerUrl.to_string())
201 .with_initial_text(config.kasl_server.as_ref().map(|s| s.url.clone()).unwrap_or_default())
202 .interact_text()?;
203 normalize_url(&entered)
204 }
205 };
206
207 if !url.starts_with("http://") && !url.starts_with("https://") {
210 return Err(msg_error_anyhow!(Message::KaslServerUrlNeedsScheme(url)));
211 }
212
213 let candidate = KaslServerConfig {
214 url: url.clone(),
215 ca_certificate: args
218 .ca_certificate
219 .or_else(|| config.kasl_server.as_ref().and_then(|s| s.ca_certificate.clone())),
220 pulse: keeps_pulse(config.kasl_server.as_ref(), &url),
221 };
222
223 let client = KaslServer::new(&candidate)?;
224
225 let health = client.health().await?;
228 msg_info!(Message::KaslServerReached {
229 url: url.clone(),
230 version: health.version.clone(),
231 });
232 if health.database != "ok" {
233 msg_warning!(Message::KaslServerDatabaseUnhealthy(health.database.clone()));
236 }
237
238 let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
239 let token: String = Password::with_theme(&ColorfulTheme::default())
240 .with_prompt(Message::PromptKaslServerToken.to_string())
241 .interact()?;
242 let token = token.trim().to_string();
243 if token.is_empty() {
244 return Err(msg_error_anyhow!(Message::KaslServerTokenEmpty));
245 }
246
247 let identity = client.identify(&token).await?;
249
250 secret
252 .store(&token)
253 .context("the token was accepted but could not be stored in the OS keyring")?;
254 config.kasl_server = Some(candidate);
255 config.save()?;
256
257 msg_success!(Message::KaslServerConnected {
258 user_name: identity.user_name,
259 agent_name: identity.agent_name,
260 });
261 Ok(())
262}
263
264async fn status(json: bool) -> Result<()> {
270 let config = Config::read().unwrap_or_default();
271 let Some(server_config) = config.kasl_server else {
272 if json {
273 return output::print_json(&json::ServerStatus {
274 connected: false,
275 url: None,
276 token: None,
277 reachable: None,
278 server_version: None,
279 api_version: None,
280 user: None,
281 agent: None,
282 problem: None,
283 pulse: None,
284 });
285 }
286 msg_print!(Message::KaslServerNotConnected);
287 return Ok(());
288 };
289
290 let check = check_connection(&server_config).await?;
291 if json {
292 let identity = check.identity.as_ref().and_then(|identity| identity.as_ref().ok());
293 return output::print_json(&json::ServerStatus {
294 connected: true,
295 url: Some(server_config.url.clone()),
296 token: Some(check.token),
297 reachable: check.health.as_ref().map(Result::is_ok),
298 server_version: check.health.as_ref().and_then(|health| health.as_ref().ok().cloned()),
299 api_version: identity.map(|identity| identity.api_version.clone()),
300 user: identity.map(|identity| identity.user_name.clone()),
301 agent: identity.map(|identity| identity.agent_name.clone()),
302 problem: check.problem().map(|problem| problem.to_string()),
303 pulse: Some(pulse_json(server_config.pulse)?),
304 });
305 }
306
307 msg_info!(Message::KaslServerConfigured(server_config.url.clone()));
308 report_connection(&server_config, check);
309
310 report_pulse(server_config.pulse)
314}
315
316struct ConnectionCheck {
321 token: bool,
323 health: Option<Result<String, String>>,
325 identity: Option<Result<crate::api::kasl_server::AgentIdentity, String>>,
327}
328
329impl ConnectionCheck {
330 fn problem(&self) -> Option<Message> {
332 if !self.token {
333 return Some(Message::KaslServerTokenMissing);
334 }
335 if let Some(Err(error)) = &self.health {
336 return Some(Message::KaslServerUnreachable(error.clone()));
337 }
338 if let Some(Err(error)) = &self.identity {
339 return Some(Message::KaslServerTokenRejected(error.clone()));
340 }
341 None
342 }
343}
344
345async fn check_connection(server_config: &KaslServerConfig) -> Result<ConnectionCheck> {
347 let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
348 let Some(token) = secret.try_get_cached() else {
349 return Ok(ConnectionCheck {
350 token: false,
351 health: None,
352 identity: None,
353 });
354 };
355
356 let client = KaslServer::new(server_config)?;
357 let health = client.health().await.map(|health| health.version).map_err(|error| error.to_string());
358 if health.is_err() {
359 return Ok(ConnectionCheck {
360 token: true,
361 health: Some(health),
362 identity: None,
363 });
364 }
365 let identity = client.identify(&token).await.map_err(|error| error.to_string());
366 Ok(ConnectionCheck {
367 token: true,
368 health: Some(health),
369 identity: Some(identity),
370 })
371}
372
373fn report_connection(server_config: &KaslServerConfig, check: ConnectionCheck) {
375 if !check.token {
376 msg_warning!(Message::KaslServerTokenMissing);
379 return;
380 }
381
382 match check.health {
383 Some(Ok(version)) => msg_info!(Message::KaslServerReached {
384 url: server_config.url.clone(),
385 version,
386 }),
387 Some(Err(error)) => {
388 msg_warning!(Message::KaslServerUnreachable(error));
389 return;
390 }
391 None => return,
392 }
393
394 match check.identity {
395 Some(Ok(identity)) => {
396 msg_success!(Message::KaslServerConnected {
397 user_name: identity.user_name,
398 agent_name: identity.agent_name,
399 });
400 msg_info!(Message::KaslServerCompatibility {
407 server_version: identity.server_version,
408 api_version: identity.api_version,
409 });
410 }
411 Some(Err(error)) => msg_warning!(Message::KaslServerTokenRejected(error)),
412 None => {}
413 }
414}
415
416fn report_pulse(enabled: bool) -> Result<()> {
418 let record = if enabled { ServerPulse::new()?.last()? } else { None };
419 for line in pulse::describe(enabled, record.as_ref(), Utc::now()) {
420 if line.warning {
421 msg_warning!(line.message);
422 } else {
423 msg_info!(line.message);
424 }
425 }
426 Ok(())
427}
428
429fn pulse_json(enabled: bool) -> Result<json::Pulse> {
431 let record = if enabled { ServerPulse::new()?.last()? } else { None };
432 let warnings = pulse::describe(enabled, record.as_ref(), Utc::now())
433 .into_iter()
434 .filter(|line| line.warning)
435 .map(|line| line.message.to_string())
436 .collect();
437 Ok(json::Pulse {
438 enabled,
439 sent_at: record.as_ref().and_then(|record| record.sent_at),
440 state: record.as_ref().and_then(|record| record.state).map(|state| state.as_str()),
441 attempted_at: record.as_ref().map(|record| record.attempted_at),
442 error: record.as_ref().and_then(|record| record.error.clone()),
443 warnings,
444 })
445}
446
447fn keeps_pulse(previous: Option<&KaslServerConfig>, url: &str) -> bool {
453 previous.is_some_and(|previous| previous.pulse && previous.url == url)
454}
455
456const FIRST_PULSE_WAIT: std::time::Duration = std::time::Duration::from_secs(12);
461
462async fn pulse_enable() -> Result<()> {
470 let mut config = Config::read().unwrap_or_default();
471 let Some(server) = config.kasl_server.as_mut() else {
472 return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
473 };
474 if Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).try_get_cached().is_none() {
477 return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
478 }
479
480 ServerPulse::new()?.clear()?;
481 server.pulse = true;
482 let url = server.url.clone();
483 config.save()?;
484 msg_success!(Message::PulseEnabled(url));
485
486 if !daemon::is_running() {
487 msg_warning!(Message::PulseNoWatcher);
488 return Ok(());
489 }
490
491 let deadline = std::time::Instant::now() + FIRST_PULSE_WAIT;
492 loop {
493 if let Some(record) = ServerPulse::new()?.last()? {
494 match (record.error, record.state) {
495 (None, Some(state)) => msg_success!(Message::PulseFirstArrived(state.to_string())),
496 (Some(error), _) => msg_warning!(Message::PulseFirstFailed(error)),
497 (None, None) => msg_info!(Message::PulseFirstPending),
498 }
499 return Ok(());
500 }
501 if std::time::Instant::now() >= deadline {
502 msg_info!(Message::PulseFirstPending);
503 return Ok(());
504 }
505 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
506 }
507}
508
509fn pulse_disable() -> Result<()> {
516 let mut config = Config::read().unwrap_or_default();
517 let Some(server) = config.kasl_server.as_mut() else {
518 msg_print!(Message::KaslServerNotConnected);
519 return Ok(());
520 };
521 if !server.pulse {
522 msg_print!(Message::PulseAlreadyOff);
523 return Ok(());
524 }
525
526 server.pulse = false;
527 config.save()?;
528
529 let mut store = ServerPulse::new()?;
530 let stale_after = store.last()?.and_then(|record| record.stale_after_seconds).map(format_ago);
531 store.clear()?;
532
533 msg_success!(Message::PulseDisabled(stale_after));
534 Ok(())
535}
536
537fn pulse_show(json: bool) -> Result<()> {
539 let config = Config::read().unwrap_or_default();
540 let Some(server) = config.kasl_server else {
541 if json {
542 return output::print_json(&pulse_json(false)?);
544 }
545 msg_print!(Message::KaslServerNotConnected);
546 return Ok(());
547 };
548 if json {
549 return output::print_json(&pulse_json(server.pulse)?);
550 }
551 report_pulse(server.pulse)
552}
553
554async fn manifest(json: bool) -> Result<()> {
568 let (client, token) = connected_client()?;
569
570 let manifest = client.privacy(&token).await?;
571 if json {
572 return output::print_json(&json::Manifest::from(manifest));
573 }
574
575 msg_info!(Message::KaslServerPrivacyHeading(manifest.level));
576 msg_print!(Message::KaslServerPrivacySummary(manifest.summary));
577
578 msg_print!(Message::KaslServerPrivacyStoredHeading);
579 for stored in manifest.stored {
580 msg_print!(Message::KaslServerPrivacyStored {
581 what: stored.what,
582 detail: stored.detail,
583 });
584 }
585
586 msg_print!(Message::KaslServerPrivacyNeverHeading);
587 for line in manifest.never_collected {
588 msg_print!(Message::KaslServerPrivacyBullet(line));
589 }
590
591 msg_print!(Message::KaslServerPrivacyVisibleHeading);
592 for line in manifest.visible_to {
593 msg_print!(Message::KaslServerPrivacyBullet(line));
594 }
595
596 msg_print!(Message::KaslServerPrivacyRetention(manifest.retention));
597 msg_print!(Message::KaslServerPrivacyOnChange(manifest.on_change));
598
599 if let Some(updated_at) = manifest.updated_at {
603 msg_print!(Message::KaslServerPrivacyUpdatedAt(
604 updated_at.with_timezone(&Local).format("%Y-%m-%d %H:%M").to_string()
605 ));
606 }
607
608 msg_print!(Message::KaslServerPrivacySetByAdmin);
609
610 Ok(())
611}
612
613async fn push(args: PushArgs) -> Result<()> {
630 let date = match args.date {
631 Some(date) => date,
632 None if args.last => (Local::now() - Duration::days(1)).date_naive(),
633 None => Local::now().date_naive(),
634 };
635
636 let config = Config::read().unwrap_or_default();
637 let Some(server_config) = config.kasl_server else {
638 return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
639 };
640
641 let Some(day) = build_day_upload(date)? else {
642 msg_print!(Message::KaslServerNoDayToPush(date.to_string()));
643 return Ok(());
644 };
645
646 let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
647 let Some(token) = secret.try_get_cached() else {
648 return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
649 };
650
651 let client = KaslServer::new(&server_config)?;
652
653 match client.upload_day(&token, &day).await {
654 Ok(accepted) => {
655 msg_success!(Message::KaslServerDayPushed {
656 date: accepted.date.to_string(),
657 pauses: accepted.pauses,
658 tasks: accepted.tasks,
659 });
660 if accepted.deleted_tasks > 0 {
664 msg_info!(Message::KaslServerTasksDeleted(accepted.deleted_tasks));
665 }
666
667 ServerOutbox::new()?.remove(date)?;
671
672 flush_with(&client, &token).await?;
677 Ok(())
678 }
679 Err(error) => {
685 let outcome = record_single(&mut ServerOutbox::new()?, date, &error)?;
689 if matches!(outcome, Delivered::Deferred { .. }) {
690 msg_info!(Message::KaslServerDayQueued(date.to_string()));
691 }
692
693 match error {
694 error @ UploadError::Rejected {
695 status: StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN,
696 ..
697 } => Err(msg_error_anyhow!(Message::KaslServerPushTokenRejected(error.to_string()))),
698 error @ UploadError::Rejected { .. } => Err(msg_error_anyhow!(Message::KaslServerPushRejected(error.to_string()))),
699 error => Err(msg_error_anyhow!(Message::KaslServerPushRetryable(error.to_string()))),
700 }
701 }
702 }
703}
704
705async fn flush() -> Result<()> {
707 if ServerOutbox::new()?.count()? == 0 {
713 msg_print!(Message::KaslServerQueueEmpty);
714 return Ok(());
715 }
716
717 let (client, token) = connected_client()?;
718 flush_with(&client, &token).await
719}
720
721async fn flush_with(client: &KaslServer, token: &str) -> Result<()> {
726 let mut outbox = ServerOutbox::new()?;
727 let dates: Vec<NaiveDate> = outbox.pending()?.into_iter().map(|owed| owed.date).collect();
728 if dates.is_empty() {
729 return Ok(());
730 }
731
732 msg_info!(Message::KaslServerQueueSending(dates.len()));
733
734 let outcomes = deliver(client, token, &mut outbox, &dates).await?;
735
736 let (mut accepted, mut refused, mut deferred) = (0, 0, 0);
737 for outcome in &outcomes {
738 match outcome {
739 Delivered::Accepted {
740 date,
741 pauses,
742 tasks,
743 deleted_tasks,
744 } => {
745 accepted += 1;
746 msg_success!(Message::KaslServerDayPushed {
747 date: date.to_string(),
748 pauses: *pauses,
749 tasks: *tasks,
750 });
751 if *deleted_tasks > 0 {
752 msg_info!(Message::KaslServerTasksDeleted(*deleted_tasks));
753 }
754 }
755 Delivered::Refused { date, reason } => {
760 refused += 1;
761 msg_warning!(Message::KaslServerDayRefused {
762 date: date.to_string(),
763 reason: reason.clone(),
764 });
765 }
766 Delivered::Deferred { date, reason } => {
767 deferred += 1;
768 msg_warning!(Message::KaslServerDayDeferred {
769 date: date.to_string(),
770 reason: reason.clone(),
771 });
772 }
773 }
774 }
775
776 msg_print!(Message::KaslServerFlushSummary { accepted, refused, deferred });
777 Ok(())
778}
779
780fn queue(json: bool) -> Result<()> {
785 let outbox = ServerOutbox::new()?;
786 let owed = outbox.pending()?;
787 if json {
788 return output::print_json(&json::Queue {
789 days: owed.iter().map(json::OwedDay::from).collect(),
790 });
791 }
792
793 if owed.is_empty() {
794 msg_print!(Message::KaslServerQueueEmpty);
795 return Ok(());
796 }
797
798 msg_info!(Message::KaslServerQueueOwed(owed.len() as i64));
799 for day in &owed {
800 msg_print!(Message::KaslServerQueueEntry {
801 date: day.date.to_string(),
802 attempts: day.attempts,
803 last_error: day.last_error.clone(),
804 });
805 }
806
807 Ok(())
808}
809
810async fn backfill(args: BackfillArgs) -> Result<()> {
823 let to = args.to.unwrap_or_else(|| Local::now().date_naive());
824 if let Some(from) = args.from
825 && from > to
826 {
827 return Err(msg_error_anyhow!(Message::KaslServerBackfillOrderReversed));
828 }
829
830 let (client, token) = connected_client()?;
831
832 let dates = Workdays::new()?.recorded_dates(args.from, Some(to))?;
833
834 if dates.is_empty() {
835 match args.from {
836 Some(from) => msg_print!(Message::KaslServerBackfillNoDays {
837 from: from.to_string(),
838 to: to.to_string(),
839 }),
840 None => msg_print!(Message::KaslServerBackfillNothingRecorded),
846 }
847 return Ok(());
848 }
849
850 match args.from {
851 Some(from) => msg_info!(Message::KaslServerBackfillRange {
852 from: from.to_string(),
853 to: to.to_string(),
854 days: dates.len(),
855 }),
856 None => msg_info!(Message::KaslServerBackfillWholeHistory {
861 from: dates[0].to_string(),
862 to: to.to_string(),
863 days: dates.len(),
864 }),
865 }
866
867 let mut outbox = ServerOutbox::new()?;
870 for date in &dates {
871 outbox.enqueue(*date, "queued by backfill")?;
872 }
873
874 flush_with(&client, &token).await
875}
876
877fn connected_client() -> Result<(KaslServer, String)> {
883 let config = Config::read().unwrap_or_default();
884 let Some(server_config) = config.kasl_server else {
885 return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
886 };
887
888 let Some(token) = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).try_get_cached() else {
889 return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
890 };
891
892 Ok((KaslServer::new(&server_config)?, token))
893}
894
895fn disconnect() -> Result<()> {
901 let mut config = Config::read().unwrap_or_default();
902
903 if let Err(error) = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).delete() {
910 msg_warning!(Message::KaslServerTokenNotRemoved(error.to_string()));
911 }
912
913 if config.kasl_server.take().is_some() {
914 config.save()?;
915 ServerPulse::new()?.clear()?;
918 msg_success!(Message::KaslServerDisconnected);
919 } else {
920 msg_print!(Message::KaslServerNotConnected);
922 }
923
924 Ok(())
925}
926
927#[cfg(test)]
928mod tests {
929 use super::*;
930
931 fn connection(url: &str, pulse: bool) -> KaslServerConfig {
932 KaslServerConfig {
933 url: url.to_string(),
934 ca_certificate: None,
935 pulse,
936 }
937 }
938
939 #[test]
940 fn reconnecting_to_the_same_server_keeps_the_pulse() {
941 assert!(keeps_pulse(Some(&connection("https://kasl.example.com", true)), "https://kasl.example.com"));
942 }
943
944 #[test]
945 fn another_server_does_not_inherit_the_consent() {
946 assert!(!keeps_pulse(Some(&connection("https://kasl.example.com", true)), "https://kasl.other.example"));
949 assert!(!keeps_pulse(Some(&connection("https://kasl.example.com", false)), "https://kasl.example.com"));
950 assert!(!keeps_pulse(None, "https://kasl.example.com"));
951 }
952}