systemprompt-analytics 0.67.0

//! Tool execution leaderboard queries (filtered + unfiltered) for the
//! analytics CLI surface. A host-native tool a client hook reported (Bash,
//! Read, …) is recorded under the vantage point's own name as its server;
//! those pseudo-servers are not MCP tools and stay out of every list. Every
//! list reads the hourly rollup (`report_tool_usage_hourly`), widening the
//! window to whole UTC hours at its start.
//!
//! Copyright (c) systemprompt.io — Business Source License 1.1.
//! See <https://systemprompt.io> for licensing details.

use crate::Result;
use chrono::{DateTime, Utc};

use super::ToolAnalyticsRepository;
use crate::models::ToolListDbRow;
use crate::models::reporting::ToolListRow;

#[derive(Debug)]
pub struct ToolListParams<'a> {
    pub start: DateTime<Utc>,
    pub end: DateTime<Utc>,
    pub limit: i64,
    pub server_filter: Option<&'a str>,
    pub sort_order: &'a str,
}

impl ToolAnalyticsRepository {
    pub async fn list_tools(&self, params: ToolListParams<'_>) -> Result<Vec<ToolListRow>> {
        let rows = if let Some(server) = params.server_filter {
            let pattern = format!("%{}%", server);
            self.list_tools_with_filter(&params, &pattern).await
        } else {
            self.list_tools_unfiltered(&params).await
        }?;
        Ok(rows.into_iter().map(ToolListRow::from).collect())
    }

    async fn list_tools_with_filter(
        &self,
        params: &ToolListParams<'_>,
        pattern: &str,
    ) -> Result<Vec<ToolListDbRow>> {
        let ToolListParams {
            start,
            end,
            limit,
            sort_order,
            ..
        } = *params;
        match sort_order {
            "success_rate" => {
                self.filtered_by_success_rate(start, end, pattern, limit)
                    .await
            },
            "avg_time" => self.filtered_by_avg_time(start, end, pattern, limit).await,
            _ => self.filtered_by_count(start, end, pattern, limit).await,
        }
    }

    async fn filtered_by_success_rate(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        pattern: &str,
        limit: i64,
    ) -> Result<Vec<ToolListDbRow>> {
        sqlx::query_as!(
            ToolListDbRow,
            r#"
            SELECT
                tool_name as "tool_name!",
                server_name as "server_name!",
                SUM(execution_count)::bigint as "execution_count!",
                COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::bigint as "success_count!",
                COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_time!",
                MAX(last_at) as "last_used!"
            FROM report_tool_usage_hourly
            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2 AND server_name ILIKE $3
              AND server_name NOT IN ('in_process', 'proxy', 'gateway', 'hook_claude_code', 'hook_opencode')
            GROUP BY tool_name, server_name
            ORDER BY CASE WHEN SUM(execution_count) > 0
                THEN COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::float / SUM(execution_count)::float
                ELSE 0 END DESC
            LIMIT $4
            "#,
            start,
            end,
            pattern,
            limit
        )
        .fetch_all(&*self.pool)
        .await
        .map_err(Into::into)
    }

