206 lines
6.9 KiB
Go
206 lines
6.9 KiB
Go
package platform
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
)
|
|
|
|
func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, project, account string) error {
|
|
if r.Method != http.MethodPost {
|
|
reject(w, 405, "method_not_allowed")
|
|
return nil
|
|
}
|
|
key, id := canvasIdempotencyKey(r), r.PathValue("run")
|
|
if !uuidSlugPattern.MatchString(key) || !uuidSlugPattern.MatchString(id) {
|
|
reject(w, 400, "invalid_ai_request")
|
|
return nil
|
|
}
|
|
raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 4096))
|
|
if err != nil {
|
|
reject(w, 413, "context_too_large")
|
|
return nil
|
|
}
|
|
var input struct {
|
|
ProtocolVersion int `json:"protocol_version,omitempty"`
|
|
RequestID string `json:"request_id,omitempty"`
|
|
OperationID string `json:"operation_id,omitempty"`
|
|
}
|
|
if len(raw) > 0 && (json.Unmarshal(raw, &input) != nil || string(raw) == "null") {
|
|
reject(w, 400, "invalid_ai_request")
|
|
return nil
|
|
}
|
|
var fields map[string]any
|
|
if len(raw) > 0 {
|
|
if json.Unmarshal(raw, &fields) != nil || fields == nil {
|
|
reject(w, 400, "invalid_ai_request")
|
|
return nil
|
|
}
|
|
for key := range fields {
|
|
if key != "protocol_version" && key != "request_id" && key != "operation_id" {
|
|
reject(w, 400, "invalid_ai_request")
|
|
return nil
|
|
}
|
|
}
|
|
}
|
|
canonical := input.ProtocolVersion != 0 || input.RequestID != "" || input.OperationID != ""
|
|
if canonical && (input.ProtocolVersion != 2 || input.RequestID != key || !uuidSlugPattern.MatchString(input.OperationID) || !acceptsCanvasV2(r)) {
|
|
reject(w, 400, "invalid_ai_protocol")
|
|
return nil
|
|
}
|
|
if !canonical && len(fields) > 0 {
|
|
reject(w, 400, "invalid_ai_request")
|
|
return nil
|
|
}
|
|
hash := payloadDigest(map[string]any{})
|
|
if canonical {
|
|
hash = payloadDigest(input)
|
|
}
|
|
workspace, board := r.PathValue("workspace"), r.PathValue("material")
|
|
tx, err := s.db.BeginTx(r.Context(), nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
_, role, err := lockCanvasAIRunRoute(r.Context(), tx, project, workspace, board, account)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var receipt *canvasAIRequest
|
|
if canonical {
|
|
receipt = &canvasAIRequest{Key: key, Hash: hash, Action: "run.cancel", Board: board, Run: id, Operation: input.OperationID}
|
|
if done, e := beginCanvasAIRequest(w, r, tx, project, workspace, account, receipt); done || e != nil {
|
|
return e
|
|
}
|
|
}
|
|
rejectRequest := func(status int, code string) error {
|
|
return rejectCanvasAIRequest(w, r, tx, project, workspace, account, receipt, status, code)
|
|
}
|
|
if role == "viewer" {
|
|
return rejectRequest(403, "forbidden")
|
|
}
|
|
var status string
|
|
var rawPlan json.RawMessage
|
|
err = tx.QueryRowContext(r.Context(), `SELECT status,plan FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND id=$4 FOR UPDATE`, project, workspace, board, id).Scan(&status, &rawPlan)
|
|
if err == sql.ErrNoRows {
|
|
return rejectRequest(404, "not_found")
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var priorHash string
|
|
var prior json.RawMessage
|
|
err = tx.QueryRowContext(r.Context(), `SELECT payload_hash,response_body FROM canvas_ai_mutations WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND account_id=$4 AND action='cancel' AND idempotency_key=$5`, project, workspace, id, account, key).Scan(&priorHash, &prior)
|
|
if err == nil {
|
|
if priorHash != hash {
|
|
reject(w, 409, "idempotency_conflict")
|
|
return nil
|
|
}
|
|
reply(w, 200, prior)
|
|
return nil
|
|
}
|
|
if err != sql.ErrNoRows {
|
|
return err
|
|
}
|
|
var plan canvasAIPlan
|
|
if err = json.Unmarshal(rawPlan, &plan); err != nil {
|
|
return err
|
|
}
|
|
for _, step := range plan.Steps {
|
|
target, ok, e := resolveCanvasAITarget(r.Context(), tx, project, workspace, step)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if !ok {
|
|
target = step.BoardID
|
|
}
|
|
_, role, err := lockCanvasAIRunRoute(r.Context(), tx, project, workspace, target, account)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if role == "viewer" {
|
|
return rejectRequest(403, "forbidden")
|
|
}
|
|
}
|
|
if status == "queued" || status == "running" {
|
|
_, err = tx.ExecContext(r.Context(), `UPDATE canvas_ai_runs SET status='cancelled',cancel_requested=true,completed_at=now() WHERE project=$1 AND workspace_id=$2 AND id=$3`, project, workspace, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
rows, err := tx.QueryContext(r.Context(), `UPDATE canvas_ai_steps SET status='cancelled',error_code='generation_cancelled',completed_at=now() WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND status IN('queued','running') RETURNING id,board_id::text,node_id`, project, workspace, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
events := []response{}
|
|
for rows.Next() {
|
|
var step, board, node string
|
|
if err = rows.Scan(&step, &board, &node); err != nil {
|
|
break
|
|
}
|
|
events = append(events, response{"step_id": step, "board_id": board, "node_id": node, "status": "cancelled"})
|
|
}
|
|
if err == nil {
|
|
err = rows.Err()
|
|
}
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
cancelled := map[string]bool{}
|
|
for _, event := range events {
|
|
cancelled[event["step_id"].(string)] = true
|
|
if err = appendAIEvent(r.Context(), tx, project, workspace, id, "step.cancelled", event); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for _, edge := range plan.Edges {
|
|
if cancelled[edge.Target] {
|
|
if err = appendAIEvent(r.Context(), tx, project, workspace, id, "edge.skipped", response{"logical_edge_id": edge.ID, "status": "skipped"}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
status = "cancelled"
|
|
if err = appendAIEvent(r.Context(), tx, project, workspace, id, "run.completed", response{"status": status}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
rows, err := tx.QueryContext(r.Context(), `SELECT id,COALESCE(applied_board_id,board_id)::text,node_id,status,applied_version FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 ORDER BY ordinal`, project, workspace, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
applied, cancelledSteps := []any{}, []any{}
|
|
for rows.Next() {
|
|
var step, board, node, state string
|
|
var version sql.NullInt64
|
|
if err = rows.Scan(&step, &board, &node, &state, &version); err != nil {
|
|
break
|
|
}
|
|
if version.Valid {
|
|
applied = append(applied, response{"step_id": step, "board_id": board, "node_id": node, "canvas_version": version.Int64})
|
|
}
|
|
if state == "cancelled" {
|
|
cancelledSteps = append(cancelledSteps, step)
|
|
}
|
|
}
|
|
if err == nil {
|
|
err = rows.Err()
|
|
}
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
result := response{"run_id": id, "status": status, "applied": applied, "cancelled_step_ids": cancelledSteps, "worker_stop_grace_ms": 1000}
|
|
if receipt != nil {
|
|
result["run_status"], result["status"] = status, "applied"
|
|
bindCanvasAIReceipt(result, receipt)
|
|
}
|
|
encoded, _ := json.Marshal(result)
|
|
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_mutations(project,workspace_id,run_id,account_id,action,idempotency_key,payload_hash,response_body) VALUES($1,$2,$3,$4,'cancel',$5,$6,$7)`, project, workspace, id, account, key, hash, encoded)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return commitCanvasAIReceipt(w, r, tx, project, workspace, account, receipt, 200, "applied", result)
|
|
}
|