Add web product Phase 1 skeleton

2026-03-20 00:20:38 +08:00
parent 0355d7a847
commit a7ef1e0154
24 changed files with 2287 additions and 6 deletions
@@ -0,0 +1,211 @@
+package query
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+
+	"ai-workflow-skill/internal/store"
+)
+
+type ReadService struct {
+	db    *sql.DB
+	orch  *store.OrchStore
+	inbox *store.InboxStore
+}
+
+type RunListItem struct {
+	Run        store.Run      `json:"run"`
+	TaskCounts map[string]int `json:"task_counts"`
+	TotalTasks int            `json:"total_tasks"`
+}
+
+type RunDetail struct {
+	Run          store.Run           `json:"run"`
+	TaskCounts   map[string]int      `json:"task_counts"`
+	TotalTasks   int                 `json:"total_tasks"`
+	Tasks        []store.Task        `json:"tasks"`
+	BlockedTasks []store.BlockedTask `json:"blocked_tasks"`
+}
+
+func NewReadService(db *sql.DB) *ReadService {
+	return &ReadService{
+		db:    db,
+		orch:  store.NewOrchStore(db),
+		inbox: store.NewInboxStore(db),
+	}
+}
+
+func (s *ReadService) ListRuns(ctx context.Context) ([]RunListItem, error) {
+	rows, err := s.db.QueryContext(
+		ctx,
+		`SELECT run_id, goal, summary, status, created_at, updated_at
+		 FROM runs
+		 ORDER BY updated_at DESC, created_at DESC`,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("query runs: %w", err)
+	}
+	defer rows.Close()
+
+	var runs []store.Run
+	runIDs := make([]string, 0)
+	for rows.Next() {
+		var (
+			run                store.Run
+			createdAt, updated string
+		)
+		if err := rows.Scan(
+			&run.RunID,
+			&run.Goal,
+			&run.Summary,
+			&run.Status,
+			&createdAt,
+			&updated,
+		); err != nil {
+			return nil, fmt.Errorf("scan run list row: %w", err)
+		}
+
+		run.CreatedAt = parseRFC3339(createdAt)
+		run.UpdatedAt = parseRFC3339(updated)
+		runs = append(runs, run)
+		runIDs = append(runIDs, run.RunID)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate runs: %w", err)
+	}
+
+	countsByRunID, err := s.collectTaskCounts(ctx, runIDs)
+	if err != nil {
+		return nil, err
+	}
+
+	items := make([]RunListItem, 0, len(runs))
+	for _, run := range runs {
+		taskCounts := countsByRunID[run.RunID]
+		if taskCounts == nil {
+			taskCounts = map[string]int{}
+		}
+
+		items = append(items, RunListItem{
+			Run:        run,
+			TaskCounts: taskCounts,
+			TotalTasks: totalTasks(taskCounts),
+		})
+	}
+
+	return items, nil
+}
+
+func (s *ReadService) GetRunDetail(ctx context.Context, runID string) (RunDetail, error) {
+	overview, err := s.orch.GetRunOverview(ctx, runID)
+	if err != nil {
+		return RunDetail{}, err
+	}
+
+	blocked, err := s.orch.ListBlockedTasks(ctx, runID)
+	if err != nil {
+		return RunDetail{}, err
+	}
+
+	return RunDetail{
+		Run:          overview.Run,
+		TaskCounts:   overview.TaskCounts,
+		TotalTasks:   totalTasks(overview.TaskCounts),
+		Tasks:        overview.Tasks,
+		BlockedTasks: blocked,
+	}, nil
+}
+
+func (s *ReadService) ListRunTasks(ctx context.Context, runID string) ([]store.Task, error) {
+	detail, err := s.GetRunDetail(ctx, runID)
+	if err != nil {
+		return nil, err
+	}
+	return detail.Tasks, nil
+}
+
+func (s *ReadService) ListBlockedTasks(ctx context.Context, runID string) ([]store.BlockedTask, error) {
+	return s.orch.ListBlockedTasks(ctx, runID)
+}
+
+func (s *ReadService) GetThreadDetail(ctx context.Context, threadID string) (store.ThreadDetail, error) {
+	return s.inbox.GetThread(ctx, threadID)
+}
+
+func (s *ReadService) collectTaskCounts(ctx context.Context, runIDs []string) (map[string]map[string]int, error) {
+	result := make(map[string]map[string]int, len(runIDs))
+	if len(runIDs) == 0 {
+		return result, nil
+	}
+
+	args := make([]any, 0, len(runIDs))
+	for _, runID := range runIDs {
+		args = append(args, runID)
+	}
+
+	rows, err := s.db.QueryContext(
+		ctx,
+		`SELECT run_id, status, COUNT(*)
+		 FROM tasks
+		 WHERE run_id IN (`+placeholders(len(runIDs))+`)
+		 GROUP BY run_id, status`,
+		args...,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("query task counts for runs: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var (
+			runID  string
+			status string
+			count  int
+		)
+		if err := rows.Scan(&runID, &status, &count); err != nil {
+			return nil, fmt.Errorf("scan run task count: %w", err)
+		}
+		if result[runID] == nil {
+			result[runID] = make(map[string]int)
+		}
+		result[runID][status] = count
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate run task counts: %w", err)
+	}
+
+	return result, nil
+}
+
+func totalTasks(counts map[string]int) int {
+	total := 0
+	for _, count := range counts {
+		total += count
+	}
+	return total
+}
+
+func placeholders(count int) string {
+	if count <= 0 {
+		return ""
+	}
+
+	buf := make([]byte, 0, count*2-1)
+	for i := 0; i < count; i++ {
+		if i > 0 {
+			buf = append(buf, ',')
+		}
+		buf = append(buf, '?')
+	}
+	return string(buf)
+}
+
+func parseRFC3339(value string) time.Time {
+	parsed, err := time.Parse(time.RFC3339Nano, value)
+	if err != nil {
+		return time.Time{}
+	}
+	return parsed
+}