    async fn filtered_by_avg_time(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        pattern: &str,
        limit: i64,
    ) -> Result<Vec<ToolListDbRow>> {
        sqlx::query_as!(
            ToolListDbRow,
            r#"
            SELECT
                tool_name as "tool_name!",
                server_name as "server_name!",
                SUM(execution_count)::bigint as "execution_count!",
                COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::bigint as "success_count!",
                COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_time!",
                MAX(last_at) as "last_used!"
            FROM report_tool_usage_hourly
            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2 AND server_name ILIKE $3
              AND server_name NOT IN ('in_process', 'proxy', 'gateway', 'hook_claude_code', 'hook_opencode')
            GROUP BY tool_name, server_name
            ORDER BY COALESCE(SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0), 0) DESC
            LIMIT $4
            "#,
            start,
            end,
            pattern,
            limit
        )
        .fetch_all(&*self.pool)
        .await
        .map_err(Into::into)
    }

    async fn filtered_by_count(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        pattern: &str,
        limit: i64,
    ) -> Result<Vec<ToolListDbRow>> {
        sqlx::query_as!(
            ToolListDbRow,
            r#"
            SELECT
                tool_name as "tool_name!",
                server_name as "server_name!",
                SUM(execution_count)::bigint as "execution_count!",
                COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::bigint as "success_count!",
                COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_time!",
                MAX(last_at) as "last_used!"
            FROM report_tool_usage_hourly
            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2 AND server_name ILIKE $3
              AND server_name NOT IN ('in_process', 'proxy', 'gateway', 'hook_claude_code', 'hook_opencode')
            GROUP BY tool_name, server_name
            ORDER BY SUM(execution_count) DESC
            LIMIT $4
            "#,
            start,
            end,
            pattern,
            limit
        )
        .fetch_all(&*self.pool)
        .await
        .map_err(Into::into)
    }

    async fn list_tools_unfiltered(
        &self,
        params: &ToolListParams<'_>,
    ) -> Result<Vec<ToolListDbRow>> {
        let ToolListParams {
            start,
            end,
            limit,
            sort_order,
            ..
        } = *params;
        match sort_order {
            "success_rate" => self.unfiltered_by_success_rate(start, end, limit).await,
            "avg_time" => self.unfiltered_by_avg_time(start, end, limit).await,
            _ => self.unfiltered_by_count(start, end, limit).await,
        }
    }

    async fn unfiltered_by_success_rate(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        limit: i64,
    ) -> Result<Vec<ToolListDbRow>> {
        sqlx::query_as!(
            ToolListDbRow,
            r#"
            SELECT
                tool_name as "tool_name!",
                server_name as "server_name!",
                SUM(execution_count)::bigint as "execution_count!",
                COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::bigint as "success_count!",
                COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_time!",
                MAX(last_at) as "last_used!"
            FROM report_tool_usage_hourly
            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
              AND server_name NOT IN ('in_process', 'proxy', 'gateway', 'hook_claude_code', 'hook_opencode')
            GROUP BY tool_name, server_name
            ORDER BY CASE WHEN SUM(execution_count) > 0
                THEN COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::float / SUM(execution_count)::float
                ELSE 0 END DESC
            LIMIT $3
            "#,
            start,
            end,
            limit
        )
        .fetch_all(&*self.pool)
        .await
        .map_err(Into::into)
    }

    async fn unfiltered_by_avg_time(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        limit: i64,
    ) -> Result<Vec<ToolListDbRow>> {
        sqlx::query_as!(
            ToolListDbRow,
            r#"
            SELECT
                tool_name as "tool_name!",
                server_name as "server_name!",
                SUM(execution_count)::bigint as "execution_count!",
                COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::bigint as "success_count!",
                COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_time!",
                MAX(last_at) as "last_used!"
            FROM report_tool_usage_hourly
            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
              AND server_name NOT IN ('in_process', 'proxy', 'gateway', 'hook_claude_code', 'hook_opencode')
            GROUP BY tool_name, server_name
            ORDER BY COALESCE(SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0), 0) DESC
            LIMIT $3
            "#,
            start,
            end,
            limit
        )
        .fetch_all(&*self.pool)
        .await
        .map_err(Into::into)
    }

    async fn unfiltered_by_count(
        &self,
        start: DateTime<Utc>,
        end: DateTime<Utc>,
        limit: i64,
    ) -> Result<Vec<ToolListDbRow>> {
        sqlx::query_as!(
            ToolListDbRow,
            r#"
            SELECT
                tool_name as "tool_name!",
                server_name as "server_name!",
                SUM(execution_count)::bigint as "execution_count!",
                COALESCE(SUM(execution_count) FILTER (WHERE status = 'success'), 0)::bigint as "success_count!",
                COALESCE((SUM(exec_sum_ms)::numeric / NULLIF(SUM(exec_n), 0))::float8, 0) as "avg_time!",
                MAX(last_at) as "last_used!"
            FROM report_tool_usage_hourly
            WHERE bucket >= date_trunc('hour', $1::timestamptz, 'UTC') AND bucket < $2
              AND server_name NOT IN ('in_process', 'proxy', 'gateway', 'hook_claude_code', 'hook_opencode')
            GROUP BY tool_name, server_name
            ORDER BY SUM(execution_count) DESC
            LIMIT $3
            "#,
            start,
            end,
            limit
        )
        .fetch_all(&*self.pool)
        .await
        .map_err(Into::into)
    }
}