From 6e4d22cae09b28059ff356996be85b2863c85855 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Sat, 21 Feb 2026 23:02:52 +0000 Subject: [PATCH] feat: improve sql queries (#260) --- apps/backend/cmd/kandev/helpers.go | 8 ++ apps/backend/cmd/kandev/main.go | 1 + .../backend/internal/common/httpmw/tracing.go | 43 +++++++++ .../orchestrator/executor/executor_test.go | 9 ++ apps/backend/internal/task/dto/dto.go | 64 ++++++++++++++ .../task/handlers/process_handlers_test.go | 9 ++ .../task/handlers/repository_handlers.go | 21 +++-- .../task/handlers/task_http_handlers.go | 6 +- .../task/handlers/task_ws_handlers.go | 6 +- .../task/handlers/workflow_handlers.go | 15 ++++ .../internal/task/repository/interface.go | 3 + .../task/repository/sqlite/message.go | 4 + .../repository/sqlite/repository_entity.go | 40 +++++++++ .../task/repository/sqlite/session.go | 88 ++++++++++++++----- .../internal/task/repository/sqlite/task.go | 5 ++ .../task/repository/sqlite/task_repository.go | 54 ++++++++++++ .../task/service/service_resources.go | 5 ++ .../internal/task/service/service_tasks.go | 39 ++++---- .../workflow/controller/controller.go | 9 ++ .../internal/workflow/handlers/handlers.go | 11 +++ .../internal/workflow/repository/sqlite.go | 18 ++++ .../internal/workflow/service/service.go | 10 +++ apps/web/app/actions/workspaces.ts | 13 +++ apps/web/app/page.tsx | 58 ++++++------ apps/web/app/s/[sessionId]/page.tsx | 20 ++--- .../[id]/profile/[profileId]/page.tsx | 18 +++- .../settings/executors/[profileId]/page.tsx | 27 +++++- .../settings/executors/new/[type]/page.tsx | 37 ++++++-- apps/web/app/tasks/page.tsx | 23 +++-- 29 files changed, 543 insertions(+), 121 deletions(-) create mode 100644 apps/backend/internal/common/httpmw/tracing.go diff --git a/apps/backend/cmd/kandev/helpers.go b/apps/backend/cmd/kandev/helpers.go index e899594cf2..7630dc3344 100644 --- a/apps/backend/cmd/kandev/helpers.go +++ b/apps/backend/cmd/kandev/helpers.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/agent/lifecycle" agentsettingscontroller "github.com/kandev/kandev/internal/agent/settings/controller" agentsettingshandlers "github.com/kandev/kandev/internal/agent/settings/handlers" + "github.com/kandev/kandev/internal/agentctl/tracing" analyticshandlers "github.com/kandev/kandev/internal/analytics/handlers" analyticsrepository "github.com/kandev/kandev/internal/analytics/repository" "github.com/kandev/kandev/internal/clarification" @@ -392,6 +393,13 @@ func runGracefulShutdown( stopLifecycleManager(lifecycleMgr, log) runCleanups() + // Flush pending OTel spans before exit + traceCtx, traceCancel := context.WithTimeout(context.Background(), 5*time.Second) + if err := tracing.Shutdown(traceCtx); err != nil { + log.Error("Tracer shutdown error", zap.Error(err)) + } + traceCancel() + log.Info("Kandev stopped") _ = log.Sync() } diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index fb6a0d9043..791bb38a83 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -389,6 +389,7 @@ func buildHTTPServer( } router := gin.New() router.Use(httpmw.RequestLogger(log, "kandev")) + router.Use(httpmw.OtelTracing("kandev")) router.Use(gin.Recovery()) router.Use(corsMiddleware()) diff --git a/apps/backend/internal/common/httpmw/tracing.go b/apps/backend/internal/common/httpmw/tracing.go new file mode 100644 index 0000000000..9db0d1194a --- /dev/null +++ b/apps/backend/internal/common/httpmw/tracing.go @@ -0,0 +1,43 @@ +package httpmw + +import ( + "fmt" + + "github.com/gin-gonic/gin" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/codes" + semconv "go.opentelemetry.io/otel/semconv/v1.26.0" + + "github.com/kandev/kandev/internal/agentctl/tracing" +) + +// OtelTracing creates a Gin middleware that wraps each request in an OTel span. +// When tracing is disabled (no OTEL_EXPORTER_OTLP_ENDPOINT), this is a no-op. +func OtelTracing(serverName string) gin.HandlerFunc { + tracer := tracing.Tracer(serverName) + + return func(c *gin.Context) { + path := c.FullPath() + if path == "" { + path = c.Request.URL.Path + } + spanName := fmt.Sprintf("%s %s", c.Request.Method, path) + + ctx, span := tracer.Start(c.Request.Context(), spanName) + defer span.End() + + c.Request = c.Request.WithContext(ctx) + c.Next() + + status := c.Writer.Status() + span.SetAttributes( + semconv.HTTPRequestMethodKey.String(c.Request.Method), + semconv.HTTPRouteKey.String(path), + semconv.HTTPResponseStatusCodeKey.Int(status), + attribute.Int("http.response.size", c.Writer.Size()), + ) + if status >= 500 { + span.SetStatus(codes.Error, fmt.Sprintf("HTTP %d", status)) + } + } +} diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index 08e818d26d..a108f9e928 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -228,6 +228,9 @@ func (m *mockRepository) GetTaskRepository(ctx context.Context, id string) (*mod func (m *mockRepository) ListTaskRepositories(ctx context.Context, taskID string) ([]*models.TaskRepository, error) { return nil, nil } +func (m *mockRepository) ListTaskRepositoriesByTaskIDs(_ context.Context, _ []string) (map[string][]*models.TaskRepository, error) { + return make(map[string][]*models.TaskRepository), nil +} func (m *mockRepository) UpdateTaskRepository(ctx context.Context, taskRepo *models.TaskRepository) error { return nil } @@ -350,6 +353,9 @@ func (m *mockRepository) UpdateSessionReviewStatus(ctx context.Context, sessionI func (m *mockRepository) ListTaskSessionWorktrees(ctx context.Context, sessionID string) ([]*models.TaskSessionWorktree, error) { return nil, nil } +func (m *mockRepository) ListWorktreesBySessionIDs(_ context.Context, _ []string) (map[string][]*models.TaskSessionWorktree, error) { + return make(map[string][]*models.TaskSessionWorktree), nil +} func (m *mockRepository) DeleteTaskSessionWorktree(ctx context.Context, id string) error { return nil } func (m *mockRepository) DeleteTaskSessionWorktreesBySession(ctx context.Context, sessionID string) error { return nil @@ -407,6 +413,9 @@ func (m *mockRepository) DeleteRepositoryScript(ctx context.Context, id string) func (m *mockRepository) ListRepositoryScripts(ctx context.Context, repositoryID string) ([]*models.RepositoryScript, error) { return nil, nil } +func (m *mockRepository) ListScriptsByRepositoryIDs(_ context.Context, _ []string) (map[string][]*models.RepositoryScript, error) { + return make(map[string][]*models.RepositoryScript), nil +} // Executor operations func (m *mockRepository) CreateExecutor(ctx context.Context, executor *models.Executor) error { diff --git a/apps/backend/internal/task/dto/dto.go b/apps/backend/internal/task/dto/dto.go index 208da33324..e6b86e7ee8 100644 --- a/apps/backend/internal/task/dto/dto.go +++ b/apps/backend/internal/task/dto/dto.go @@ -167,6 +167,39 @@ type TaskSessionDTO struct { ReviewStatus *string `json:"review_status,omitempty"` } +// TaskSessionSummaryDTO is a lightweight version of TaskSessionDTO without snapshot fields. +// Used for list endpoints where snapshots are not needed, reducing response size by ~40-60%. +type TaskSessionSummaryDTO struct { + ID string `json:"id"` + TaskID string `json:"task_id"` + AgentExecutionID string `json:"agent_execution_id,omitempty"` + ContainerID string `json:"container_id,omitempty"` + AgentProfileID string `json:"agent_profile_id,omitempty"` + ExecutorID string `json:"executor_id,omitempty"` + EnvironmentID string `json:"environment_id,omitempty"` + RepositoryID string `json:"repository_id,omitempty"` + BaseBranch string `json:"base_branch,omitempty"` + WorktreeID string `json:"worktree_id,omitempty"` + WorktreePath string `json:"worktree_path,omitempty"` + WorktreeBranch string `json:"worktree_branch,omitempty"` + State models.TaskSessionState `json:"state"` + ErrorMessage string `json:"error_message,omitempty"` + Metadata map[string]interface{} `json:"metadata,omitempty"` + StartedAt time.Time `json:"started_at"` + CompletedAt *time.Time `json:"completed_at,omitempty"` + UpdatedAt time.Time `json:"updated_at"` + IsPrimary bool `json:"is_primary"` + IsPassthrough bool `json:"is_passthrough"` + WorkflowStepID *string `json:"workflow_step_id,omitempty"` + ReviewStatus *string `json:"review_status,omitempty"` +} + +// ListTaskSessionSummariesResponse is the list response using summary DTOs. +type ListTaskSessionSummariesResponse struct { + Sessions []TaskSessionSummaryDTO `json:"sessions"` + Total int `json:"total"` +} + type GetTaskSessionResponse struct { Session TaskSessionDTO `json:"session"` } @@ -455,6 +488,37 @@ func FromTaskWithSessionInfo(task *models.Task, primarySessionID *string, sessio } } +// FromTaskSessionSummary converts a session model to a summary DTO (no snapshot fields). +func FromTaskSessionSummary(session *models.TaskSession) TaskSessionSummaryDTO { + result := TaskSessionSummaryDTO{ + ID: session.ID, + TaskID: session.TaskID, + AgentExecutionID: session.AgentExecutionID, + ContainerID: session.ContainerID, + AgentProfileID: session.AgentProfileID, + ExecutorID: session.ExecutorID, + EnvironmentID: session.EnvironmentID, + RepositoryID: session.RepositoryID, + BaseBranch: session.BaseBranch, + State: session.State, + ErrorMessage: session.ErrorMessage, + Metadata: session.Metadata, + StartedAt: session.StartedAt, + CompletedAt: session.CompletedAt, + UpdatedAt: session.UpdatedAt, + IsPrimary: session.IsPrimary, + IsPassthrough: session.IsPassthrough, + WorkflowStepID: session.WorkflowStepID, + ReviewStatus: session.ReviewStatus, + } + if len(session.Worktrees) > 0 { + result.WorktreeID = session.Worktrees[0].WorktreeID + result.WorktreePath = session.Worktrees[0].WorktreePath + result.WorktreeBranch = session.Worktrees[0].WorktreeBranch + } + return result +} + func FromTaskSession(session *models.TaskSession) TaskSessionDTO { result := TaskSessionDTO{ ID: session.ID, diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index 46ff959c8d..8de7f91f22 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -90,6 +90,9 @@ func (m *mockRepository) GetTaskRepository(ctx context.Context, id string) (*mod func (m *mockRepository) ListTaskRepositories(ctx context.Context, taskID string) ([]*models.TaskRepository, error) { return nil, nil } +func (m *mockRepository) ListTaskRepositoriesByTaskIDs(_ context.Context, _ []string) (map[string][]*models.TaskRepository, error) { + return make(map[string][]*models.TaskRepository), nil +} func (m *mockRepository) UpdateTaskRepository(ctx context.Context, taskRepo *models.TaskRepository) error { return nil } @@ -221,6 +224,9 @@ func (m *mockRepository) CreateTaskSessionWorktree(ctx context.Context, sessionW func (m *mockRepository) ListTaskSessionWorktrees(ctx context.Context, sessionID string) ([]*models.TaskSessionWorktree, error) { return nil, nil } +func (m *mockRepository) ListWorktreesBySessionIDs(_ context.Context, _ []string) (map[string][]*models.TaskSessionWorktree, error) { + return make(map[string][]*models.TaskSessionWorktree), nil +} func (m *mockRepository) DeleteTaskSessionWorktree(ctx context.Context, id string) error { return nil } @@ -260,6 +266,9 @@ func (m *mockRepository) ListRepositoryScripts(ctx context.Context, repositoryID } return m.scriptsByRepo[repositoryID], nil } +func (m *mockRepository) ListScriptsByRepositoryIDs(_ context.Context, _ []string) (map[string][]*models.RepositoryScript, error) { + return make(map[string][]*models.RepositoryScript), nil +} func (m *mockRepository) CreateExecutor(ctx context.Context, executor *models.Executor) error { return nil } diff --git a/apps/backend/internal/task/handlers/repository_handlers.go b/apps/backend/internal/task/handlers/repository_handlers.go index 7d3df16250..086f782685 100644 --- a/apps/backend/internal/task/handlers/repository_handlers.go +++ b/apps/backend/internal/task/handlers/repository_handlers.go @@ -81,15 +81,22 @@ func (h *RepositoryHandlers) httpListRepositories(c *gin.Context) { Repositories: make([]dto.RepositoryDTO, 0, len(repositories)), Total: len(repositories), } + + var scriptsByRepo map[string][]*models.RepositoryScript + if includeScripts && len(repositories) > 0 { + repoIDs := make([]string, len(repositories)) + for i, r := range repositories { + repoIDs[i] = r.ID + } + scriptsByRepo, _ = h.service.ListScriptsByRepositoryIDs(c.Request.Context(), repoIDs) + } + for _, repository := range repositories { repoDTO := dto.FromRepository(repository) - if includeScripts { - scripts, scriptsErr := h.service.ListRepositoryScripts(c.Request.Context(), repository.ID) - if scriptsErr == nil { - repoDTO.Scripts = make([]dto.RepositoryScriptDTO, 0, len(scripts)) - for _, script := range scripts { - repoDTO.Scripts = append(repoDTO.Scripts, dto.FromRepositoryScript(script)) - } + if scripts, ok := scriptsByRepo[repository.ID]; ok { + repoDTO.Scripts = make([]dto.RepositoryScriptDTO, 0, len(scripts)) + for _, script := range scripts { + repoDTO.Scripts = append(repoDTO.Scripts, dto.FromRepositoryScript(script)) } } resp.Repositories = append(resp.Repositories, repoDTO) diff --git a/apps/backend/internal/task/handlers/task_http_handlers.go b/apps/backend/internal/task/handlers/task_http_handlers.go index e9c1048667..3a01be2547 100644 --- a/apps/backend/internal/task/handlers/task_http_handlers.go +++ b/apps/backend/internal/task/handlers/task_http_handlers.go @@ -131,11 +131,11 @@ func (h *TaskHandlers) httpListTaskSessions(c *gin.Context) { handleNotFound(c, h.logger, err, "task sessions not found") return } - sessionDTOs := make([]dto.TaskSessionDTO, 0, len(sessions)) + sessionDTOs := make([]dto.TaskSessionSummaryDTO, 0, len(sessions)) for _, session := range sessions { - sessionDTOs = append(sessionDTOs, dto.FromTaskSession(session)) + sessionDTOs = append(sessionDTOs, dto.FromTaskSessionSummary(session)) } - c.JSON(http.StatusOK, dto.ListTaskSessionsResponse{ + c.JSON(http.StatusOK, dto.ListTaskSessionSummariesResponse{ Sessions: sessionDTOs, Total: len(sessionDTOs), }) diff --git a/apps/backend/internal/task/handlers/task_ws_handlers.go b/apps/backend/internal/task/handlers/task_ws_handlers.go index f612d00d07..b61230d80f 100644 --- a/apps/backend/internal/task/handlers/task_ws_handlers.go +++ b/apps/backend/internal/task/handlers/task_ws_handlers.go @@ -32,11 +32,11 @@ func (h *TaskHandlers) doListTaskSessions(ctx context.Context, msg *ws.Message, h.logger.Error("failed to list task sessions", zap.Error(err)) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "Failed to list task sessions", nil) } - sessionDTOs := make([]dto.TaskSessionDTO, 0, len(sessions)) + sessionDTOs := make([]dto.TaskSessionSummaryDTO, 0, len(sessions)) for _, session := range sessions { - sessionDTOs = append(sessionDTOs, dto.FromTaskSession(session)) + sessionDTOs = append(sessionDTOs, dto.FromTaskSessionSummary(session)) } - resp := dto.ListTaskSessionsResponse{ + resp := dto.ListTaskSessionSummariesResponse{ Sessions: sessionDTOs, Total: len(sessionDTOs), } diff --git a/apps/backend/internal/task/handlers/workflow_handlers.go b/apps/backend/internal/task/handlers/workflow_handlers.go index 8e464ba088..b722fa7cac 100644 --- a/apps/backend/internal/task/handlers/workflow_handlers.go +++ b/apps/backend/internal/task/handlers/workflow_handlers.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net/http" + "strconv" "github.com/gin-gonic/gin" @@ -199,6 +200,8 @@ func (h *WorkflowHandlers) httpGetWorkflowSnapshot(c *gin.Context) { return } + tasks = applyTaskLimit(c, tasks) + taskDTOs, err := h.convertTasksWithPrimarySessions(c.Request.Context(), tasks) if err != nil { handleNotFound(c, h.logger, err, "workflow not found") @@ -259,6 +262,8 @@ func (h *WorkflowHandlers) httpGetWorkspaceSnapshot(c *gin.Context) { return } + tasks = applyTaskLimit(c, tasks) + taskDTOs, err := h.convertTasksWithPrimarySessions(c.Request.Context(), tasks) if err != nil { handleNotFound(c, h.logger, err, "workflow not found") @@ -272,6 +277,16 @@ func (h *WorkflowHandlers) httpGetWorkspaceSnapshot(c *gin.Context) { }) } +// applyTaskLimit truncates tasks if a task_limit query param is set. +func applyTaskLimit(c *gin.Context, tasks []*models.Task) []*models.Task { + if limitStr := c.Query("task_limit"); limitStr != "" { + if limit, err := strconv.Atoi(limitStr); err == nil && limit > 0 && limit < len(tasks) { + return tasks[:limit] + } + } + return tasks +} + // WS handlers func (h *WorkflowHandlers) wsListWorkflows(ctx context.Context, msg *ws.Message) (*ws.Message, error) { diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 8aa6f13181..47a65a8e94 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -36,6 +36,7 @@ type Repository interface { CreateTaskRepository(ctx context.Context, taskRepo *models.TaskRepository) error GetTaskRepository(ctx context.Context, id string) (*models.TaskRepository, error) ListTaskRepositories(ctx context.Context, taskID string) ([]*models.TaskRepository, error) + ListTaskRepositoriesByTaskIDs(ctx context.Context, taskIDs []string) (map[string][]*models.TaskRepository, error) UpdateTaskRepository(ctx context.Context, taskRepo *models.TaskRepository) error DeleteTaskRepository(ctx context.Context, id string) error DeleteTaskRepositoriesByTask(ctx context.Context, taskID string) error @@ -97,6 +98,7 @@ type Repository interface { // Task Session Worktree operations CreateTaskSessionWorktree(ctx context.Context, sessionWorktree *models.TaskSessionWorktree) error ListTaskSessionWorktrees(ctx context.Context, sessionID string) ([]*models.TaskSessionWorktree, error) + ListWorktreesBySessionIDs(ctx context.Context, sessionIDs []string) (map[string][]*models.TaskSessionWorktree, error) DeleteTaskSessionWorktree(ctx context.Context, id string) error DeleteTaskSessionWorktreesBySession(ctx context.Context, sessionID string) error @@ -125,6 +127,7 @@ type Repository interface { UpdateRepositoryScript(ctx context.Context, script *models.RepositoryScript) error DeleteRepositoryScript(ctx context.Context, id string) error ListRepositoryScripts(ctx context.Context, repositoryID string) ([]*models.RepositoryScript, error) + ListScriptsByRepositoryIDs(ctx context.Context, repoIDs []string) (map[string][]*models.RepositoryScript, error) // Executor operations CreateExecutor(ctx context.Context, executor *models.Executor) error diff --git a/apps/backend/internal/task/repository/sqlite/message.go b/apps/backend/internal/task/repository/sqlite/message.go index 69a5003dda..fdd97b7fd5 100644 --- a/apps/backend/internal/task/repository/sqlite/message.go +++ b/apps/backend/internal/task/repository/sqlite/message.go @@ -9,6 +9,7 @@ import ( "github.com/google/uuid" + "github.com/kandev/kandev/internal/agentctl/tracing" "github.com/kandev/kandev/internal/db/dialect" "github.com/kandev/kandev/internal/task/models" ) @@ -119,6 +120,9 @@ func (r *Repository) ListMessages(ctx context.Context, sessionID string) ([]*mod // ListMessagesPaginated returns messages for a session ordered by creation time with pagination. func (r *Repository) ListMessagesPaginated(ctx context.Context, sessionID string, opts models.ListMessagesOptions) ([]*models.Message, bool, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListMessagesPaginated") + defer span.End() + limit := opts.Limit if limit < 0 { limit = 0 diff --git a/apps/backend/internal/task/repository/sqlite/repository_entity.go b/apps/backend/internal/task/repository/sqlite/repository_entity.go index d237910dde..6eef4a4365 100644 --- a/apps/backend/internal/task/repository/sqlite/repository_entity.go +++ b/apps/backend/internal/task/repository/sqlite/repository_entity.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "strings" "time" "github.com/google/uuid" @@ -179,6 +180,45 @@ func (r *Repository) DeleteRepositoryScript(ctx context.Context, id string) erro return nil } +// ListScriptsByRepositoryIDs returns all scripts for the given repository IDs, +// grouped by repository ID. This eliminates N+1 queries when loading scripts for multiple repos. +func (r *Repository) ListScriptsByRepositoryIDs(ctx context.Context, repoIDs []string) (map[string][]*models.RepositoryScript, error) { + result := make(map[string][]*models.RepositoryScript, len(repoIDs)) + if len(repoIDs) == 0 { + return result, nil + } + + placeholders := make([]string, len(repoIDs)) + args := make([]interface{}, len(repoIDs)) + for i, id := range repoIDs { + placeholders[i] = "?" + args[i] = id + } + + query := fmt.Sprintf(` + SELECT id, repository_id, name, command, position, created_at, updated_at + FROM repository_scripts + WHERE repository_id IN (%s) + ORDER BY position + `, strings.Join(placeholders, ",")) + + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + script := &models.RepositoryScript{} + err := rows.Scan(&script.ID, &script.RepositoryID, &script.Name, &script.Command, &script.Position, &script.CreatedAt, &script.UpdatedAt) + if err != nil { + return nil, err + } + result[script.RepositoryID] = append(result[script.RepositoryID], script) + } + return result, rows.Err() +} + // ListRepositoryScripts returns all scripts for a repository func (r *Repository) ListRepositoryScripts(ctx context.Context, repositoryID string) ([]*models.RepositoryScript, error) { rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index e3808e3abf..fa7f94ad98 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -11,6 +11,7 @@ import ( "github.com/google/uuid" + "github.com/kandev/kandev/internal/agentctl/tracing" "github.com/kandev/kandev/internal/db/dialect" "github.com/kandev/kandev/internal/task/models" ) @@ -122,6 +123,8 @@ func (r *Repository) CompleteTurn(ctx context.Context, id string) error { // ListTurnsBySession returns all turns for a session ordered by start time func (r *Repository) ListTurnsBySession(ctx context.Context, sessionID string) ([]*models.Turn, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTurnsBySession") + defer span.End() rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` SELECT id, task_session_id, task_id, started_at, completed_at, metadata, created_at, updated_at FROM task_session_turns WHERE task_session_id = ? ORDER BY started_at ASC @@ -421,6 +424,8 @@ func (r *Repository) ClearSessionExecutionID(ctx context.Context, id string) err // ListTaskSessions returns all agent sessions for a task func (r *Repository) ListTaskSessions(ctx context.Context, taskID string) ([]*models.TaskSession, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTaskSessions") + defer span.End() rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` SELECT id, task_id, agent_execution_id, container_id, agent_profile_id, executor_id, executor_profile_id, environment_id, repository_id, base_branch, @@ -438,18 +443,13 @@ func (r *Repository) ListTaskSessions(ctx context.Context, taskID string) ([]*mo if err != nil { return nil, err } - for _, session := range sessions { - worktrees, err := r.ListTaskSessionWorktrees(ctx, session.ID) - if err != nil { - return nil, fmt.Errorf("failed to load session worktrees: %w", err) - } - session.Worktrees = worktrees - } - return sessions, nil + return r.loadWorktreesBatch(ctx, sessions) } // ListActiveTaskSessions returns all active agent sessions across all tasks func (r *Repository) ListActiveTaskSessions(ctx context.Context) ([]*models.TaskSession, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListActiveTaskSessions") + defer span.End() rows, err := r.ro.QueryContext(ctx, ` SELECT id, task_id, agent_execution_id, container_id, agent_profile_id, executor_id, executor_profile_id, environment_id, repository_id, base_branch, @@ -467,14 +467,7 @@ func (r *Repository) ListActiveTaskSessions(ctx context.Context) ([]*models.Task if err != nil { return nil, err } - for _, session := range sessions { - worktrees, err := r.ListTaskSessionWorktrees(ctx, session.ID) - if err != nil { - return nil, fmt.Errorf("failed to load session worktrees: %w", err) - } - session.Worktrees = worktrees - } - return sessions, nil + return r.loadWorktreesBatch(ctx, sessions) } // ListActiveTaskSessionsByTaskID returns all active agent sessions for a specific task @@ -496,12 +489,24 @@ func (r *Repository) ListActiveTaskSessionsByTaskID(ctx context.Context, taskID if err != nil { return nil, err } + return r.loadWorktreesBatch(ctx, sessions) +} + +// loadWorktreesBatch loads worktrees for multiple sessions in a single query. +func (r *Repository) loadWorktreesBatch(ctx context.Context, sessions []*models.TaskSession) ([]*models.TaskSession, error) { + if len(sessions) == 0 { + return sessions, nil + } + sessionIDs := make([]string, len(sessions)) + for i, s := range sessions { + sessionIDs[i] = s.ID + } + worktreeMap, err := r.ListWorktreesBySessionIDs(ctx, sessionIDs) + if err != nil { + return nil, fmt.Errorf("failed to batch-load session worktrees: %w", err) + } for _, session := range sessions { - worktrees, err := r.ListTaskSessionWorktrees(ctx, session.ID) - if err != nil { - return nil, fmt.Errorf("failed to load session worktrees: %w", err) - } - session.Worktrees = worktrees + session.Worktrees = worktreeMap[session.ID] } return sessions, nil } @@ -713,6 +718,47 @@ func (r *Repository) ListTaskSessionWorktrees(ctx context.Context, sessionID str return worktrees, rows.Err() } +// ListWorktreesBySessionIDs returns all worktrees for the given session IDs, +// grouped by session ID. This eliminates N+1 queries when loading worktrees for multiple sessions. +func (r *Repository) ListWorktreesBySessionIDs(ctx context.Context, sessionIDs []string) (map[string][]*models.TaskSessionWorktree, error) { + result := make(map[string][]*models.TaskSessionWorktree, len(sessionIDs)) + if len(sessionIDs) == 0 { + return result, nil + } + + placeholders := make([]string, len(sessionIDs)) + args := make([]interface{}, len(sessionIDs)) + for i, id := range sessionIDs { + placeholders[i] = "?" + args[i] = id + } + + query := fmt.Sprintf(` + SELECT tsw.id, tsw.session_id, tsw.worktree_id, tsw.repository_id, tsw.position, + tsw.worktree_path, tsw.worktree_branch, tsw.created_at + FROM task_session_worktrees tsw + WHERE tsw.session_id IN (%s) + ORDER BY tsw.position ASC, tsw.created_at ASC + `, strings.Join(placeholders, ",")) + + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + var wt models.TaskSessionWorktree + err := rows.Scan(&wt.ID, &wt.SessionID, &wt.WorktreeID, &wt.RepositoryID, + &wt.Position, &wt.WorktreePath, &wt.WorktreeBranch, &wt.CreatedAt) + if err != nil { + return nil, err + } + result[wt.SessionID] = append(result[wt.SessionID], &wt) + } + return result, rows.Err() +} + func (r *Repository) DeleteTaskSessionWorktree(ctx context.Context, id string) error { result, err := r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM task_session_worktrees WHERE id = ?`), id) if err != nil { diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index 1d77c5feb5..ba62eefaf4 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -9,6 +9,7 @@ import ( "github.com/google/uuid" + "github.com/kandev/kandev/internal/agentctl/tracing" "github.com/kandev/kandev/internal/db/dialect" "github.com/kandev/kandev/internal/task/models" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -111,6 +112,8 @@ func (r *Repository) DeleteTask(ctx context.Context, id string) error { // ListTasks returns all non-archived tasks for a workflow func (r *Repository) ListTasks(ctx context.Context, workflowID string) ([]*models.Task, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTasks") + defer span.End() rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` SELECT id, workspace_id, workflow_id, workflow_step_id, title, description, state, priority, position, metadata, archived_at, created_at, updated_at FROM tasks @@ -164,6 +167,8 @@ func (r *Repository) ListTasksByWorkflowStep(ctx context.Context, workflowStepID // If query is non-empty, filters by task title, description, repository name, or repository path // If includeArchived is false, archived tasks are excluded func (r *Repository) ListTasksByWorkspace(ctx context.Context, workspaceID string, query string, page, pageSize int, includeArchived bool) ([]*models.Task, int, error) { + ctx, span := tracing.Tracer("kandev-db").Start(ctx, "db.ListTasksByWorkspace") + defer span.End() // Calculate offset offset := (page - 1) * pageSize if offset < 0 { diff --git a/apps/backend/internal/task/repository/sqlite/task_repository.go b/apps/backend/internal/task/repository/sqlite/task_repository.go index fb4400b21e..4a4cec1998 100644 --- a/apps/backend/internal/task/repository/sqlite/task_repository.go +++ b/apps/backend/internal/task/repository/sqlite/task_repository.go @@ -5,6 +5,7 @@ import ( "database/sql" "encoding/json" "fmt" + "strings" "time" "github.com/google/uuid" @@ -142,6 +143,59 @@ func (r *Repository) DeleteTaskRepository(ctx context.Context, id string) error return nil } +// ListTaskRepositoriesByTaskIDs returns all repository links for the given task IDs, +// grouped by task ID. This eliminates N+1 queries when loading repositories for multiple tasks. +func (r *Repository) ListTaskRepositoriesByTaskIDs(ctx context.Context, taskIDs []string) (map[string][]*models.TaskRepository, error) { + result := make(map[string][]*models.TaskRepository, len(taskIDs)) + if len(taskIDs) == 0 { + return result, nil + } + + placeholders := make([]string, len(taskIDs)) + args := make([]interface{}, len(taskIDs)) + for i, id := range taskIDs { + placeholders[i] = "?" + args[i] = id + } + + query := fmt.Sprintf(` + SELECT id, task_id, repository_id, base_branch, position, metadata, created_at, updated_at + FROM task_repositories + WHERE task_id IN (%s) + ORDER BY position ASC, created_at ASC + `, strings.Join(placeholders, ",")) + + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(query), args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + for rows.Next() { + taskRepo := &models.TaskRepository{} + var metadataJSON string + if err := rows.Scan( + &taskRepo.ID, + &taskRepo.TaskID, + &taskRepo.RepositoryID, + &taskRepo.BaseBranch, + &taskRepo.Position, + &metadataJSON, + &taskRepo.CreatedAt, + &taskRepo.UpdatedAt, + ); err != nil { + return nil, err + } + if metadataJSON != "" && metadataJSON != "{}" { + if err := json.Unmarshal([]byte(metadataJSON), &taskRepo.Metadata); err != nil { + return nil, fmt.Errorf("failed to deserialize task repository metadata: %w", err) + } + } + result[taskRepo.TaskID] = append(result[taskRepo.TaskID], taskRepo) + } + return result, rows.Err() +} + // DeleteTaskRepositoriesByTask deletes all repository links for a task func (r *Repository) DeleteTaskRepositoriesByTask(ctx context.Context, taskID string) error { _, err := r.db.ExecContext(ctx, r.db.Rebind(`DELETE FROM task_repositories WHERE task_id = ?`), taskID) diff --git a/apps/backend/internal/task/service/service_resources.go b/apps/backend/internal/task/service/service_resources.go index 8377da103f..ea21efdc5c 100644 --- a/apps/backend/internal/task/service/service_resources.go +++ b/apps/backend/internal/task/service/service_resources.go @@ -403,6 +403,11 @@ func (s *Service) ListRepositoryScripts(ctx context.Context, repositoryID string return s.repo.ListRepositoryScripts(ctx, repositoryID) } +// ListScriptsByRepositoryIDs returns scripts for multiple repositories in a single query. +func (s *Service) ListScriptsByRepositoryIDs(ctx context.Context, repoIDs []string) (map[string][]*models.RepositoryScript, error) { + return s.repo.ListScriptsByRepositoryIDs(ctx, repoIDs) +} + // Executor operations func (s *Service) CreateExecutor(ctx context.Context, req *CreateExecutorRequest) (*models.Executor, error) { diff --git a/apps/backend/internal/task/service/service_tasks.go b/apps/backend/internal/task/service/service_tasks.go index 4003951416..19425bf933 100644 --- a/apps/backend/internal/task/service/service_tasks.go +++ b/apps/backend/internal/task/service/service_tasks.go @@ -492,14 +492,8 @@ func (s *Service) ListTasks(ctx context.Context, workflowID string) ([]*models.T return nil, err } - // Load repositories for each task - for _, task := range tasks { - repos, err := s.repo.ListTaskRepositories(ctx, task.ID) - if err != nil { - s.logger.Error("failed to list task repositories", zap.String("task_id", task.ID), zap.Error(err)) - } else { - task.Repositories = repos - } + if err := s.loadTaskRepositoriesBatch(ctx, tasks); err != nil { + s.logger.Error("failed to batch-load task repositories", zap.Error(err)) } return tasks, nil @@ -513,15 +507,28 @@ func (s *Service) ListTasksByWorkspace(ctx context.Context, workspaceID string, return nil, 0, err } - // Load repositories for each task - for _, task := range tasks { - repos, err := s.repo.ListTaskRepositories(ctx, task.ID) - if err != nil { - s.logger.Error("failed to list task repositories", zap.String("task_id", task.ID), zap.Error(err)) - } else { - task.Repositories = repos - } + if err := s.loadTaskRepositoriesBatch(ctx, tasks); err != nil { + s.logger.Error("failed to batch-load task repositories", zap.Error(err)) } return tasks, total, nil } + +// loadTaskRepositoriesBatch loads repositories for multiple tasks in a single query. +func (s *Service) loadTaskRepositoriesBatch(ctx context.Context, tasks []*models.Task) error { + if len(tasks) == 0 { + return nil + } + taskIDs := make([]string, len(tasks)) + for i, t := range tasks { + taskIDs[i] = t.ID + } + repoMap, err := s.repo.ListTaskRepositoriesByTaskIDs(ctx, taskIDs) + if err != nil { + return err + } + for _, task := range tasks { + task.Repositories = repoMap[task.ID] + } + return nil +} diff --git a/apps/backend/internal/workflow/controller/controller.go b/apps/backend/internal/workflow/controller/controller.go index b11aff00c4..73eebd8ddc 100644 --- a/apps/backend/internal/workflow/controller/controller.go +++ b/apps/backend/internal/workflow/controller/controller.go @@ -70,6 +70,15 @@ func (c *Controller) ListStepsByWorkflow(ctx context.Context, req ListStepsReque return &ListStepsResponse{Steps: steps}, nil } +// ListStepsByWorkspace returns all workflow steps for all workflows in a workspace. +func (c *Controller) ListStepsByWorkspace(ctx context.Context, workspaceID string) (*ListStepsResponse, error) { + steps, err := c.svc.ListStepsByWorkspaceID(ctx, workspaceID) + if err != nil { + return nil, err + } + return &ListStepsResponse{Steps: steps}, nil +} + func (c *Controller) GetStep(ctx context.Context, id string) (*GetStepResponse, error) { step, err := c.svc.GetStep(ctx, id) if err != nil { diff --git a/apps/backend/internal/workflow/handlers/handlers.go b/apps/backend/internal/workflow/handlers/handlers.go index 3b0b6b82a4..7aa118cb45 100644 --- a/apps/backend/internal/workflow/handlers/handlers.go +++ b/apps/backend/internal/workflow/handlers/handlers.go @@ -42,6 +42,7 @@ func (h *Handlers) registerHTTP(router *gin.Engine) { // Step routes api.GET("/workflows/:id/workflow/steps", h.httpListStepsByWorkflow) + api.GET("/workspaces/:id/workflow-steps", h.httpListStepsByWorkspace) api.GET("/workflow/steps/:id", h.httpGetStep) api.POST("/workflows/:id/workflow/steps", h.httpCreateStepsFromTemplate) api.POST("/workflow/steps", h.httpCreateStep) @@ -103,6 +104,16 @@ func (h *Handlers) httpListStepsByWorkflow(c *gin.Context) { c.JSON(http.StatusOK, resp) } +func (h *Handlers) httpListStepsByWorkspace(c *gin.Context) { + resp, err := h.controller.ListStepsByWorkspace(c.Request.Context(), c.Param("id")) + if err != nil { + h.logger.Error("failed to list steps by workspace", zap.Error(err)) + c.JSON(http.StatusInternalServerError, gin.H{"error": "Failed to list steps"}) + return + } + c.JSON(http.StatusOK, resp) +} + func (h *Handlers) httpGetStep(c *gin.Context) { resp, err := h.controller.GetStep(c.Request.Context(), c.Param("id")) if err != nil { diff --git a/apps/backend/internal/workflow/repository/sqlite.go b/apps/backend/internal/workflow/repository/sqlite.go index f3715533f5..fdc14440f1 100644 --- a/apps/backend/internal/workflow/repository/sqlite.go +++ b/apps/backend/internal/workflow/repository/sqlite.go @@ -807,6 +807,24 @@ func (r *Repository) ListStepsByWorkflow(ctx context.Context, workflowID string) return r.scanSteps(rows) } +// ListStepsByWorkspaceID returns all workflow steps for all workflows in a workspace. +func (r *Repository) ListStepsByWorkspaceID(ctx context.Context, workspaceID string) ([]*models.WorkflowStep, error) { + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` + SELECT ws.id, ws.workflow_id, ws.name, ws.position, ws.color, ws.prompt, ws.events, + ws.allow_manual_move, ws.is_start_step, ws.auto_archive_after_hours, ws.created_at, ws.updated_at + FROM workflow_steps ws + JOIN workflows w ON ws.workflow_id = w.id + WHERE w.workspace_id = ? + ORDER BY ws.workflow_id, ws.position + `), workspaceID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + + return r.scanSteps(rows) +} + // scanSteps is a helper to scan multiple workflow step rows. func (r *Repository) scanSteps(rows *sql.Rows) ([]*models.WorkflowStep, error) { var result []*models.WorkflowStep diff --git a/apps/backend/internal/workflow/service/service.go b/apps/backend/internal/workflow/service/service.go index bfd3014cb8..2c7fbc1b40 100644 --- a/apps/backend/internal/workflow/service/service.go +++ b/apps/backend/internal/workflow/service/service.go @@ -89,6 +89,16 @@ func (s *Service) ListStepsByWorkflow(ctx context.Context, workflowID string) ([ return steps, nil } +// ListStepsByWorkspaceID returns all workflow steps for all workflows in a workspace. +func (s *Service) ListStepsByWorkspaceID(ctx context.Context, workspaceID string) ([]*models.WorkflowStep, error) { + steps, err := s.repo.ListStepsByWorkspaceID(ctx, workspaceID) + if err != nil { + s.logger.Error("failed to list steps by workspace", zap.String("workspace_id", workspaceID), zap.Error(err)) + return nil, err + } + return steps, nil +} + // GetStep retrieves a workflow step by ID. func (s *Service) GetStep(ctx context.Context, stepID string) (*models.WorkflowStep, error) { step, err := s.repo.GetStep(ctx, stepID) diff --git a/apps/web/app/actions/workspaces.ts b/apps/web/app/actions/workspaces.ts index 82ea170d69..813647301f 100644 --- a/apps/web/app/actions/workspaces.ts +++ b/apps/web/app/actions/workspaces.ts @@ -332,6 +332,19 @@ const transformWorkflowStep = (step: BackendWorkflowStep): WorkflowStep => ({ updated_at: step.updated_at, }); +// Workspace Workflow Steps (batch: all steps for all workflows in a workspace) +export async function listWorkspaceWorkflowStepsAction( + workspaceId: string, +): Promise { + const response = await fetchJson<{ steps: BackendWorkflowStep[] | null }>( + `${apiBaseUrl}/api/v1/workspaces/${workspaceId}/workflow-steps`, + ); + return { + steps: (response.steps ?? []).map(transformWorkflowStep), + total: response.steps?.length ?? 0, + }; +} + // Workflow Steps export async function listWorkflowStepsAction( workflowId: string, diff --git a/apps/web/app/page.tsx b/apps/web/app/page.tsx index cc911dda83..df2bab2b0b 100644 --- a/apps/web/app/page.tsx +++ b/apps/web/app/page.tsx @@ -113,37 +113,37 @@ function buildBaseState( }; } -async function loadSessionMessages( - sessionId: string, - state: Partial, +async function loadSnapshotState( + workflowId: string, + taskId: string | undefined, + sessionId: string | undefined, ): Promise> { - try { - const messagesResponse = await listTaskSessionMessages( - sessionId, - { limit: 50, sort: "desc" }, - { cache: "no-store" }, - ); + const [snapshot, messagesResponse] = await Promise.all([ + fetchWorkflowSnapshot(workflowId, { cache: "no-store" }), + taskId && sessionId + ? listTaskSessionMessages( + sessionId, + { limit: 50, sort: "desc" }, + { cache: "no-store" }, + ).catch(() => null) + : Promise.resolve(null), + ]); + const state: Partial = { ...snapshotToState(snapshot) }; + + if (sessionId && messagesResponse) { const messages = [...(messagesResponse.messages ?? [])].reverse(); - return { - ...state, - messages: { - bySession: { [sessionId]: messages }, - metaBySession: { - [sessionId]: { - isLoading: false, - hasMore: messagesResponse.has_more ?? false, - oldestCursor: messages[0]?.id ?? null, - }, + state.messages = { + bySession: { [sessionId]: messages }, + metaBySession: { + [sessionId]: { + isLoading: false, + hasMore: messagesResponse.has_more ?? false, + oldestCursor: messages[0]?.id ?? null, }, }, }; - } catch (error) { - console.warn( - "Could not SSR messages (client will load via WebSocket):", - error instanceof Error ? error.message : String(error), - ); - return state; } + return state; } export default async function Page({ searchParams }: PageProps) { @@ -216,12 +216,8 @@ export default async function Page({ searchParams }: PageProps) { ); } - const snapshot = await fetchWorkflowSnapshot(workflowId, { cache: "no-store" }); - initialState = { ...initialState, ...snapshotToState(snapshot) }; - - if (taskId && sessionId) { - initialState = await loadSessionMessages(sessionId, initialState); - } + const snapshotState = await loadSnapshotState(workflowId, taskId, sessionId); + initialState = { ...initialState, ...snapshotState }; return ( <> diff --git a/apps/web/app/s/[sessionId]/page.tsx b/apps/web/app/s/[sessionId]/page.tsx index 37db7854cc..d618cf79e0 100644 --- a/apps/web/app/s/[sessionId]/page.tsx +++ b/apps/web/app/s/[sessionId]/page.tsx @@ -240,6 +240,7 @@ async function fetchSessionData(paramSessionId: string): Promise ({ turns: [], total: 0 })), fetchUserSettings({ cache: "no-store" }).catch(() => null), fetchTerminals(paramSessionId).catch(() => []), + listTaskSessionMessages( + paramSessionId, + { limit: 50, sort: "desc" }, + { cache: "no-store" }, + ).catch(() => null as ListMessagesResponse | null), ]); const repositories = repositoriesResponse.repositories ?? []; @@ -267,20 +273,6 @@ async function fetchSessionData(paramSessionId: string): Promise )} {form.isSprites && ( - + )} listWorkflowStepsAction(workflow.id)), - ); - steps = stepsResponses.flatMap((r) => r.steps); + steps = stepsResponse.steps; } } catch (error) { console.error("Failed to load tasks page data:", error);