71 lines
2.3 KiB
Go
71 lines
2.3 KiB
Go
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
|
|
}
|