package store import ( "database/sql" "fmt" "g.fc.run/ai/byre/internal/ids" ) func migrateRunTasks(tx *sql.Tx, _ string) error { rows, err := tx.Query(`SELECT workflow_run FROM tasks WHERE workflow_run != '' GROUP BY workflow_run ORDER BY min(rowid)`) if err != nil { return err } runs := []string{} for rows.Next() { run := "" err = rows.Scan(&run) if err != nil { rows.Close() return err } runs = append(runs, run) } rows.Close() err = rows.Err() if err != nil { return err } for _, run := range runs { id := ids.Derive(ids.Task, run) _, err = tx.Exec(` INSERT INTO tasks (id, project_id, title, branch, worktree_path, base_ref, status, created_by, created_at, closed_at, todo_item, color, workflow_name, workflow_definition_id, workflow_brief, workflow_title, audit) SELECT ?1, f.project_id, f.workflow_title, '', '', '', CASE WHEN EXISTS (SELECT 1 FROM tasks s WHERE s.workflow_run = ?2 AND s.status = 'active') THEN 'active' WHEN EXISTS (SELECT 1 FROM tasks s WHERE s.workflow_run = ?2 AND s.status = 'merged') THEN 'merged' ELSE 'discarded' END, f.created_by, f.created_at, CASE WHEN EXISTS (SELECT 1 FROM tasks s WHERE s.workflow_run = ?2 AND s.status = 'active') THEN NULL ELSE (SELECT max(s.closed_at) FROM tasks s WHERE s.workflow_run = ?2) END, f.todo_item, f.color, f.workflow_name, f.workflow_definition_id, f.workflow_brief, f.workflow_title, f.audit FROM tasks f WHERE f.workflow_run = ?2 ORDER BY f.created_at LIMIT 1`, id, run) if err != nil { return fmt.Errorf("run %s: %w", run, err) } _, err = tx.Exec(` UPDATE events SET task_id = ?1, payload = json_set(payload, '$.id', ?1) WHERE kind = 'task.workflow_finished' AND task_id IN (SELECT id FROM tasks WHERE workflow_run = ?2)`, id, run) if err != nil { return fmt.Errorf("run %s: %w", run, err) } _, err = tx.Exec(` UPDATE tasks SET parent_task_id = ?1, workflow_name = '', workflow_definition_id = NULL, workflow_brief = '', workflow_title = '', audit = '', todo_item = '' WHERE workflow_run = ?2`, id, run) if err != nil { return fmt.Errorf("run %s: %w", run, err) } _, err = tx.Exec(`UPDATE tasks SET state = `+taskStateExpr+` WHERE id = ?`, id) if err != nil { return fmt.Errorf("run %s: %w", run, err) } } _, err = tx.Exec(`ALTER TABLE tasks DROP COLUMN workflow_run`) return err }