systemprompt_cli/commands/analytics/
projection.rs

1//! Reporting projection operations against the primary database.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use anyhow::Result;
7use clap::Subcommand;
8use systemprompt_runtime::reporting;
9use systemprompt_runtime::reporting::RebuildOutcome;
10
11use crate::context::CommandContext;
12use crate::shared::{CommandOutput, render_result};
13
14#[derive(Debug, Clone, Copy, Subcommand)]
15pub enum ProjectionCommands {
16    #[command(about = "Show projection generation and pending delivery backlog")]
17    Status,
18    #[command(about = "Atomically rebuild reporting tables from owner snapshots")]
19    Rebuild,
20    #[command(about = "Apply up to the requested number of pending reporting facts")]
21    Sync {
22        #[arg(long, default_value_t = 10000)]
23        limit: usize,
24    },
25}
26
27pub async fn execute(command: ProjectionCommands, ctx: &CommandContext) -> Result<()> {
28    let database = ctx.database().await?;
29    let pool = database.db_pool();
30    match command {
31        ProjectionCommands::Status => {},
32        ProjectionCommands::Rebuild => {
33            // Why: a forced rebuild can be superseded by another replica that
34            // owns the generation, in which case nothing was rebuilt here. The
35            // status card below reads the same either way, so the outcome has
36            // to be said out loud or the operator reads someone else's rebuild
37            // as their own.
38            let outcome = reporting::rebuild(pool).await?;
39            if outcome != RebuildOutcome::Rebuilt {
40                render_result(
41                    &CommandOutput::card_value(
42                        "Reporting projection",
43                        &format!("not rebuilt here: {outcome:?}"),
44                    ),
45                    &ctx.cli,
46                );
47            }
48        },
49        ProjectionCommands::Sync { limit } => {
50            reporting::process_pending(pool, limit).await?;
51        },
52    }
53    let status = reporting::status(pool).await?;
54    render_result(
55        &CommandOutput::card_value("Reporting projection", &status),
56        &ctx.cli,
57    );
58    Ok(())
59}
systemprompt_cli/commands/analytics/projection.rs

systemprompt_cli/commands/analytics/
projection.rs
