perf(analytics): speed up stats endpoint aggregation (#766)
This commit is contained in:
@@ -90,6 +90,7 @@ type GitStatsDTO struct {
|
||||
type StatsResponse struct {
|
||||
Global GlobalStatsDTO `json:"global"`
|
||||
TaskStats []TaskStatsDTO `json:"task_stats"`
|
||||
TaskStatsHasMore bool `json:"task_stats_has_more"`
|
||||
DailyActivity []DailyActivityDTO `json:"daily_activity"`
|
||||
CompletedActivity []CompletedTaskActivityDTO `json:"completed_activity"`
|
||||
AgentUsage []AgentUsageDTO `json:"agent_usage"`
|
||||
|
||||
@@ -13,10 +13,12 @@ import (
|
||||
"github.com/kandev/kandev/internal/analytics/repository"
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/sync/errgroup"
|
||||
)
|
||||
|
||||
// allTimeActivityDays is the number of days shown in the daily activity heatmap for the "all" range.
|
||||
const allTimeActivityDays = 365
|
||||
const taskStatsLimit = 200
|
||||
|
||||
type StatsHandlers struct {
|
||||
repo repository.Repository
|
||||
@@ -84,6 +86,7 @@ func (h *StatsHandlers) httpGetStats(c *gin.Context) {
|
||||
AvgDurationMsPerTask: raw.globalStats.AvgDurationMsPerTask,
|
||||
},
|
||||
TaskStats: taskStatsToDTOs(raw.taskStats),
|
||||
TaskStatsHasMore: raw.globalStats.TotalTasks > len(raw.taskStats),
|
||||
DailyActivity: dailyActivityToDTOs(raw.dailyActivity),
|
||||
CompletedActivity: completedActivityToDTOs(raw.completedActivity),
|
||||
AgentUsage: agentUsageToDTOs(raw.agentUsage),
|
||||
@@ -100,34 +103,77 @@ func (h *StatsHandlers) httpGetStats(c *gin.Context) {
|
||||
}
|
||||
|
||||
func (h *StatsHandlers) fetchStats(ctx context.Context, workspaceID string, start *time.Time, days int) (*rawStats, error) {
|
||||
globalStats, err := h.repo.GetGlobalStats(ctx, workspaceID, start)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("global stats: %w", err)
|
||||
}
|
||||
taskStats, err := h.repo.GetTaskStats(ctx, workspaceID, start)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("task stats: %w", err)
|
||||
}
|
||||
dailyActivity, err := h.repo.GetDailyActivity(ctx, workspaceID, days)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("daily activity: %w", err)
|
||||
}
|
||||
completedActivity, err := h.repo.GetCompletedTaskActivity(ctx, workspaceID, days)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("completed activity: %w", err)
|
||||
}
|
||||
agentUsage, err := h.repo.GetAgentUsage(ctx, workspaceID, 5, start)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("agent usage: %w", err)
|
||||
}
|
||||
repoStats, err := h.repo.GetRepositoryStats(ctx, workspaceID, start)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("repository stats: %w", err)
|
||||
}
|
||||
gitStats, err := h.repo.GetGitStats(ctx, workspaceID, start)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("git stats: %w", err)
|
||||
var (
|
||||
globalStats *models.GlobalStats
|
||||
taskStats []*models.TaskStats
|
||||
dailyActivity []*models.DailyActivity
|
||||
completedActivity []*models.CompletedTaskActivity
|
||||
agentUsage []*models.AgentUsage
|
||||
repoStats []*models.RepositoryStats
|
||||
gitStats *models.GitStats
|
||||
)
|
||||
|
||||
group, groupCtx := errgroup.WithContext(ctx)
|
||||
group.Go(func() error {
|
||||
result, err := h.repo.GetGlobalStats(groupCtx, workspaceID, start)
|
||||
if err != nil {
|
||||
return fmt.Errorf("global stats: %w", err)
|
||||
}
|
||||
globalStats = result
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
result, err := h.repo.GetTaskStats(groupCtx, workspaceID, start, taskStatsLimit)
|
||||
if err != nil {
|
||||
return fmt.Errorf("task stats: %w", err)
|
||||
}
|
||||
taskStats = result
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
result, err := h.repo.GetDailyActivity(groupCtx, workspaceID, days)
|
||||
if err != nil {
|
||||
return fmt.Errorf("daily activity: %w", err)
|
||||
}
|
||||
dailyActivity = result
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
result, err := h.repo.GetCompletedTaskActivity(groupCtx, workspaceID, days)
|
||||
if err != nil {
|
||||
return fmt.Errorf("completed activity: %w", err)
|
||||
}
|
||||
completedActivity = result
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
result, err := h.repo.GetAgentUsage(groupCtx, workspaceID, 5, start)
|
||||
if err != nil {
|
||||
return fmt.Errorf("agent usage: %w", err)
|
||||
}
|
||||
agentUsage = result
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
result, err := h.repo.GetRepositoryStats(groupCtx, workspaceID, start)
|
||||
if err != nil {
|
||||
return fmt.Errorf("repository stats: %w", err)
|
||||
}
|
||||
repoStats = result
|
||||
return nil
|
||||
})
|
||||
group.Go(func() error {
|
||||
result, err := h.repo.GetGitStats(groupCtx, workspaceID, start)
|
||||
if err != nil {
|
||||
return fmt.Errorf("git stats: %w", err)
|
||||
}
|
||||
gitStats = result
|
||||
return nil
|
||||
})
|
||||
if err := group.Wait(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &rawStats{
|
||||
globalStats: globalStats,
|
||||
taskStats: taskStats,
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
|
||||
// Repository defines the interface for analytics/statistics operations.
|
||||
type Repository interface {
|
||||
GetTaskStats(ctx context.Context, workspaceID string, start *time.Time) ([]*models.TaskStats, error)
|
||||
GetTaskStats(ctx context.Context, workspaceID string, start *time.Time, limit int) ([]*models.TaskStats, error)
|
||||
GetGlobalStats(ctx context.Context, workspaceID string, start *time.Time) (*models.GlobalStats, error)
|
||||
GetDailyActivity(ctx context.Context, workspaceID string, days int) ([]*models.DailyActivity, error)
|
||||
GetCompletedTaskActivity(ctx context.Context, workspaceID string, days int) ([]*models.CompletedTaskActivity, error)
|
||||
|
||||
@@ -2,6 +2,7 @@ package sqlite
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
@@ -370,7 +371,7 @@ func TestGetTaskStats_ExcludesEphemeralTasks(t *testing.T) {
|
||||
execOrFatal(t, dbConn, `INSERT INTO tasks (id, workspace_id, board_id, workflow_step_id, title, is_ephemeral, created_at, updated_at) VALUES ('task-regular', 'ws-1', 'board-1', '', 'Regular', 0, ?, ?)`, nowStr, nowStr)
|
||||
execOrFatal(t, dbConn, `INSERT INTO tasks (id, workspace_id, board_id, workflow_step_id, title, is_ephemeral, created_at, updated_at) VALUES ('task-ephemeral', 'ws-1', 'board-1', '', 'Quick Chat', 1, ?, ?)`, nowStr, nowStr)
|
||||
|
||||
results, err := repo.GetTaskStats(ctx, "ws-1", nil)
|
||||
results, err := repo.GetTaskStats(ctx, "ws-1", nil, 100)
|
||||
if err != nil {
|
||||
t.Fatalf("GetTaskStats failed: %v", err)
|
||||
}
|
||||
@@ -382,6 +383,43 @@ func TestGetTaskStats_ExcludesEphemeralTasks(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetTaskStats_RespectsLimit(t *testing.T) {
|
||||
dbConn := createTestDB(t)
|
||||
repo, err := NewWithDB(dbConn, dbConn)
|
||||
if err != nil {
|
||||
t.Fatalf("NewWithDB failed: %v", err)
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
nowStr := now.Format(time.RFC3339)
|
||||
|
||||
execOrFatal(t, dbConn, `INSERT INTO workspaces (id, name, created_at, updated_at) VALUES ('ws-1', 'Test', ?, ?)`, nowStr, nowStr)
|
||||
execOrFatal(t, dbConn, `INSERT INTO boards (id, workspace_id, name, created_at, updated_at) VALUES ('board-1', 'ws-1', 'Board', ?, ?)`, nowStr, nowStr)
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
taskID := fmt.Sprintf("task-%d", i)
|
||||
taskTitle := fmt.Sprintf("Task %d", i)
|
||||
execOrFatal(
|
||||
t,
|
||||
dbConn,
|
||||
`INSERT INTO tasks (id, workspace_id, board_id, workflow_step_id, title, is_ephemeral, created_at, updated_at) VALUES (?, 'ws-1', 'board-1', '', ?, 0, ?, ?)`,
|
||||
taskID,
|
||||
taskTitle,
|
||||
nowStr,
|
||||
now.Add(time.Duration(i)*time.Second).Format(time.RFC3339),
|
||||
)
|
||||
}
|
||||
|
||||
results, err := repo.GetTaskStats(ctx, "ws-1", nil, 3)
|
||||
if err != nil {
|
||||
t.Fatalf("GetTaskStats failed: %v", err)
|
||||
}
|
||||
if len(results) != 3 {
|
||||
t.Fatalf("expected 3 task stats due to limit, got %d", len(results))
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetTaskStats_IncludesActiveAndElapsedDurations(t *testing.T) {
|
||||
dbConn := createTestDB(t)
|
||||
repo, err := NewWithDB(dbConn, dbConn)
|
||||
@@ -402,7 +440,7 @@ func TestGetTaskStats_IncludesActiveAndElapsedDurations(t *testing.T) {
|
||||
execOrFatal(t, dbConn, `INSERT INTO task_session_turns (id, task_session_id, task_id, started_at, completed_at, created_at, updated_at) VALUES ('turn-1', 'sess-1', 'task-1', '2026-01-01T10:00:00Z', '2026-01-01T10:10:00Z', ?, ?)`, nowStr, nowStr)
|
||||
execOrFatal(t, dbConn, `INSERT INTO task_session_turns (id, task_session_id, task_id, started_at, completed_at, created_at, updated_at) VALUES ('turn-2', 'sess-1', 'task-1', '2026-01-01T11:10:00Z', '2026-01-01T11:20:00Z', ?, ?)`, nowStr, nowStr)
|
||||
|
||||
results, err := repo.GetTaskStats(ctx, "ws-1", nil)
|
||||
results, err := repo.GetTaskStats(ctx, "ws-1", nil, 100)
|
||||
if err != nil {
|
||||
t.Fatalf("GetTaskStats failed: %v", err)
|
||||
}
|
||||
|
||||
@@ -33,12 +33,20 @@ func parseTimeString(s string) time.Time {
|
||||
return time.Time{}
|
||||
}
|
||||
|
||||
// GetTaskStats retrieves aggregated statistics for all tasks in a workspace
|
||||
func (r *Repository) GetTaskStats(ctx context.Context, workspaceID string, start *time.Time) ([]*models.TaskStats, error) {
|
||||
// GetTaskStats retrieves aggregated statistics for tasks in a workspace.
|
||||
func (r *Repository) GetTaskStats(
|
||||
ctx context.Context,
|
||||
workspaceID string,
|
||||
start *time.Time,
|
||||
limit int,
|
||||
) ([]*models.TaskStats, error) {
|
||||
var startArg any
|
||||
if start != nil {
|
||||
startArg = start.UTC().Format(time.RFC3339)
|
||||
}
|
||||
if limit <= 0 {
|
||||
limit = 200
|
||||
}
|
||||
|
||||
drv := r.ro.DriverName()
|
||||
dur := dialect.DurationMs(drv, "turn.completed_at", "turn.started_at")
|
||||
@@ -81,6 +89,7 @@ func (r *Repository) GetTaskStats(ctx context.Context, workspaceID string, start
|
||||
) turn_stats ON turn_stats.task_id = t.id
|
||||
WHERE t.workspace_id = ? AND t.is_ephemeral = 0 AND (? IS NULL OR t.created_at >= ?)
|
||||
ORDER BY t.updated_at DESC
|
||||
LIMIT ?
|
||||
`, dur, dialect.DurationMs(
|
||||
drv,
|
||||
"MAX(CASE WHEN turn.completed_at IS NOT NULL THEN turn.completed_at END)",
|
||||
@@ -89,7 +98,7 @@ func (r *Repository) GetTaskStats(ctx context.Context, workspaceID string, start
|
||||
|
||||
rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query),
|
||||
startArg, startArg, startArg, startArg,
|
||||
workspaceID, startArg, startArg,
|
||||
workspaceID, startArg, startArg, limit,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
Reference in New Issue
Block a user