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::messages::Message;
32use crate::libs::pulse::{self, format_ago};
33use crate::libs::secret::Secret;
34use crate::{msg_error_anyhow, msg_info, msg_print, msg_success, msg_warning};
35use anyhow::{Context, Result};
36use chrono::{Duration, Local, NaiveDate, Utc};
37use clap::{Args, Subcommand};
38use dialoguer::{Input, Password, theme::ColorfulTheme};
39use reqwest::StatusCode;
40
41#[derive(Debug, Args)]
43pub struct ServerArgs {
44 #[command(subcommand)]
45 command: ServerCommand,
46}
47
48#[derive(Debug, Subcommand)]
50enum ServerCommand {
51 #[command(about = "Connect this machine to a kasl-server")]
53 Connect(ConnectArgs),
54
55 #[command(about = "Show the current connection to a kasl-server")]
57 Status,
58
59 #[command(about = "Send a day's work to the connected kasl-server")]
61 Push(PushArgs),
62
63 #[command(about = "Send every day still waiting to reach the server")]
65 Flush,
66
67 #[command(about = "Show the days still waiting to reach the server")]
69 Queue,
70
71 #[command(about = "Show what the connected kasl-server stores about you")]
73 Manifest,
74
75 #[command(about = "Queue every recorded day in a date range and send them")]
77 Backfill(BackfillArgs),
78
79 #[command(about = "Turn the live pulse on or off, or show how it is going")]
81 Pulse(PulseArgs),
82
83 #[command(about = "Forget the connection and the stored agent token")]
85 Disconnect,
86}
87
88#[derive(Debug, Args)]
90pub struct BackfillArgs {
91 #[arg(long, value_name = "YYYY-MM-DD")]
93 from: Option<NaiveDate>,
94
95 #[arg(long, value_name = "YYYY-MM-DD")]
97 to: Option<NaiveDate>,
98}
99
100#[derive(Debug, Args)]
102pub struct PulseArgs {
103 #[command(subcommand)]
104 command: Option<PulseCommand>,
105}
106
107#[derive(Debug, Subcommand)]
110enum PulseCommand {
111 #[command(about = "Start telling the server whether you are working, on a break, or not in a day")]
113 Enable,
114
115 #[command(about = "Stop telling the server whether you are working right now")]
117 Disable,
118}
119
120#[derive(Debug, Args)]
122pub struct PushArgs {
123 #[arg(long, short, help = "Send the last day instead of today")]
125 last: bool,
126
127 #[arg(long, value_name = "YYYY-MM-DD", conflicts_with = "last")]
129 date: Option<NaiveDate>,
130}
131
132#[derive(Debug, Args)]
134pub struct ConnectArgs {
135 #[arg(long, value_name = "URL")]
137 url: Option<String>,
138
139 #[arg(long, value_name = "PATH")]
141 ca_certificate: Option<String>,
142}
143
144pub async fn cmd(args: ServerArgs) -> Result<()> {
146 match args.command {
147 ServerCommand::Connect(args) => connect(args).await,
148 ServerCommand::Status => status().await,
149 ServerCommand::Push(args) => push(args).await,
150 ServerCommand::Flush => flush().await,
151 ServerCommand::Queue => queue(),
152 ServerCommand::Manifest => manifest().await,
153 ServerCommand::Backfill(args) => backfill(args).await,
154 ServerCommand::Pulse(args) => match args.command {
155 Some(PulseCommand::Enable) => pulse_enable().await,
156 Some(PulseCommand::Disable) => pulse_disable(),
157 None => pulse_show(),
158 },
159 ServerCommand::Disconnect => disconnect(),
160 }
161}
162
163async fn connect(args: ConnectArgs) -> Result<()> {
169 crate::libs::prompt::ensure_interactive("`kasl server connect` needs a terminal to ask for the agent token")?;
176
177 let mut config = Config::read().unwrap_or_default();
178
179 let url = match args.url {
180 Some(url) => normalize_url(&url),
181 None => {
182 let entered: String = Input::with_theme(&ColorfulTheme::default())
183 .with_prompt(Message::PromptKaslServerUrl.to_string())
184 .with_initial_text(config.kasl_server.as_ref().map(|s| s.url.clone()).unwrap_or_default())
185 .interact_text()?;
186 normalize_url(&entered)
187 }
188 };
189
190 if !url.starts_with("http://") && !url.starts_with("https://") {
193 return Err(msg_error_anyhow!(Message::KaslServerUrlNeedsScheme(url)));
194 }
195
196 let candidate = KaslServerConfig {
197 url: url.clone(),
198 ca_certificate: args
201 .ca_certificate
202 .or_else(|| config.kasl_server.as_ref().and_then(|s| s.ca_certificate.clone())),
203 pulse: keeps_pulse(config.kasl_server.as_ref(), &url),
204 };
205
206 let client = KaslServer::new(&candidate)?;
207
208 let health = client.health().await?;
211 msg_info!(Message::KaslServerReached {
212 url: url.clone(),
213 version: health.version.clone(),
214 });
215 if health.database != "ok" {
216 msg_warning!(Message::KaslServerDatabaseUnhealthy(health.database.clone()));
219 }
220
221 let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
222 let token: String = Password::with_theme(&ColorfulTheme::default())
223 .with_prompt(Message::PromptKaslServerToken.to_string())
224 .interact()?;
225 let token = token.trim().to_string();
226 if token.is_empty() {
227 return Err(msg_error_anyhow!(Message::KaslServerTokenEmpty));
228 }
229
230 let identity = client.identify(&token).await?;
232
233 secret
235 .store(&token)
236 .context("the token was accepted but could not be stored in the OS keyring")?;
237 config.kasl_server = Some(candidate);
238 config.save()?;
239
240 msg_success!(Message::KaslServerConnected {
241 user_name: identity.user_name,
242 agent_name: identity.agent_name,
243 });
244 Ok(())
245}
246
247async fn status() -> Result<()> {
253 let config = Config::read().unwrap_or_default();
254 let Some(server_config) = config.kasl_server else {
255 msg_print!(Message::KaslServerNotConnected);
256 return Ok(());
257 };
258
259 msg_info!(Message::KaslServerConfigured(server_config.url.clone()));
260 report_connection(&server_config).await?;
261
262 report_pulse(server_config.pulse)
266}
267
268async fn report_connection(server_config: &KaslServerConfig) -> Result<()> {
270 let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
271 let Some(token) = secret.try_get_cached() else {
272 msg_warning!(Message::KaslServerTokenMissing);
275 return Ok(());
276 };
277
278 let client = KaslServer::new(server_config)?;
279
280 match client.health().await {
281 Ok(health) => msg_info!(Message::KaslServerReached {
282 url: server_config.url.clone(),
283 version: health.version,
284 }),
285 Err(error) => {
286 msg_warning!(Message::KaslServerUnreachable(error.to_string()));
287 return Ok(());
288 }
289 }
290
291 match client.identify(&token).await {
292 Ok(identity) => {
293 msg_success!(Message::KaslServerConnected {
294 user_name: identity.user_name,
295 agent_name: identity.agent_name,
296 });
297 msg_info!(Message::KaslServerCompatibility {
304 server_version: identity.server_version,
305 api_version: identity.api_version,
306 });
307 }
308 Err(error) => msg_warning!(Message::KaslServerTokenRejected(error.to_string())),
309 }
310
311 Ok(())
312}
313
314fn report_pulse(enabled: bool) -> Result<()> {
316 let record = if enabled { ServerPulse::new()?.last()? } else { None };
317 for line in pulse::describe(enabled, record.as_ref(), Utc::now()) {
318 if line.warning {
319 msg_warning!(line.message);
320 } else {
321 msg_info!(line.message);
322 }
323 }
324 Ok(())
325}
326
327fn keeps_pulse(previous: Option<&KaslServerConfig>, url: &str) -> bool {
333 previous.is_some_and(|previous| previous.pulse && previous.url == url)
334}
335
336const FIRST_PULSE_WAIT: std::time::Duration = std::time::Duration::from_secs(12);
341
342async fn pulse_enable() -> Result<()> {
350 let mut config = Config::read().unwrap_or_default();
351 let Some(server) = config.kasl_server.as_mut() else {
352 return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
353 };
354 if Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).try_get_cached().is_none() {
357 return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
358 }
359
360 ServerPulse::new()?.clear()?;
361 server.pulse = true;
362 let url = server.url.clone();
363 config.save()?;
364 msg_success!(Message::PulseEnabled(url));
365
366 if !daemon::is_running() {
367 msg_warning!(Message::PulseNoWatcher);
368 return Ok(());
369 }
370
371 let deadline = std::time::Instant::now() + FIRST_PULSE_WAIT;
372 loop {
373 if let Some(record) = ServerPulse::new()?.last()? {
374 match (record.error, record.state) {
375 (None, Some(state)) => msg_success!(Message::PulseFirstArrived(state.to_string())),
376 (Some(error), _) => msg_warning!(Message::PulseFirstFailed(error)),
377 (None, None) => msg_info!(Message::PulseFirstPending),
378 }
379 return Ok(());
380 }
381 if std::time::Instant::now() >= deadline {
382 msg_info!(Message::PulseFirstPending);
383 return Ok(());
384 }
385 tokio::time::sleep(std::time::Duration::from_millis(250)).await;
386 }
387}
388
389fn pulse_disable() -> Result<()> {
396 let mut config = Config::read().unwrap_or_default();
397 let Some(server) = config.kasl_server.as_mut() else {
398 msg_print!(Message::KaslServerNotConnected);
399 return Ok(());
400 };
401 if !server.pulse {
402 msg_print!(Message::PulseAlreadyOff);
403 return Ok(());
404 }
405
406 server.pulse = false;
407 config.save()?;
408
409 let mut store = ServerPulse::new()?;
410 let stale_after = store.last()?.and_then(|record| record.stale_after_seconds).map(format_ago);
411 store.clear()?;
412
413 msg_success!(Message::PulseDisabled(stale_after));
414 Ok(())
415}
416
417fn pulse_show() -> Result<()> {
419 let config = Config::read().unwrap_or_default();
420 let Some(server) = config.kasl_server else {
421 msg_print!(Message::KaslServerNotConnected);
422 return Ok(());
423 };
424 report_pulse(server.pulse)
425}
426
427async fn manifest() -> Result<()> {
441 let (client, token) = connected_client()?;
442
443 let manifest = client.privacy(&token).await?;
444
445 msg_info!(Message::KaslServerPrivacyHeading(manifest.level));
446 msg_print!(Message::KaslServerPrivacySummary(manifest.summary));
447
448 msg_print!(Message::KaslServerPrivacyStoredHeading);
449 for stored in manifest.stored {
450 msg_print!(Message::KaslServerPrivacyStored {
451 what: stored.what,
452 detail: stored.detail,
453 });
454 }
455
456 msg_print!(Message::KaslServerPrivacyNeverHeading);
457 for line in manifest.never_collected {
458 msg_print!(Message::KaslServerPrivacyBullet(line));
459 }
460
461 msg_print!(Message::KaslServerPrivacyVisibleHeading);
462 for line in manifest.visible_to {
463 msg_print!(Message::KaslServerPrivacyBullet(line));
464 }
465
466 msg_print!(Message::KaslServerPrivacyRetention(manifest.retention));
467 msg_print!(Message::KaslServerPrivacyOnChange(manifest.on_change));
468
469 if let Some(updated_at) = manifest.updated_at {
473 msg_print!(Message::KaslServerPrivacyUpdatedAt(
474 updated_at.with_timezone(&Local).format("%Y-%m-%d %H:%M").to_string()
475 ));
476 }
477
478 msg_print!(Message::KaslServerPrivacySetByAdmin);
479
480 Ok(())
481}
482
483async fn push(args: PushArgs) -> Result<()> {
500 let date = match args.date {
501 Some(date) => date,
502 None if args.last => (Local::now() - Duration::days(1)).date_naive(),
503 None => Local::now().date_naive(),
504 };
505
506 let config = Config::read().unwrap_or_default();
507 let Some(server_config) = config.kasl_server else {
508 return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
509 };
510
511 let Some(day) = build_day_upload(date)? else {
512 msg_print!(Message::KaslServerNoDayToPush(date.to_string()));
513 return Ok(());
514 };
515
516 let secret = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT);
517 let Some(token) = secret.try_get_cached() else {
518 return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
519 };
520
521 let client = KaslServer::new(&server_config)?;
522
523 match client.upload_day(&token, &day).await {
524 Ok(accepted) => {
525 msg_success!(Message::KaslServerDayPushed {
526 date: accepted.date.to_string(),
527 pauses: accepted.pauses,
528 tasks: accepted.tasks,
529 });
530 if accepted.deleted_tasks > 0 {
534 msg_info!(Message::KaslServerTasksDeleted(accepted.deleted_tasks));
535 }
536
537 ServerOutbox::new()?.remove(date)?;
541
542 flush_with(&client, &token).await?;
547 Ok(())
548 }
549 Err(error) => {
555 let outcome = record_single(&mut ServerOutbox::new()?, date, &error)?;
559 if matches!(outcome, Delivered::Deferred { .. }) {
560 msg_info!(Message::KaslServerDayQueued(date.to_string()));
561 }
562
563 match error {
564 error @ UploadError::Rejected {
565 status: StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN,
566 ..
567 } => Err(msg_error_anyhow!(Message::KaslServerPushTokenRejected(error.to_string()))),
568 error @ UploadError::Rejected { .. } => Err(msg_error_anyhow!(Message::KaslServerPushRejected(error.to_string()))),
569 error => Err(msg_error_anyhow!(Message::KaslServerPushRetryable(error.to_string()))),
570 }
571 }
572 }
573}
574
575async fn flush() -> Result<()> {
577 if ServerOutbox::new()?.count()? == 0 {
583 msg_print!(Message::KaslServerQueueEmpty);
584 return Ok(());
585 }
586
587 let (client, token) = connected_client()?;
588 flush_with(&client, &token).await
589}
590
591async fn flush_with(client: &KaslServer, token: &str) -> Result<()> {
596 let mut outbox = ServerOutbox::new()?;
597 let dates: Vec<NaiveDate> = outbox.pending()?.into_iter().map(|owed| owed.date).collect();
598 if dates.is_empty() {
599 return Ok(());
600 }
601
602 msg_info!(Message::KaslServerQueueSending(dates.len()));
603
604 let outcomes = deliver(client, token, &mut outbox, &dates).await?;
605
606 let (mut accepted, mut refused, mut deferred) = (0, 0, 0);
607 for outcome in &outcomes {
608 match outcome {
609 Delivered::Accepted {
610 date,
611 pauses,
612 tasks,
613 deleted_tasks,
614 } => {
615 accepted += 1;
616 msg_success!(Message::KaslServerDayPushed {
617 date: date.to_string(),
618 pauses: *pauses,
619 tasks: *tasks,
620 });
621 if *deleted_tasks > 0 {
622 msg_info!(Message::KaslServerTasksDeleted(*deleted_tasks));
623 }
624 }
625 Delivered::Refused { date, reason } => {
630 refused += 1;
631 msg_warning!(Message::KaslServerDayRefused {
632 date: date.to_string(),
633 reason: reason.clone(),
634 });
635 }
636 Delivered::Deferred { date, reason } => {
637 deferred += 1;
638 msg_warning!(Message::KaslServerDayDeferred {
639 date: date.to_string(),
640 reason: reason.clone(),
641 });
642 }
643 }
644 }
645
646 msg_print!(Message::KaslServerFlushSummary { accepted, refused, deferred });
647 Ok(())
648}
649
650fn queue() -> Result<()> {
655 let outbox = ServerOutbox::new()?;
656 let owed = outbox.pending()?;
657
658 if owed.is_empty() {
659 msg_print!(Message::KaslServerQueueEmpty);
660 return Ok(());
661 }
662
663 msg_info!(Message::KaslServerQueueOwed(owed.len() as i64));
664 for day in &owed {
665 msg_print!(Message::KaslServerQueueEntry {
666 date: day.date.to_string(),
667 attempts: day.attempts,
668 last_error: day.last_error.clone(),
669 });
670 }
671
672 Ok(())
673}
674
675async fn backfill(args: BackfillArgs) -> Result<()> {
688 let to = args.to.unwrap_or_else(|| Local::now().date_naive());
689 if let Some(from) = args.from
690 && from > to
691 {
692 return Err(msg_error_anyhow!(Message::KaslServerBackfillOrderReversed));
693 }
694
695 let (client, token) = connected_client()?;
696
697 let dates = Workdays::new()?.recorded_dates(args.from, Some(to))?;
698
699 if dates.is_empty() {
700 match args.from {
701 Some(from) => msg_print!(Message::KaslServerBackfillNoDays {
702 from: from.to_string(),
703 to: to.to_string(),
704 }),
705 None => msg_print!(Message::KaslServerBackfillNothingRecorded),
711 }
712 return Ok(());
713 }
714
715 match args.from {
716 Some(from) => msg_info!(Message::KaslServerBackfillRange {
717 from: from.to_string(),
718 to: to.to_string(),
719 days: dates.len(),
720 }),
721 None => msg_info!(Message::KaslServerBackfillWholeHistory {
726 from: dates[0].to_string(),
727 to: to.to_string(),
728 days: dates.len(),
729 }),
730 }
731
732 let mut outbox = ServerOutbox::new()?;
735 for date in &dates {
736 outbox.enqueue(*date, "queued by backfill")?;
737 }
738
739 flush_with(&client, &token).await
740}
741
742fn connected_client() -> Result<(KaslServer, String)> {
748 let config = Config::read().unwrap_or_default();
749 let Some(server_config) = config.kasl_server else {
750 return Err(msg_error_anyhow!(Message::KaslServerNotConnected));
751 };
752
753 let Some(token) = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).try_get_cached() else {
754 return Err(msg_error_anyhow!(Message::KaslServerTokenMissing));
755 };
756
757 Ok((KaslServer::new(&server_config)?, token))
758}
759
760fn disconnect() -> Result<()> {
766 let mut config = Config::read().unwrap_or_default();
767
768 if let Err(error) = Secret::new(AGENT_TOKEN_SECRET, AGENT_TOKEN_PROMPT).delete() {
775 msg_warning!(Message::KaslServerTokenNotRemoved(error.to_string()));
776 }
777
778 if config.kasl_server.take().is_some() {
779 config.save()?;
780 ServerPulse::new()?.clear()?;
783 msg_success!(Message::KaslServerDisconnected);
784 } else {
785 msg_print!(Message::KaslServerNotConnected);
787 }
788
789 Ok(())
790}
791
792#[cfg(test)]
793mod tests {
794 use super::*;
795
796 fn connection(url: &str, pulse: bool) -> KaslServerConfig {
797 KaslServerConfig {
798 url: url.to_string(),
799 ca_certificate: None,
800 pulse,
801 }
802 }
803
804 #[test]
805 fn reconnecting_to_the_same_server_keeps_the_pulse() {
806 assert!(keeps_pulse(Some(&connection("https://kasl.example.com", true)), "https://kasl.example.com"));
807 }
808
809 #[test]
810 fn another_server_does_not_inherit_the_consent() {
811 assert!(!keeps_pulse(Some(&connection("https://kasl.example.com", true)), "https://kasl.other.example"));
814 assert!(!keeps_pulse(Some(&connection("https://kasl.example.com", false)), "https://kasl.example.com"));
815 assert!(!keeps_pulse(None, "https://kasl.example.com"));
816 }
817}