84 lines
3.4 KiB
Go
84 lines
3.4 KiB
Go
package platform
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"net/http"
|
|
)
|
|
|
|
type canvasAIRequest struct {
|
|
Key, Hash, Action, Board, Run, Operation string
|
|
}
|
|
|
|
// The workspace lock is acquired by the caller before this savepoint. Rejected
|
|
// requests roll back every staged effect and retain that lock while recording
|
|
// their zero-effect receipt, so a response can never race a second invocation.
|
|
func beginCanvasAIRequest(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account string, request *canvasAIRequest) (bool, error) {
|
|
if request == nil {
|
|
return false, nil
|
|
}
|
|
var board, run, action, hash string
|
|
var status int
|
|
var body json.RawMessage
|
|
err := tx.QueryRowContext(r.Context(), `SELECT board_id::text,COALESCE(run_id::text,''),action,payload_hash,response_status,response_body FROM canvas_ai_requests WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND request_id=$4`, project, workspace, account, request.Key).Scan(&board, &run, &action, &hash, &status, &body)
|
|
if err == nil {
|
|
if board != request.Board || run != request.Run || action != request.Action || hash != request.Hash {
|
|
reject(w, 409, "idempotency_conflict")
|
|
return true, nil
|
|
}
|
|
reply(w, status, body)
|
|
return true, nil
|
|
}
|
|
if err != sql.ErrNoRows {
|
|
return false, err
|
|
}
|
|
var completed bool
|
|
err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_requests WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND operation_id=$4 AND outcome='applied')`, project, workspace, account, request.Operation).Scan(&completed)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if completed {
|
|
reject(w, 409, "operation_identity_conflict")
|
|
return true, nil
|
|
}
|
|
_, err = tx.ExecContext(r.Context(), `SAVEPOINT canvas_ai_effects`)
|
|
return false, err
|
|
}
|
|
func bindCanvasAIReceipt(body response, request *canvasAIRequest) {
|
|
if request == nil {
|
|
return
|
|
}
|
|
body["protocol_version"], body["request_id"], body["operation_id"], body["action"] = 2, request.Key, request.Operation, request.Action
|
|
if request.Run != "" {
|
|
body["run_id"] = request.Run
|
|
}
|
|
}
|
|
func commitCanvasAIReceipt(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account string, request *canvasAIRequest, status int, outcome string, body response) error {
|
|
if request != nil {
|
|
bindCanvasAIReceipt(body, request)
|
|
raw, err := json.Marshal(body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_requests(project,workspace_id,account_id,request_id,board_id,run_id,action,operation_id,payload_hash,outcome,response_status,response_body) VALUES($1,$2,$3,$4,$5,NULLIF($6,'')::uuid,$7,$8,$9,$10,$11,$12)`, project, workspace, account, request.Key, request.Board, request.Run, request.Action, request.Operation, request.Hash, outcome, status, raw)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return err
|
|
}
|
|
reply(w, status, body)
|
|
return nil
|
|
}
|
|
func rejectCanvasAIRequest(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account string, request *canvasAIRequest, status int, code string) error {
|
|
if request == nil {
|
|
reject(w, status, code)
|
|
return nil
|
|
}
|
|
if _, err := tx.ExecContext(r.Context(), `ROLLBACK TO SAVEPOINT canvas_ai_effects`); err != nil {
|
|
return err
|
|
}
|
|
return commitCanvasAIReceipt(w, r, tx, project, workspace, account, request, status, "not-applied", response{"status": "not-applied", "terminal": true, "error": code})
|
|
}
|