101 lines
3.5 KiB
Go
101 lines
3.5 KiB
Go
package platform
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"net/http"
|
|
"sort"
|
|
)
|
|
|
|
// Deletion or kind/ownership changes revoke active steps in the same snapshot
|
|
// transaction. Recreating an identical node ID later cannot revive that step.
|
|
func invalidateCanvasAITargets(r *http.Request, tx *sql.Tx, project, workspace, board string, before, after json.RawMessage) error {
|
|
old, next := canvasNodeData(before), canvasNodeData(after)
|
|
removed := []string{}
|
|
for node, data := range old {
|
|
if next[node] == nil || next[node]["kind"] != data["kind"] {
|
|
removed = append(removed, node)
|
|
}
|
|
}
|
|
sort.Strings(removed)
|
|
for _, node := range removed {
|
|
rows, err := tx.QueryContext(r.Context(), `SELECT DISTINCT run_id::text FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND node_id=$4 AND status IN('queued','running') ORDER BY run_id::text`, project, workspace, board, node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
runs := []string{}
|
|
for rows.Next() {
|
|
var id string
|
|
if err = rows.Scan(&id); err != nil {
|
|
break
|
|
}
|
|
runs = append(runs, id)
|
|
}
|
|
if err == nil {
|
|
err = rows.Err()
|
|
}
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, run := range runs {
|
|
var rawPlan json.RawMessage
|
|
err = tx.QueryRowContext(r.Context(), `SELECT plan FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND id=$3 FOR UPDATE`, project, workspace, run).Scan(&rawPlan)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var step string
|
|
err = tx.QueryRowContext(r.Context(), `UPDATE canvas_ai_steps SET status='cancelled',error_code='target_removed',completed_at=now() WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND board_id=$4 AND node_id=$5 AND status IN('queued','running') RETURNING id`, project, workspace, run, board, node).Scan(&step)
|
|
if err == sql.ErrNoRows {
|
|
continue
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = appendAIEvent(r.Context(), tx, project, workspace, run, "step.cancelled", response{"step_id": step, "board_id": board, "node_id": node, "error": "target_removed"}); err != nil {
|
|
return err
|
|
}
|
|
var plan canvasAIPlan
|
|
if err = json.Unmarshal(rawPlan, &plan); err != nil {
|
|
return err
|
|
}
|
|
for _, edge := range plan.Edges {
|
|
if edge.Target == step {
|
|
if err = appendAIEvent(r.Context(), tx, project, workspace, run, "edge.skipped", response{"logical_edge_id": edge.ID, "status": "skipped"}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func invalidateCanvasAISubtree(r *http.Request, tx *sql.Tx, project, workspace, child string) error {
|
|
rows, err := tx.QueryContext(r.Context(), `WITH RECURSIVE descendants(id,depth) AS(SELECT $3::uuid,0 UNION ALL SELECT s.child_board_id,d.depth+1 FROM descendants d JOIN canvas_subgraphs s ON s.project=$1 AND s.workspace_id=$2 AND s.parent_board_id=d.id AND s.archived_at IS NULL WHERE d.depth<8) SELECT m.id::text,COALESCE(m.canvas_document,'{"nodes":[]}'::jsonb) FROM descendants d JOIN materials m ON m.project=$1 AND m.workspace_id=$2 AND m.id=d.id ORDER BY m.id`, project, workspace, child)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
snapshots := []canvasSnapshot{}
|
|
for rows.Next() {
|
|
var snapshot canvasSnapshot
|
|
if err = rows.Scan(&snapshot.BoardID, &snapshot.Document); err != nil {
|
|
break
|
|
}
|
|
snapshots = append(snapshots, snapshot)
|
|
}
|
|
if err == nil {
|
|
err = rows.Err()
|
|
}
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, snapshot := range snapshots {
|
|
if err = invalidateCanvasAITargets(r, tx, project, workspace, snapshot.BoardID, snapshot.Document, json.RawMessage(`{"nodes":[]}`)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|