plausible_cli/cli/
mod.rs

1use crate::{
2    client::{
3        AggregateQuery, BreakdownQuery, BreakdownResponse, CreateSiteRequest, PlausibleClient,
4        RealtimeVisitorsResponse, ResetSiteStatsRequest, SiteSummary, TimeseriesQuery,
5        TimeseriesResponse, UpdateSiteRequest,
6    },
7    config::accounts::{
8        AccountExport, AccountProfile, AccountRecord, AccountStore, AccountSummary,
9    },
10    queue::{
11        self, JobKind, JobRequest, JobResponse, QueueJobState, QueueSnapshot, Worker, WorkerError,
12    },
13    rate_limit::{RateLimitConfig, RateLimiter},
14    Error,
15};
16use async_trait::async_trait;
17use clap::{Args, Parser, Subcommand, ValueEnum};
18use std::fs;
19use std::io::{self, Read};
20use std::num::NonZeroU32;
21use std::path::PathBuf;
22use std::sync::Arc;
23use tabled::{Table, Tabled};
24use time::{format_description::well_known::Rfc3339, OffsetDateTime};
25
26use crate::config::ConfigPaths;
27#[derive(Parser, Debug)]
28#[command(name = "plausible", version = env!("CARGO_PKG_VERSION"), arg_required_else_help = true)]
29pub struct Cli {
30    /// Override the account alias to use for this invocation.
31    #[arg(long, short = 'A', global = true)]
32    pub account: Option<String>,
33
34    /// Output format for command results.
35    #[arg(long, value_enum, global = true, default_value_t = OutputFormat::Human)]
36    pub output: OutputFormat,
37
38    /// Custom Plausible base URL (needed for self-hosted instances).
39    #[arg(long, global = true)]
40    pub base_url: Option<String>,
41
42    #[command(subcommand)]
43    pub command: Commands,
44}
45
46#[derive(Subcommand, Debug, Clone)]
47pub enum Commands {
48    /// Display current rate-limit usage.
49    Status,
50    /// Operations for Plausible sites.
51    Sites {
52        #[command(subcommand)]
53        command: SitesCommand,
54    },
55    /// Fetch Plausible statistics.
56    Stats {
57        #[command(subcommand)]
58        command: StatsCommand,
59    },
60    /// Manage or send custom events.
61    Events {
62        #[command(subcommand)]
63        command: EventsCommand,
64    },
65    /// Inspect and control the background queue.
66    Queue {
67        #[command(subcommand)]
68        command: QueueCommand,
69    },
70    /// Manage stored Plausible accounts.
71    Accounts {
72        #[command(subcommand)]
73        command: AccountsCommand,
74    },
75}
76
77#[derive(Subcommand, Debug, Clone)]
78pub enum SitesCommand {
79    /// List all sites visible to the current account.
80    List,
81    /// Create a new Plausible site.
82    Create(SiteCreateArgs),
83    /// Update properties on an existing site.
84    Update(SiteUpdateArgs),
85    /// Reset statistics for a site.
86    Reset(SiteResetArgs),
87    /// Delete a site.
88    Delete(SiteDeleteArgs),
89}
90
91#[derive(Subcommand, Debug, Clone)]
92pub enum StatsCommand {
93    /// Run an aggregate stats query.
94    Aggregate(StatsAggregateArgs),
95    /// Run a timeseries stats query.
96    Timeseries(StatsTimeseriesArgs),
97    /// Run a breakdown stats query.
98    Breakdown(StatsBreakdownArgs),
99    /// Fetch realtime visitors.
100    Realtime(StatsRealtimeArgs),
101}
102
103#[derive(Subcommand, Debug, Clone)]
104pub enum EventsCommand {
105    /// Print an example events payload.
106    Template,
107    /// Send a single custom event.
108    Send(EventSendArgs),
109    /// Import newline-delimited events.
110    Import(EventImportArgs),
111}
112
113#[derive(Subcommand, Debug, Clone)]
114pub enum QueueCommand {
115    /// Show pending and in-flight jobs.
116    Inspect,
117    /// Block until the queue is empty.
118    Drain,
119}
120
121#[derive(Subcommand, Debug, Clone)]
122pub enum AccountsCommand {
123    /// List configured accounts.
124    List,
125    /// Add a new account alias.
126    Add(AddAccountArgs),
127    /// Set an existing account as default.
128    Use { alias: String },
129    /// Remove an account alias and its credentials.
130    Remove { alias: String },
131    /// Export account metadata without secrets.
132    Export {
133        #[arg(long)]
134        json: bool,
135    },
136    /// Configure account-specific rate limit budgets.
137    Budget(SetBudgetArgs),
138}
139
140#[derive(Args, Debug, Clone, Default)]
141pub struct AddAccountArgs {
142    /// Alias used to reference this account.
143    #[arg(long)]
144    pub alias: String,
145    /// API key associated with the Plausible account.
146    #[arg(long, env = "PLAUSIBLE_API_KEY")]
147    pub api_key: String,
148    /// Optional friendly label.
149    #[arg(long)]
150    pub label: Option<String>,
151    /// Optional contact email.
152    #[arg(long)]
153    pub email: Option<String>,
154    /// Optional description for humans/LLMs.
155    #[arg(long)]
156    pub description: Option<String>,
157}
158
159#[derive(Args, Debug, Clone, Default)]
160pub struct SetBudgetArgs {
161    /// Account alias to configure.
162    #[arg(long)]
163    pub alias: String,
164    /// Daily request budget (omit or set to zero to clear).
165    #[arg(long)]
166    pub daily: Option<u32>,
167    /// Clear any configured daily budget.
168    #[arg(long)]
169    pub clear: bool,
170}
171
172#[derive(Args, Debug, Default, Clone)]
173pub struct StatsAggregateArgs {
174    /// Site domain or site_id to query.
175    #[arg(long)]
176    pub site: String,
177    /// Metrics to include (repeatable).
178    #[arg(long = "metric", short = 'm')]
179    pub metrics: Vec<String>,
180    /// Period string (e.g., 7d, 30d, month, custom).
181    #[arg(long)]
182    pub period: Option<String>,
183    /// Custom date range (YYYY-MM-DD,YYYY-MM-DD).
184    #[arg(long)]
185    pub date: Option<String>,
186    /// Filters to apply (repeatable, Plausible filter syntax).
187    #[arg(long)]
188    pub filters: Vec<String>,
189    /// Event properties to include (repeatable).
190    #[arg(long)]
191    pub properties: Vec<String>,
192    /// Compare parameter (e.g., previous_period).
193    #[arg(long)]
194    pub compare: Option<String>,
195    /// Interval parameter (e.g., date, week).
196    #[arg(long)]
197    pub interval: Option<String>,
198    /// Sort parameter for breakdowns.
199    #[arg(long)]
200    pub sort: Option<String>,
201    /// Limit number of rows returned.
202    #[arg(long)]
203    pub limit: Option<u32>,
204    /// Page number (for paginated queries).
205    #[arg(long)]
206    pub page: Option<u32>,
207}
208
209#[derive(Args, Debug, Default, Clone)]
210pub struct SiteCreateArgs {
211    /// Domain for the new site.
212    #[arg(long)]
213    pub domain: String,
214    /// Timezone identifier.
215    #[arg(long)]
216    pub timezone: Option<String>,
217    /// Whether the site should be public (provide true/false).
218    #[arg(long)]
219    pub public: Option<bool>,
220}
221
222#[derive(Args, Debug, Default, Clone)]
223pub struct SiteUpdateArgs {
224    /// Site domain or site_id to update.
225    #[arg(long)]
226    pub site: String,
227    /// Update the timezone.
228    #[arg(long)]
229    pub timezone: Option<String>,
230    /// Update the public flag (true/false).
231    #[arg(long)]
232    pub public: Option<bool>,
233    /// Toggle whether this is the main site (true/false).
234    #[arg(long = "main-site")]
235    pub main_site: Option<bool>,
236}
237
238#[derive(Args, Debug, Default, Clone)]
239pub struct SiteResetArgs {
240    /// Site domain or site_id to reset.
241    #[arg(long)]
242    pub site: String,
243    /// Optional cut-off date (YYYY-MM-DD).
244    #[arg(long)]
245    pub date: Option<String>,
246}
247
248#[derive(Args, Debug, Clone, Default)]
249pub struct SiteDeleteArgs {
250    /// Site domain or site_id to delete.
251    #[arg(long)]
252    pub site: String,
253    /// Skip interactive confirmation.
254    #[arg(long)]
255    pub force: bool,
256}
257
258#[derive(Args, Debug, Default, Clone)]
259pub struct StatsTimeseriesArgs {
260    /// Site domain or site_id to query.
261    #[arg(long)]
262    pub site: String,
263    /// Metrics to include (repeatable).
264    #[arg(long = "metric", short = 'm')]
265    pub metrics: Vec<String>,
266    /// Period string (e.g., 7d, 30d, month, custom).
267    #[arg(long)]
268    pub period: Option<String>,
269    /// Custom date range (YYYY-MM-DD,YYYY-MM-DD).
270    #[arg(long)]
271    pub date: Option<String>,
272    /// Filters to apply (repeatable, Plausible filter syntax).
273    #[arg(long)]
274    pub filters: Vec<String>,
275    /// Event properties to include (repeatable).
276    #[arg(long)]
277    pub properties: Vec<String>,
278    /// Compare parameter (e.g., previous_period).
279    #[arg(long)]
280    pub compare: Option<String>,
281    /// Interval parameter (e.g., date, week).
282    #[arg(long)]
283    pub interval: Option<String>,
284    /// Sort parameter for breakdowns.
285    #[arg(long)]
286    pub sort: Option<String>,
287    /// Limit number of rows returned.
288    #[arg(long)]
289    pub limit: Option<u32>,
290    /// Page number (for paginated queries).
291    #[arg(long)]
292    pub page: Option<u32>,
293}
294
295#[derive(Args, Debug, Default, Clone)]
296pub struct StatsBreakdownArgs {
297    /// Site domain or site_id to query.
298    #[arg(long)]
299    pub site: String,
300    /// Property to breakdown by (e.g., event:page, visit:source).
301    #[arg(long)]
302    pub property: String,
303    /// Metrics to include (repeatable).
304    #[arg(long = "metric", short = 'm')]
305    pub metrics: Vec<String>,
306    /// Period string (e.g., 7d, 30d, month, custom).
307    #[arg(long)]
308    pub period: Option<String>,
309    /// Custom date range (YYYY-MM-DD,YYYY-MM-DD).
310    #[arg(long)]
311    pub date: Option<String>,
312    /// Filters to apply (repeatable, Plausible filter syntax).
313    #[arg(long)]
314    pub filters: Vec<String>,
315    /// Event properties to include (repeatable).
316    #[arg(long)]
317    pub properties: Vec<String>,
318    /// Compare parameter (e.g., previous_period).
319    #[arg(long)]
320    pub compare: Option<String>,
321    /// Sort parameter for breakdowns.
322    #[arg(long)]
323    pub sort: Option<String>,
324    /// Limit number of rows returned.
325    #[arg(long)]
326    pub limit: Option<u32>,
327    /// Page number (for paginated queries).
328    #[arg(long)]
329    pub page: Option<u32>,
330    /// Include parameter (e.g., previous).
331    #[arg(long)]
332    pub include: Option<String>,
333}
334
335#[derive(Args, Debug, Default, Clone)]
336pub struct StatsRealtimeArgs {
337    /// Site domain or site_id to query.
338    #[arg(long)]
339    pub site: String,
340}
341
342#[derive(Args, Debug, Clone)]
343pub struct EventSendArgs {
344    /// Provide a JSON string payload inline.
345    #[arg(long)]
346    pub data: Option<String>,
347    /// Path to a file containing JSON payload.
348    #[arg(long)]
349    pub file: Option<PathBuf>,
350    /// Read payload from STDIN (JSON).
351    #[arg(long)]
352    pub stdin: bool,
353    /// Override the `domain` field in the payload.
354    #[arg(long)]
355    pub domain: Option<String>,
356}
357
358#[derive(Args, Debug, Clone)]
359pub struct EventImportArgs {
360    /// Path to a newline-delimited JSON file.
361    #[arg(long)]
362    pub file: Option<PathBuf>,
363    /// Read newline-delimited JSON from STDIN.
364    #[arg(long)]
365    pub stdin: bool,
366    /// Print summary without sending to Plausible.
367    #[arg(long)]
368    pub dry_run: bool,
369    /// Override the `domain` field for each event.
370    #[arg(long)]
371    pub domain: Option<String>,
372}
373
374#[derive(Copy, Clone, Debug, ValueEnum, Default)]
375pub enum OutputFormat {
376    #[default]
377    Human,
378    Json,
379}
380
381pub async fn execute(cli: Cli) -> Result<(), Error> {
382    let paths = ConfigPaths::with_project_dirs()?;
383    let account_store = AccountStore::new(paths.clone())?;
384
385    if let Commands::Accounts { command } = &cli.command {
386        handle_account_command(&account_store, command, cli.output)?;
387        return Ok(());
388    }
389
390    let account_alias = resolve_account(&account_store, &cli.account)?;
391    let account = account_store.get_account(&account_alias)?;
392    let client = build_client(&cli, &account)?;
393    let rate_config = RateLimitConfig::default().with_daily_quota(account.daily_budget);
394    let rate_limiter = RateLimiter::new(paths.clone(), &account_alias, rate_config).await?;
395
396    let executor = Arc::new(PlausibleExecutor::new(client));
397    let queue = Worker::spawn(executor, rate_limiter.clone(), None);
398
399    match &cli.command {
400        Commands::Status => {
401            render_status(&rate_limiter, cli.output).await?;
402        }
403        Commands::Sites { command } => match command {
404            SitesCommand::List => {
405                let ticket = queue
406                    .submit(
407                        JobRequest {
408                            account: account_alias.clone(),
409                            kind: JobKind::ListSites,
410                            max_retries: queue::DEFAULT_MAX_RETRIES,
411                        },
412                        NonZeroU32::new(1).unwrap(),
413                    )
414                    .await?;
415                let response = ticket.await_result().await?;
416                if let JobResponse::Sites(sites) = response {
417                    render_sites(&sites, cli.output)?;
418                }
419            }
420            SitesCommand::Create(args) => {
421                let request = build_create_site_request(args);
422                let ticket = queue
423                    .submit(
424                        JobRequest {
425                            account: account_alias.clone(),
426                            kind: JobKind::SiteCreate {
427                                request: Box::new(request),
428                            },
429                            max_retries: queue::DEFAULT_MAX_RETRIES,
430                        },
431                        NonZeroU32::new(1).unwrap(),
432                    )
433                    .await?;
434                let response = ticket.await_result().await?;
435                if let JobResponse::SiteCreated(site) = response {
436                    render_site_summary(&site, cli.output)?;
437                }
438            }
439            SitesCommand::Update(args) => {
440                let (site_id, request) = build_update_site_request(args);
441                let ticket = queue
442                    .submit(
443                        JobRequest {
444                            account: account_alias.clone(),
445                            kind: JobKind::SiteUpdate {
446                                site_id,
447                                request: Box::new(request),
448                            },
449                            max_retries: queue::DEFAULT_MAX_RETRIES,
450                        },
451                        NonZeroU32::new(1).unwrap(),
452                    )
453                    .await?;
454                let response = ticket.await_result().await?;
455                if let JobResponse::SiteUpdated(site) = response {
456                    render_site_summary(&site, cli.output)?;
457                }
458            }
459            SitesCommand::Reset(args) => {
460                let (site_id, request) = build_reset_site_request(args);
461                let ticket = queue
462                    .submit(
463                        JobRequest {
464                            account: account_alias.clone(),
465                            kind: JobKind::SiteReset {
466                                site_id,
467                                request: Box::new(request),
468                            },
469                            max_retries: queue::DEFAULT_MAX_RETRIES,
470                        },
471                        NonZeroU32::new(1).unwrap(),
472                    )
473                    .await?;
474                let response = ticket.await_result().await?;
475                if matches!(response, JobResponse::SiteReset | JobResponse::Acknowledged) {
476                    println!("Site statistics reset queued successfully.");
477                }
478            }
479            SitesCommand::Delete(args) => {
480                if !args.force && !confirm_deletion(&args.site)? {
481                    println!("Aborted.");
482                } else {
483                    let ticket = queue
484                        .submit(
485                            JobRequest {
486                                account: account_alias.clone(),
487                                kind: JobKind::SiteDelete {
488                                    site_id: args.site.clone(),
489                                },
490                                max_retries: queue::DEFAULT_MAX_RETRIES,
491                            },
492                            NonZeroU32::new(1).unwrap(),
493                        )
494                        .await?;
495                    let response = ticket.await_result().await?;
496                    if matches!(
497                        response,
498                        JobResponse::SiteDeleted | JobResponse::Acknowledged
499                    ) {
500                        println!("Site '{}' deleted.", args.site);
501                    }
502                }
503            }
504        },
505        Commands::Stats { command } => match command {
506            StatsCommand::Aggregate(args) => {
507                let query = build_aggregate_query(args);
508                let ticket = queue
509                    .submit(
510                        JobRequest {
511                            account: account_alias.clone(),
512                            kind: JobKind::StatsAggregate {
513                                query: Box::new(query),
514                            },
515                            max_retries: queue::DEFAULT_MAX_RETRIES,
516                        },
517                        NonZeroU32::new(1).unwrap(),
518                    )
519                    .await?;
520                let response = ticket.await_result().await?;
521                if let JobResponse::StatsAggregate(result) = response {
522                    render_aggregate(&result, cli.output)?;
523                }
524            }
525            StatsCommand::Timeseries(args) => {
526                let query = build_timeseries_query(args);
527                let ticket = queue
528                    .submit(
529                        JobRequest {
530                            account: account_alias.clone(),
531                            kind: JobKind::StatsTimeseries {
532                                query: Box::new(query),
533                            },
534                            max_retries: queue::DEFAULT_MAX_RETRIES,
535                        },
536                        NonZeroU32::new(1).unwrap(),
537                    )
538                    .await?;
539                let response = ticket.await_result().await?;
540                if let JobResponse::StatsTimeseries(result) = response {
541                    render_timeseries(&result, cli.output)?;
542                }
543            }
544            StatsCommand::Breakdown(args) => {
545                let query = build_breakdown_query(args);
546                let ticket = queue
547                    .submit(
548                        JobRequest {
549                            account: account_alias.clone(),
550                            kind: JobKind::StatsBreakdown {
551                                query: Box::new(query),
552                            },
553                            max_retries: queue::DEFAULT_MAX_RETRIES,
554                        },
555                        NonZeroU32::new(1).unwrap(),
556                    )
557                    .await?;
558                let response = ticket.await_result().await?;
559                if let JobResponse::StatsBreakdown(result) = response {
560                    render_breakdown(&result, cli.output)?;
561                }
562            }
563            StatsCommand::Realtime(args) => {
564                let ticket = queue
565                    .submit(
566                        JobRequest {
567                            account: account_alias.clone(),
568                            kind: JobKind::StatsRealtime {
569                                site_id: args.site.clone(),
570                            },
571                            max_retries: queue::DEFAULT_MAX_RETRIES,
572                        },
573                        NonZeroU32::new(1).unwrap(),
574                    )
575                    .await?;
576                let response = ticket.await_result().await?;
577                if let JobResponse::StatsRealtime(result) = response {
578                    render_realtime(&result, cli.output)?;
579                }
580            }
581        },
582        Commands::Events { command } => match command {
583            EventsCommand::Template => {
584                render_event_template(cli.output)?;
585            }
586            EventsCommand::Send(args) => {
587                let event = load_event_payload(args)?;
588                let ticket = queue
589                    .submit(
590                        JobRequest {
591                            account: account_alias.clone(),
592                            kind: JobKind::EventSend { event },
593                            max_retries: queue::DEFAULT_MAX_RETRIES,
594                        },
595                        NonZeroU32::new(1).unwrap(),
596                    )
597                    .await?;
598                let response = ticket.await_result().await?;
599                match response {
600                    JobResponse::EventAck | JobResponse::Acknowledged => {
601                        println!("Event dispatched to Plausible.");
602                    }
603                    JobResponse::EventsProcessed { processed } => {
604                        println!("Processed batch containing {processed} events.");
605                    }
606                    JobResponse::Custom(value) => {
607                        println!("{}", serde_json::to_string_pretty(&value)?);
608                    }
609                    other => {
610                        println!("Event response: {:?}", other);
611                    }
612                }
613            }
614            EventsCommand::Import(args) => {
615                let events = load_import_payload(args)?;
616                if args.dry_run {
617                    let count = events.len();
618                    match cli.output {
619                        OutputFormat::Human => {
620                            println!("Dry run: would import {count} events.");
621                        }
622                        OutputFormat::Json => {
623                            let value = serde_json::json!({
624                                "dry_run": true,
625                                "count": count,
626                            });
627                            println!("{}", serde_json::to_string_pretty(&value)?);
628                        }
629                    }
630                } else {
631                    let count = events.len();
632                    let weight_u32 = u32::try_from(count).map_err(|_| {
633                        Error::InvalidInput("too many events for single import batch".into())
634                    })?;
635                    let ticket = queue
636                        .submit(
637                            JobRequest {
638                                account: account_alias.clone(),
639                                kind: JobKind::EventsImport { events },
640                                max_retries: queue::DEFAULT_MAX_RETRIES,
641                            },
642                            NonZeroU32::new(weight_u32).unwrap(),
643                        )
644                        .await?;
645                    let response = ticket.await_result().await?;
646                    match response {
647                        JobResponse::EventsProcessed { processed } => match cli.output {
648                            OutputFormat::Human => {
649                                println!("Imported {processed} events.");
650                            }
651                            OutputFormat::Json => {
652                                let value = serde_json::json!({ "processed": processed });
653                                println!("{}", serde_json::to_string_pretty(&value)?);
654                            }
655                        },
656                        JobResponse::EventAck | JobResponse::Acknowledged => {
657                            println!("Import completed.");
658                        }
659                        JobResponse::Custom(value) => {
660                            println!("{}", serde_json::to_string_pretty(&value)?);
661                        }
662                        other => println!("Import response: {:?}", other),
663                    }
664                }
665            }
666        },
667        Commands::Queue { command } => match command {
668            QueueCommand::Inspect => {
669                let snapshot = queue.snapshot().await;
670                render_queue_snapshot(&snapshot, cli.output)?;
671            }
672            QueueCommand::Drain => {
673                queue.wait_idle().await;
674                match cli.output {
675                    OutputFormat::Human => println!("Queue drained."),
676                    OutputFormat::Json => {
677                        let value = serde_json::json!({ "drained": true });
678                        println!("{}", serde_json::to_string_pretty(&value)?);
679                    }
680                }
681            }
682        },
683        Commands::Accounts { .. } => unreachable!(),
684    }
685
686    Ok(())
687}
688
689fn build_client(cli: &Cli, account: &AccountRecord) -> Result<PlausibleClient, Error> {
690    if let Some(base) = &cli.base_url {
691        let url = url::Url::parse(base).map_err(crate::client::ClientError::InvalidBaseUrl)?;
692        Ok(PlausibleClient::with_base_url(
693            account.api_key.clone(),
694            url,
695        )?)
696    } else {
697        Ok(PlausibleClient::new(account.api_key.clone())?)
698    }
699}
700
701fn resolve_account(store: &AccountStore, override_alias: &Option<String>) -> Result<String, Error> {
702    if let Some(alias) = override_alias {
703        return Ok(alias.clone());
704    }
705    match store.default_alias()? {
706        Some(alias) => Ok(alias),
707        None => Err(Error::NoDefaultAccount),
708    }
709}
710
711fn build_aggregate_query(args: &StatsAggregateArgs) -> AggregateQuery {
712    AggregateQuery {
713        site_id: args.site.clone(),
714        metrics: args.metrics.clone(),
715        period: args.period.clone(),
716        date: args.date.clone(),
717        filters: args.filters.clone(),
718        properties: args.properties.clone(),
719        compare: args.compare.clone(),
720        interval: args.interval.clone(),
721        sort: args.sort.clone(),
722        limit: args.limit,
723        page: args.page,
724    }
725}
726
727fn build_timeseries_query(args: &StatsTimeseriesArgs) -> TimeseriesQuery {
728    TimeseriesQuery {
729        site_id: args.site.clone(),
730        metrics: args.metrics.clone(),
731        period: args.period.clone(),
732        date: args.date.clone(),
733        interval: args.interval.clone(),
734        filters: args.filters.clone(),
735        properties: args.properties.clone(),
736        compare: args.compare.clone(),
737        sort: args.sort.clone(),
738        limit: args.limit,
739        page: args.page,
740    }
741}
742
743fn build_breakdown_query(args: &StatsBreakdownArgs) -> BreakdownQuery {
744    BreakdownQuery {
745        site_id: args.site.clone(),
746        property: args.property.clone(),
747        metrics: args.metrics.clone(),
748        period: args.period.clone(),
749        date: args.date.clone(),
750        filters: args.filters.clone(),
751        properties: args.properties.clone(),
752        compare: args.compare.clone(),
753        sort: args.sort.clone(),
754        limit: args.limit,
755        page: args.page,
756        include: args.include.clone(),
757    }
758}
759
760fn build_create_site_request(args: &SiteCreateArgs) -> CreateSiteRequest {
761    CreateSiteRequest {
762        domain: args.domain.clone(),
763        timezone: args.timezone.clone(),
764        public: args.public,
765    }
766}
767
768fn build_update_site_request(args: &SiteUpdateArgs) -> (String, UpdateSiteRequest) {
769    (
770        args.site.clone(),
771        UpdateSiteRequest {
772            timezone: args.timezone.clone(),
773            public: args.public,
774            main_site: args.main_site,
775        },
776    )
777}
778
779fn build_reset_site_request(args: &SiteResetArgs) -> (String, ResetSiteStatsRequest) {
780    (
781        args.site.clone(),
782        ResetSiteStatsRequest {
783            date: args.date.clone(),
784        },
785    )
786}
787
788fn load_event_payload(args: &EventSendArgs) -> Result<serde_json::Value, Error> {
789    let sources = args.data.is_some() as u8 + args.file.is_some() as u8 + args.stdin as u8;
790    if sources == 0 {
791        return Err(Error::InvalidInput(
792            "provide one of --data, --file, or --stdin for events send".into(),
793        ));
794    }
795    if sources > 1 {
796        return Err(Error::InvalidInput(
797            "choose only one of --data, --file, or --stdin for events send".into(),
798        ));
799    }
800    let raw = if let Some(data) = &args.data {
801        data.clone()
802    } else if let Some(path) = &args.file {
803        fs::read_to_string(path)?
804    } else {
805        let mut buf = String::new();
806        io::stdin().read_to_string(&mut buf)?;
807        buf
808    };
809    parse_event_json(&raw, &args.domain)
810}
811
812fn load_import_payload(args: &EventImportArgs) -> Result<Vec<serde_json::Value>, Error> {
813    let sources = args.file.is_some() as u8 + args.stdin as u8;
814    if sources == 0 {
815        return Err(Error::InvalidInput(
816            "provide --file or --stdin for events import".into(),
817        ));
818    }
819    if sources > 1 {
820        return Err(Error::InvalidInput(
821            "choose either --file or --stdin for events import".into(),
822        ));
823    }
824    let raw = if let Some(path) = &args.file {
825        fs::read_to_string(path)?
826    } else {
827        let mut buf = String::new();
828        io::stdin().read_to_string(&mut buf)?;
829        buf
830    };
831    parse_ndjson(&raw, &args.domain)
832}
833
834fn parse_event_json(contents: &str, domain: &Option<String>) -> Result<serde_json::Value, Error> {
835    let trimmed = contents.trim();
836    if trimmed.is_empty() {
837        return Err(Error::InvalidInput("event payload is empty".into()));
838    }
839    let mut value: serde_json::Value = serde_json::from_str(trimmed)?;
840    if !value.is_object() {
841        return Err(Error::InvalidInput(
842            "event payload must be a JSON object".into(),
843        ));
844    }
845    apply_domain_override(&mut value, domain);
846    Ok(value)
847}
848
849fn parse_ndjson(contents: &str, domain: &Option<String>) -> Result<Vec<serde_json::Value>, Error> {
850    let mut events = Vec::new();
851    for (idx, line) in contents.lines().enumerate() {
852        let trimmed = line.trim();
853        if trimmed.is_empty() {
854            continue;
855        }
856        let mut value: serde_json::Value = serde_json::from_str(trimmed).map_err(|err| {
857            Error::InvalidInput(format!("failed to parse line {}: {}", idx + 1, err))
858        })?;
859        if !value.is_object() {
860            return Err(Error::InvalidInput(format!(
861                "line {} must be a JSON object",
862                idx + 1
863            )));
864        }
865        apply_domain_override(&mut value, domain);
866        events.push(value);
867    }
868    if events.is_empty() {
869        return Err(Error::InvalidInput("no events found in input".into()));
870    }
871    Ok(events)
872}
873
874fn apply_domain_override(value: &mut serde_json::Value, domain: &Option<String>) {
875    if let (Some(domain), serde_json::Value::Object(map)) = (domain, value) {
876        map.insert("domain".into(), serde_json::Value::String(domain.clone()));
877    }
878}
879
880async fn render_status(rate_limiter: &RateLimiter, format: OutputFormat) -> Result<(), Error> {
881    let status = rate_limiter.status(OffsetDateTime::now_utc()).await?;
882    match format {
883        OutputFormat::Human => {
884            println!(
885                "Hourly usage: {}/{} (resets at {})",
886                status.hourly_used, status.hourly_limit, status.hourly_reset_at
887            );
888            if let Some(limit) = status.daily_limit {
889                let used = status.daily_used.unwrap_or(0);
890                let remaining = status
891                    .daily_remaining
892                    .unwrap_or_else(|| limit.saturating_sub(used));
893                if let Some(reset) = status.daily_reset_at {
894                    println!(
895                        "Daily usage: {}/{limit} (remaining {remaining}) – resets at {}",
896                        used, reset
897                    );
898                } else {
899                    println!("Daily usage: {}/{limit} (remaining {remaining})", used);
900                }
901            }
902        }
903        OutputFormat::Json => {
904            let value = serde_json::json!({
905                "hourly": {
906                    "used": status.hourly_used,
907                    "limit": status.hourly_limit,
908                    "remaining": status.hourly_remaining,
909                    "reset_at": status.hourly_reset_at,
910                },
911                "daily": {
912                    "used": status.daily_used,
913                    "limit": status.daily_limit,
914                    "remaining": status.daily_remaining,
915                    "reset_at": status.daily_reset_at,
916                }
917            });
918            println!("{}", serde_json::to_string_pretty(&value)?);
919        }
920    }
921    Ok(())
922}
923
924fn render_sites(sites: &[SiteSummary], format: OutputFormat) -> Result<(), Error> {
925    match format {
926        OutputFormat::Human => {
927            let rows: Vec<_> = sites.iter().map(SiteRow::from).collect();
928            let table = Table::new(rows).to_string();
929            println!("{}", table);
930        }
931        OutputFormat::Json => {
932            println!("{}", serde_json::to_string_pretty(&sites)?);
933        }
934    }
935    Ok(())
936}
937
938fn render_site_summary(site: &SiteSummary, format: OutputFormat) -> Result<(), Error> {
939    match format {
940        OutputFormat::Human => {
941            let table = Table::new(vec![SiteRow::from(site)]).to_string();
942            println!("{}", table);
943        }
944        OutputFormat::Json => {
945            println!("{}", serde_json::to_string_pretty(site)?);
946        }
947    }
948    Ok(())
949}
950
951fn render_aggregate(
952    response: &crate::client::AggregateResponse,
953    format: OutputFormat,
954) -> Result<(), Error> {
955    match format {
956        OutputFormat::Human => {
957            let mut rows = Vec::new();
958            for (metric, value) in &response.results {
959                let formatted = match value {
960                    serde_json::Value::Number(num) => num.to_string(),
961                    serde_json::Value::String(s) => s.clone(),
962                    other => other.to_string(),
963                };
964                rows.push(MetricRow {
965                    metric: metric.clone(),
966                    value: formatted,
967                });
968            }
969            let table = Table::new(rows).to_string();
970            println!("{}", table);
971        }
972        OutputFormat::Json => {
973            println!("{}", serde_json::to_string_pretty(&response)?);
974        }
975    }
976    Ok(())
977}
978
979fn render_timeseries(response: &TimeseriesResponse, format: OutputFormat) -> Result<(), Error> {
980    match format {
981        OutputFormat::Human => {
982            let rows: Vec<_> = response.results.iter().map(TimeseriesRow::from).collect();
983            if rows.is_empty() {
984                println!("No timeseries data.");
985            } else {
986                let table = Table::new(rows).to_string();
987                println!("{}", table);
988            }
989            if !response.totals.is_empty() {
990                println!("Totals: {}", format_metrics(&response.totals, &[]));
991            }
992        }
993        OutputFormat::Json => {
994            println!("{}", serde_json::to_string_pretty(&response)?);
995        }
996    }
997    Ok(())
998}
999
1000fn render_breakdown(response: &BreakdownResponse, format: OutputFormat) -> Result<(), Error> {
1001    match format {
1002        OutputFormat::Human => {
1003            let rows: Vec<_> = response.results.iter().map(BreakdownRow::from).collect();
1004            if let Some(page) = response.page {
1005                if let Some(total) = response.total_pages {
1006                    println!("Page {}/{}", page, total);
1007                } else {
1008                    println!("Page {}", page);
1009                }
1010            }
1011            if rows.is_empty() {
1012                println!("No breakdown data.");
1013            } else {
1014                let table = Table::new(rows).to_string();
1015                println!("{}", table);
1016            }
1017            if let Some(totals) = &response.totals {
1018                if !totals.is_empty() {
1019                    println!("Totals: {}", format_metrics(totals, &[]));
1020                }
1021            }
1022        }
1023        OutputFormat::Json => {
1024            println!("{}", serde_json::to_string_pretty(&response)?);
1025        }
1026    }
1027    Ok(())
1028}
1029
1030fn render_realtime(response: &RealtimeVisitorsResponse, format: OutputFormat) -> Result<(), Error> {
1031    match format {
1032        OutputFormat::Human => {
1033            println!("Visitors: {}", response.visitors);
1034            if let Some(pageviews) = response.pageviews {
1035                println!("Pageviews: {}", pageviews);
1036            }
1037            if let Some(bounce) = response.bounce_rate {
1038                println!("Bounce rate: {:.2}%", bounce * 100.0);
1039            }
1040            if let Some(duration) = response.visit_duration {
1041                println!("Visit duration: {:.2}s", duration);
1042            }
1043        }
1044        OutputFormat::Json => {
1045            println!("{}", serde_json::to_string_pretty(response)?);
1046        }
1047    }
1048    Ok(())
1049}
1050
1051fn render_event_template(format: OutputFormat) -> Result<(), Error> {
1052    let sample = serde_json::json!({
1053        "name": "Signup",
1054        "url": "https://example.com/signup",
1055        "domain": "example.com",
1056        "referrer": "https://google.com",
1057        "utm_source": "newsletter",
1058        "utm_medium": "email",
1059        "device_type": "desktop"
1060    });
1061    match format {
1062        OutputFormat::Human => {
1063            println!(
1064                "Sample event payload:\n{}",
1065                serde_json::to_string_pretty(&sample)?
1066            );
1067        }
1068        OutputFormat::Json => {
1069            println!("{}", serde_json::to_string_pretty(&sample)?);
1070        }
1071    }
1072    Ok(())
1073}
1074
1075fn render_queue_snapshot(snapshot: &[QueueSnapshot], format: OutputFormat) -> Result<(), Error> {
1076    match format {
1077        OutputFormat::Human => {
1078            if snapshot.is_empty() {
1079                println!("Queue is empty.");
1080            } else {
1081                let rows: Vec<_> = snapshot.iter().map(QueueRow::from).collect();
1082                let table = Table::new(rows).to_string();
1083                println!("{}", table);
1084            }
1085        }
1086        OutputFormat::Json => {
1087            let payload: Vec<_> = snapshot
1088                .iter()
1089                .map(|entry| {
1090                    serde_json::json!({
1091                        "id": entry.id,
1092                        "account": entry.account,
1093                        "description": entry.description,
1094                        "state": match entry.state {
1095                            QueueJobState::Pending => "pending",
1096                            QueueJobState::InFlight => "in_flight",
1097                        },
1098                        "enqueued_at": format_timestamp(&entry.enqueued_at),
1099                        "started_at": entry.started_at.as_ref().map(format_timestamp),
1100                        "attempt": entry.attempt,
1101                        "max_retries": entry.max_retries,
1102                        "next_retry_at": entry.next_retry_at.as_ref().map(format_timestamp),
1103                        "last_error": entry.last_error,
1104                    })
1105                })
1106                .collect();
1107            println!("{}", serde_json::to_string_pretty(&payload)?);
1108        }
1109    }
1110    Ok(())
1111}
1112
1113fn handle_account_command(
1114    store: &AccountStore,
1115    command: &AccountsCommand,
1116    format: OutputFormat,
1117) -> Result<(), Error> {
1118    match command {
1119        AccountsCommand::List => {
1120            let accounts = store.list_accounts()?;
1121            render_account_list(&accounts, format)?;
1122        }
1123        AccountsCommand::Add(args) => {
1124            store.add_account(
1125                &args.alias,
1126                &args.api_key,
1127                AccountProfile {
1128                    label: args.label.clone(),
1129                    email: args.email.clone(),
1130                    description: args.description.clone(),
1131                },
1132            )?;
1133            println!("Added account '{}'.", args.alias);
1134        }
1135        AccountsCommand::Use { alias } => {
1136            store.set_default(alias)?;
1137            println!("Set '{}' as the default account.", alias);
1138        }
1139        AccountsCommand::Remove { alias } => {
1140            store.remove_account(alias)?;
1141            println!("Removed account '{}'.", alias);
1142        }
1143        AccountsCommand::Export { json } => {
1144            let exports = store.export_accounts()?;
1145            if *json || matches!(format, OutputFormat::Json) {
1146                println!("{}", serde_json::to_string_pretty(&exports)?);
1147            } else {
1148                render_account_export(&exports)?;
1149            }
1150        }
1151        AccountsCommand::Budget(args) => {
1152            if args.clear && args.daily.is_some() {
1153                return Err(Error::InvalidInput(
1154                    "use either --daily or --clear when configuring budgets".into(),
1155                ));
1156            }
1157
1158            let budget = if args.clear {
1159                None
1160            } else if let Some(value) = args.daily {
1161                if value == 0 {
1162                    None
1163                } else {
1164                    Some(NonZeroU32::new(value).ok_or_else(|| {
1165                        Error::InvalidInput("daily budget must be greater than zero".into())
1166                    })?)
1167                }
1168            } else {
1169                return Err(Error::InvalidInput(
1170                    "provide --daily to set or --clear to remove a budget".into(),
1171                ));
1172            };
1173
1174            store.set_daily_budget(&args.alias, budget)?;
1175            match budget {
1176                Some(limit) => println!(
1177                    "Set daily budget for '{}' to {} requests.",
1178                    args.alias, limit
1179                ),
1180                None => println!("Cleared daily budget for '{}'.", args.alias),
1181            }
1182        }
1183    }
1184    Ok(())
1185}
1186
1187fn render_account_list(accounts: &[AccountSummary], format: OutputFormat) -> Result<(), Error> {
1188    match format {
1189        OutputFormat::Human => {
1190            let rows: Vec<_> = accounts.iter().map(AccountRow::from).collect();
1191            let table = Table::new(rows).to_string();
1192            println!("{}", table);
1193        }
1194        OutputFormat::Json => {
1195            let payload: Vec<_> = accounts
1196                .iter()
1197                .map(|account| {
1198                    serde_json::json!({
1199                        "alias": account.alias,
1200                        "label": account.profile.label,
1201                        "email": account.profile.email,
1202                        "description": account.profile.description,
1203                        "default": account.is_default,
1204                        "daily_budget": account.daily_budget.map(|v| v.get()),
1205                    })
1206                })
1207                .collect();
1208            println!("{}", serde_json::to_string_pretty(&payload)?);
1209        }
1210    }
1211    Ok(())
1212}
1213
1214fn render_account_export(exports: &[AccountExport]) -> Result<(), Error> {
1215    let rows: Vec<_> = exports.iter().map(ExportRow::from).collect();
1216    let table = Table::new(rows).to_string();
1217    println!("{}", table);
1218    Ok(())
1219}
1220
1221#[derive(Tabled)]
1222struct SiteRow {
1223    domain: String,
1224    timezone: String,
1225    public: String,
1226    verified: String,
1227}
1228
1229impl From<&SiteSummary> for SiteRow {
1230    fn from(site: &SiteSummary) -> Self {
1231        Self {
1232            domain: site.domain.clone(),
1233            timezone: site.timezone.clone().unwrap_or_else(|| "n/a".into()),
1234            public: site
1235                .public
1236                .map(|v| v.to_string())
1237                .unwrap_or_else(|| "n/a".into()),
1238            verified: site
1239                .verified
1240                .map(|v| v.to_string())
1241                .unwrap_or_else(|| "n/a".into()),
1242        }
1243    }
1244}
1245
1246#[derive(Tabled)]
1247struct MetricRow {
1248    metric: String,
1249    value: String,
1250}
1251
1252#[derive(Tabled)]
1253struct TimeseriesRow {
1254    timestamp: String,
1255    metrics: String,
1256}
1257
1258impl From<&serde_json::Map<String, serde_json::Value>> for TimeseriesRow {
1259    fn from(entry: &serde_json::Map<String, serde_json::Value>) -> Self {
1260        let timestamp = entry
1261            .get("datetime")
1262            .or_else(|| entry.get("date"))
1263            .or_else(|| entry.get("time"))
1264            .map(format_value)
1265            .unwrap_or_else(|| "n/a".into());
1266        let metrics = format_metrics(entry, &["datetime", "date", "time"]);
1267        Self { timestamp, metrics }
1268    }
1269}
1270
1271#[derive(Tabled)]
1272struct BreakdownRow {
1273    segment: String,
1274    metrics: String,
1275}
1276
1277impl From<&serde_json::Map<String, serde_json::Value>> for BreakdownRow {
1278    fn from(entry: &serde_json::Map<String, serde_json::Value>) -> Self {
1279        let segment = entry
1280            .get("value")
1281            .or_else(|| entry.get("name"))
1282            .map(format_value)
1283            .unwrap_or_else(|| "n/a".into());
1284        let metrics = format_metrics(entry, &["value", "name"]);
1285        Self { segment, metrics }
1286    }
1287}
1288
1289#[derive(Tabled)]
1290struct QueueRow {
1291    id: u64,
1292    account: String,
1293    description: String,
1294    state: String,
1295    enqueued_at: String,
1296    started_at: String,
1297    attempt: String,
1298    next_retry_at: String,
1299    last_error: String,
1300}
1301
1302impl From<&QueueSnapshot> for QueueRow {
1303    fn from(entry: &QueueSnapshot) -> Self {
1304        let attempt_display = format!("{}/{}", entry.attempt, entry.max_retries);
1305        Self {
1306            id: entry.id,
1307            account: entry.account.clone(),
1308            description: entry.description.clone(),
1309            state: match entry.state {
1310                QueueJobState::Pending => "pending".into(),
1311                QueueJobState::InFlight => "in-flight".into(),
1312            },
1313            enqueued_at: format_timestamp(&entry.enqueued_at),
1314            started_at: entry
1315                .started_at
1316                .as_ref()
1317                .map(format_timestamp)
1318                .unwrap_or_else(|| "n/a".into()),
1319            attempt: attempt_display,
1320            next_retry_at: entry
1321                .next_retry_at
1322                .as_ref()
1323                .map(format_timestamp)
1324                .unwrap_or_else(|| "n/a".into()),
1325            last_error: entry.last_error.clone().unwrap_or_else(|| "".into()),
1326        }
1327    }
1328}
1329
1330fn confirm_deletion(site: &str) -> Result<bool, Error> {
1331    println!("Delete site '{}'? Type 'yes' to confirm:", site);
1332    let mut input = String::new();
1333    io::stdin().read_line(&mut input)?;
1334    Ok(matches!(input.trim().to_lowercase().as_str(), "yes" | "y"))
1335}
1336
1337fn format_metrics(map: &serde_json::Map<String, serde_json::Value>, skip: &[&str]) -> String {
1338    map.iter()
1339        .filter(|(key, _)| {
1340            let key_str = key.as_str();
1341            !skip.contains(&key_str)
1342        })
1343        .map(|(key, value)| format!("{key}={}", format_value(value)))
1344        .collect::<Vec<_>>()
1345        .join(", ")
1346}
1347
1348fn format_value(value: &serde_json::Value) -> String {
1349    match value {
1350        serde_json::Value::Null => "null".into(),
1351        serde_json::Value::Bool(b) => b.to_string(),
1352        serde_json::Value::Number(n) => n.to_string(),
1353        serde_json::Value::String(s) => s.clone(),
1354        other => serde_json::to_string(other).unwrap_or_else(|_| "<unserializable>".into()),
1355    }
1356}
1357
1358fn format_timestamp(timestamp: &OffsetDateTime) -> String {
1359    timestamp
1360        .format(&Rfc3339)
1361        .unwrap_or_else(|_| timestamp.to_string())
1362}
1363
1364#[derive(Tabled)]
1365struct AccountRow {
1366    alias: String,
1367    label: String,
1368    email: String,
1369    default: String,
1370    daily_budget: String,
1371}
1372
1373impl From<&AccountSummary> for AccountRow {
1374    fn from(summary: &AccountSummary) -> Self {
1375        Self {
1376            alias: summary.alias.clone(),
1377            label: summary
1378                .profile
1379                .label
1380                .clone()
1381                .unwrap_or_else(|| "n/a".into()),
1382            email: summary
1383                .profile
1384                .email
1385                .clone()
1386                .unwrap_or_else(|| "n/a".into()),
1387            default: if summary.is_default {
1388                "yes".into()
1389            } else {
1390                "no".into()
1391            },
1392            daily_budget: summary
1393                .daily_budget
1394                .map(|v| v.get().to_string())
1395                .unwrap_or_else(|| "-".into()),
1396        }
1397    }
1398}
1399
1400#[derive(Tabled)]
1401struct ExportRow {
1402    alias: String,
1403    label: String,
1404    email: String,
1405    default: String,
1406    description: String,
1407    daily_budget: String,
1408}
1409
1410impl From<&AccountExport> for ExportRow {
1411    fn from(export: &AccountExport) -> Self {
1412        Self {
1413            alias: export.alias.clone(),
1414            label: export.label.clone().unwrap_or_else(|| "n/a".into()),
1415            email: export.email.clone().unwrap_or_else(|| "n/a".into()),
1416            default: if export.is_default {
1417                "yes".into()
1418            } else {
1419                "no".into()
1420            },
1421            description: export.description.clone().unwrap_or_else(|| "".into()),
1422            daily_budget: export
1423                .daily_budget
1424                .map(|v| v.to_string())
1425                .unwrap_or_else(|| "-".into()),
1426        }
1427    }
1428}
1429
1430struct PlausibleExecutor {
1431    client: PlausibleClient,
1432}
1433
1434impl PlausibleExecutor {
1435    fn new(client: PlausibleClient) -> Self {
1436        Self { client }
1437    }
1438}
1439
1440#[async_trait]
1441impl crate::queue::JobExecutor for PlausibleExecutor {
1442    async fn execute(&self, request: JobRequest) -> crate::queue::JobResult {
1443        match request.kind {
1444            JobKind::ListSites => {
1445                let sites = self
1446                    .client
1447                    .list_sites()
1448                    .await
1449                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1450                Ok(JobResponse::Sites(sites))
1451            }
1452            JobKind::StatsAggregate { query } => {
1453                let result = self
1454                    .client
1455                    .stats_aggregate(&query)
1456                    .await
1457                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1458                Ok(JobResponse::StatsAggregate(result))
1459            }
1460            JobKind::StatsTimeseries { query } => {
1461                let result = self
1462                    .client
1463                    .stats_timeseries(&query)
1464                    .await
1465                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1466                Ok(JobResponse::StatsTimeseries(result))
1467            }
1468            JobKind::StatsBreakdown { query } => {
1469                let result = self
1470                    .client
1471                    .stats_breakdown(&query)
1472                    .await
1473                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1474                Ok(JobResponse::StatsBreakdown(result))
1475            }
1476            JobKind::SiteCreate { request } => {
1477                let site = self
1478                    .client
1479                    .create_site(&request)
1480                    .await
1481                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1482                Ok(JobResponse::SiteCreated(site))
1483            }
1484            JobKind::SiteUpdate { site_id, request } => {
1485                let site = self
1486                    .client
1487                    .update_site(&site_id, &request)
1488                    .await
1489                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1490                Ok(JobResponse::SiteUpdated(site))
1491            }
1492            JobKind::SiteReset { site_id, request } => {
1493                self.client
1494                    .reset_site_stats(&site_id, &request)
1495                    .await
1496                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1497                Ok(JobResponse::SiteReset)
1498            }
1499            JobKind::SiteDelete { site_id } => {
1500                self.client
1501                    .delete_site(&site_id)
1502                    .await
1503                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1504                Ok(JobResponse::SiteDeleted)
1505            }
1506            JobKind::StatsRealtime { site_id } => {
1507                let realtime = self
1508                    .client
1509                    .stats_realtime_visitors(&site_id)
1510                    .await
1511                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1512                Ok(JobResponse::StatsRealtime(realtime))
1513            }
1514            JobKind::EventSend { event } => {
1515                self.client
1516                    .send_event(&event)
1517                    .await
1518                    .map_err(|err| WorkerError::Execution(err.to_string()))?;
1519                Ok(JobResponse::EventAck)
1520            }
1521            JobKind::EventsImport { events } => {
1522                let mut processed = 0usize;
1523                for event in events {
1524                    self.client
1525                        .send_event(&event)
1526                        .await
1527                        .map_err(|err| WorkerError::Execution(err.to_string()))?;
1528                    processed += 1;
1529                }
1530                Ok(JobResponse::EventsProcessed { processed })
1531            }
1532            JobKind::Custom { .. } => Ok(JobResponse::Acknowledged),
1533        }
1534    }
1535}
1536
1537#[cfg(test)]
1538mod tests {
1539    use super::*;
1540
1541    #[test]
1542    fn timeseries_builder_copies_fields() {
1543        let args = StatsTimeseriesArgs {
1544            site: "example.com".into(),
1545            metrics: vec!["visitors".into()],
1546            period: Some("7d".into()),
1547            date: Some("2024-01-01,2024-01-07".into()),
1548            filters: vec!["event:page==/docs".into()],
1549            properties: vec!["visit:source".into()],
1550            compare: Some("previous_period".into()),
1551            interval: Some("date".into()),
1552            sort: Some("visitors:desc".into()),
1553            limit: Some(10),
1554            page: Some(2),
1555        };
1556
1557        let query = build_timeseries_query(&args);
1558        assert_eq!(query.site_id, "example.com");
1559        assert_eq!(query.metrics, vec!["visitors"]);
1560        assert_eq!(query.period.as_deref(), Some("7d"));
1561        assert_eq!(query.interval.as_deref(), Some("date"));
1562        assert_eq!(query.limit, Some(10));
1563        assert_eq!(query.page, Some(2));
1564        assert_eq!(query.filters, vec!["event:page==/docs"]);
1565        assert_eq!(query.properties, vec!["visit:source"]);
1566    }
1567
1568    #[test]
1569    fn breakdown_builder_includes_property_and_include() {
1570        let args = StatsBreakdownArgs {
1571            site: "example.com".into(),
1572            property: "event:page".into(),
1573            metrics: vec!["visitors".into(), "pageviews".into()],
1574            period: Some("30d".into()),
1575            date: None,
1576            filters: vec![],
1577            properties: vec![],
1578            compare: Some("previous_period".into()),
1579            sort: Some("visitors:desc".into()),
1580            limit: Some(25),
1581            page: Some(1),
1582            include: Some("previous".into()),
1583        };
1584
1585        let query = build_breakdown_query(&args);
1586        assert_eq!(query.property, "event:page");
1587        assert_eq!(query.metrics, vec!["visitors", "pageviews"]);
1588        assert_eq!(query.include.as_deref(), Some("previous"));
1589        assert_eq!(query.limit, Some(25));
1590    }
1591
1592    #[test]
1593    fn parse_event_json_applies_domain_override() {
1594        let domain = Some(String::from("example.com"));
1595        let value = parse_event_json(r#"{"name":"Signup"}"#, &domain).expect("parse");
1596        assert_eq!(
1597            value.get("domain").and_then(|v| v.as_str()),
1598            Some("example.com")
1599        );
1600    }
1601
1602    #[test]
1603    fn parse_ndjson_parses_multiple_events() {
1604        let domain = None;
1605        let events = parse_ndjson("{\"name\":\"Signup\"}\n{\"name\":\"Upgrade\"}\n", &domain)
1606            .expect("ndjson");
1607        assert_eq!(events.len(), 2);
1608        assert_eq!(
1609            events[0].get("name").and_then(|v| v.as_str()),
1610            Some("Signup")
1611        );
1612    }
1613
1614    #[test]
1615    fn parse_ndjson_errors_when_empty() {
1616        let err = parse_ndjson("\n\n", &None).expect_err("empty");
1617        match err {
1618            Error::InvalidInput(msg) => assert!(msg.contains("no events")),
1619            _ => panic!("unexpected error"),
1620        }
1621    }
1622
1623    #[test]
1624    fn build_create_site_request_sets_optional_fields() {
1625        let args = SiteCreateArgs {
1626            domain: "example.com".into(),
1627            timezone: Some("UTC".into()),
1628            public: Some(true),
1629        };
1630        let request = build_create_site_request(&args);
1631        assert_eq!(request.domain, "example.com");
1632        assert_eq!(request.timezone.as_deref(), Some("UTC"));
1633        assert_eq!(request.public, Some(true));
1634    }
1635
1636    #[test]
1637    fn build_update_site_request_respects_options() {
1638        let args = SiteUpdateArgs {
1639            site: "example.com".into(),
1640            timezone: Some("Europe/Berlin".into()),
1641            public: Some(false),
1642            main_site: Some(true),
1643        };
1644        let (site_id, request) = build_update_site_request(&args);
1645        assert_eq!(site_id, "example.com");
1646        assert_eq!(request.timezone.as_deref(), Some("Europe/Berlin"));
1647        assert_eq!(request.public, Some(false));
1648        assert_eq!(request.main_site, Some(true));
1649    }
1650
1651    #[test]
1652    fn build_reset_site_request_handles_optional_date() {
1653        let args = SiteResetArgs {
1654            site: "example.com".into(),
1655            date: Some("2024-01-01".into()),
1656        };
1657        let (site_id, request) = build_reset_site_request(&args);
1658        assert_eq!(site_id, "example.com");
1659        assert_eq!(request.date.as_deref(), Some("2024-01-01"));
1660    }
1661}