feat: improve sql queries (#260)

This commit is contained in:
Carlos Florêncio
2026-02-21 23:02:52 +00:00
committed by GitHub
parent a23ab5be3a
commit 6e4d22cae0
29 changed files with 543 additions and 121 deletions
+8
View File
@@ -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()
}
+1
View File
@@ -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())
@@ -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))
}
}
}
@@ -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 {
+64
View File
@@ -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,
@@ -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
}
@@ -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)
@@ -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),
})
@@ -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),
}
@@ -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) {
@@ -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
@@ -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
@@ -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(`
@@ -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 {
@@ -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 {
@@ -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)
@@ -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) {
@@ -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
}
@@ -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 {
@@ -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 {
@@ -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
@@ -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)
+13
View File
@@ -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<ListWorkflowStepsResponse> {
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,
+27 -31
View File
@@ -113,37 +113,37 @@ function buildBaseState(
};
}
async function loadSessionMessages(
sessionId: string,
state: Partial<AppState>,
async function loadSnapshotState(
workflowId: string,
taskId: string | undefined,
sessionId: string | undefined,
): Promise<Partial<AppState>> {
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<AppState> = { ...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 (
<>
+6 -14
View File
@@ -240,6 +240,7 @@ async function fetchSessionData(paramSessionId: string): Promise<FetchedSessionD
turnsResponse,
userSettingsResponse,
terminalsResponse,
messagesResponse,
] = await Promise.all([
fetchWorkflowSnapshot(task.workflow_id, { cache: "no-store" }),
listAgents({ cache: "no-store" }),
@@ -251,6 +252,11 @@ async function fetchSessionData(paramSessionId: string): Promise<FetchedSessionD
listSessionTurns(paramSessionId, { cache: "no-store" }).catch(() => ({ 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<FetchedSessionD
closable: t.closable,
}));
let messagesResponse: ListMessagesResponse | null = null;
try {
messagesResponse = await listTaskSessionMessages(
paramSessionId,
{ limit: 50, sort: "desc" },
{ cache: "no-store" },
);
} catch (error) {
console.warn(
"Could not load session messages for SSR:",
error instanceof Error ? error.message : String(error),
);
}
const initialState = buildSessionPageState({
task,
sessionId: paramSessionId,
@@ -435,9 +435,21 @@ function useProfileFormState(executor: Executor, profile: ExecutorProfile) {
: "Runs on the host machine before the agent starts.";
return {
name, setName, prepareScript, setPrepareScript, cleanupScript, setCleanupScript,
envVarRows, addEnvVar, removeEnvVar, updateEnvVar, placeholders,
isRemote, isSprites, spritesSecretId, prepareDesc,
name,
setName,
prepareScript,
setPrepareScript,
cleanupScript,
setCleanupScript,
envVarRows,
addEnvVar,
removeEnvVar,
updateEnvVar,
placeholders,
isRemote,
isSprites,
spritesSecretId,
prepareDesc,
};
}
@@ -337,10 +337,29 @@ function useProfileFormState(executor: Executor, profile: ExecutorProfile) {
: "Runs on the host machine before the agent starts.";
return {
name, setName, mcpPolicy, setMcpPolicy, prepareScript, setPrepareScript,
cleanupScript, setCleanupScript, envVarRows, addEnvVar, removeEnvVar, updateEnvVar,
placeholders, spritesSecretId, setSpritesSecretId, networkPolicyRules, setNetworkPolicyRules,
isRemote, isDocker, isSprites, mcpPolicyError, buildEnvVars, prepareDesc,
name,
setName,
mcpPolicy,
setMcpPolicy,
prepareScript,
setPrepareScript,
cleanupScript,
setCleanupScript,
envVarRows,
addEnvVar,
removeEnvVar,
updateEnvVar,
placeholders,
spritesSecretId,
setSpritesSecretId,
networkPolicyRules,
setNetworkPolicyRules,
isRemote,
isDocker,
isSprites,
mcpPolicyError,
buildEnvVars,
prepareDesc,
};
}
@@ -220,11 +220,33 @@ function useCreateProfileFormState(executorType: string) {
: "Runs on the host machine before the agent starts.";
return {
name, setName, mcpPolicy, setMcpPolicy, prepareScript, setPrepareScript,
cleanupScript, setCleanupScript, envVarRows, addEnvVar, removeEnvVar, updateEnvVar,
placeholders, spritesSecretId, setSpritesSecretId, networkPolicyRules, setNetworkPolicyRules,
dockerfile, setDockerfile, imageTag, setImageTag,
isRemote, isDocker, isSprites, mcpPolicyError, buildEnvVars, prepareDesc,
name,
setName,
mcpPolicy,
setMcpPolicy,
prepareScript,
setPrepareScript,
cleanupScript,
setCleanupScript,
envVarRows,
addEnvVar,
removeEnvVar,
updateEnvVar,
placeholders,
spritesSecretId,
setSpritesSecretId,
networkPolicyRules,
setNetworkPolicyRules,
dockerfile,
setDockerfile,
imageTag,
setImageTag,
isRemote,
isDocker,
isSprites,
mcpPolicyError,
buildEnvVars,
prepareDesc,
};
}
@@ -301,7 +323,10 @@ function CreateProfileForm({
/>
)}
{form.isSprites && (
<NetworkPoliciesCard rules={form.networkPolicyRules} onRulesChange={form.setNetworkPolicyRules} />
<NetworkPoliciesCard
rules={form.networkPolicyRules}
onRulesChange={form.setNetworkPolicyRules}
/>
)}
<EnvVarsCard
rows={form.envVarRows}
+10 -13
View File
@@ -3,7 +3,7 @@ import {
listWorkflowsAction,
listTasksByWorkspaceAction,
listRepositoriesAction,
listWorkflowStepsAction,
listWorkspaceWorkflowStepsAction,
} from "@/app/actions/workspaces";
import { TasksPageClient } from "./tasks-page-client";
import type { Workflow, Task, WorkflowStep, Repository, Workspace } from "@/lib/types/http";
@@ -33,23 +33,20 @@ export default async function TasksPage({
}
if (workspaceId) {
// Fetch all data in parallel
const [workflowsResponse, repositoriesResponse, tasksResponse] = await Promise.all([
listWorkflowsAction(workspaceId),
listRepositoriesAction(workspaceId),
listTasksByWorkspaceAction(workspaceId, 1, 25),
]);
// Fetch all data in parallel (including steps via single batch endpoint)
const [workflowsResponse, repositoriesResponse, tasksResponse, stepsResponse] =
await Promise.all([
listWorkflowsAction(workspaceId),
listRepositoriesAction(workspaceId),
listTasksByWorkspaceAction(workspaceId, 1, 25),
listWorkspaceWorkflowStepsAction(workspaceId),
]);
workflows = workflowsResponse.workflows;
repositories = repositoriesResponse.repositories;
tasks = tasksResponse.tasks;
total = tasksResponse.total;
// Fetch workflow steps for each workflow
const stepsResponses = await Promise.all(
workflows.map((workflow) => listWorkflowStepsAction(workflow.id)),
);
steps = stepsResponses.flatMap((r) => r.steps);
steps = stepsResponse.steps;
}
} catch (error) {
console.error("Failed to load tasks page data:", error);