352 lines
11 KiB
Go
352 lines
11 KiB
Go
package platform
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"net/http"
|
|
"sort"
|
|
"time"
|
|
)
|
|
|
|
// Endpoints name real nodes, never transient projected copies or client-created ports.
|
|
type canvasConnectionEndpoint struct {
|
|
Board string `json:"board_id"`
|
|
Node string `json:"node_id"`
|
|
Handle string `json:"handle"`
|
|
}
|
|
type canvasConnectionInput struct {
|
|
Source canvasConnectionEndpoint `json:"source"`
|
|
Target canvasConnectionEndpoint `json:"target"`
|
|
Edge struct {
|
|
ID string `json:"id"`
|
|
Data map[string]any `json:"data,omitempty"`
|
|
} `json:"edge"`
|
|
Expected map[string]int64 `json:"expected_versions"`
|
|
}
|
|
|
|
// Paths run from a leaf board upwards to (but excluding) the coordinator.
|
|
func connectionPath(r *http.Request, tx *sql.Tx, project, workspace, coordinator, board string) ([]subgraphRelation, error) {
|
|
path := []subgraphRelation{}
|
|
seen := map[string]bool{}
|
|
for board != coordinator {
|
|
if seen[board] || len(path) >= 8 {
|
|
return nil, errGraphOwnership
|
|
}
|
|
seen[board] = true
|
|
relation, err := readSubgraphRelation(r, tx, project, workspace, board)
|
|
if err == sql.ErrNoRows {
|
|
return nil, errGraphOwnership
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if relation.Archived {
|
|
return nil, errGraphOwnership
|
|
}
|
|
path = append(path, relation)
|
|
board = relation.Parent
|
|
}
|
|
return path, nil
|
|
}
|
|
|
|
func connectionNode(doc map[string]any, id string) map[string]any {
|
|
for _, value := range doc["nodes"].([]any) {
|
|
node := value.(map[string]any)
|
|
if node["id"] == id {
|
|
return node
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Reuse ports by logical endpoint, including legacy ports with non-derived IDs.
|
|
// A derived ID collision must never retarget an existing port.
|
|
func connectionPort(data map[string]any, node, handle, direction string) (string, bool, error) {
|
|
ports, _ := data["ports"].([]any)
|
|
sum := sha256.Sum256([]byte(direction + "/" + node + "/" + handle))
|
|
id := "p-" + hex.EncodeToString(sum[:16])
|
|
collision := false
|
|
for _, value := range ports {
|
|
port := value.(map[string]any)
|
|
if port["direction"] == direction && port["nodeId"] == node && port["handle"] == handle {
|
|
return port["id"].(string), false, nil
|
|
}
|
|
collision = collision || port["id"] == id
|
|
}
|
|
if collision {
|
|
return "", false, errGraphOwnership
|
|
}
|
|
if len(ports) >= 4000 {
|
|
return "", false, errGraphCapacity
|
|
}
|
|
data["ports"] = append(ports, map[string]any{"id": id, "direction": direction, "nodeId": node, "handle": handle})
|
|
return id, true, nil
|
|
}
|
|
|
|
func exposeConnectionEndpoint(endpoint canvasConnectionEndpoint, direction string, path []subgraphRelation, documents map[string]map[string]any) (canvasConnectionEndpoint, []any, error) {
|
|
node := connectionNode(documents[endpoint.Board], endpoint.Node)
|
|
if node == nil {
|
|
return endpoint, nil, errGraphOwnership
|
|
}
|
|
data := node["data"].(map[string]any)
|
|
if !canvasChoice(data["kind"], "text", "script", "plain-text", "image", "plain-image") {
|
|
return endpoint, nil, errGraphOwnership
|
|
}
|
|
created := []any{}
|
|
for _, relation := range path {
|
|
frame := connectionNode(documents[relation.Parent], relation.Frame)
|
|
if frame == nil {
|
|
return endpoint, nil, errGraphOwnership
|
|
}
|
|
data := frame["data"].(map[string]any)
|
|
sub, _ := data["subgraph"].(map[string]any)
|
|
if relation.Child != endpoint.Board || data["kind"] != "frame" || sub["boardId"] != relation.Child {
|
|
return endpoint, nil, errGraphOwnership
|
|
}
|
|
id, added, err := connectionPort(data, endpoint.Node, endpoint.Handle, direction)
|
|
if err != nil {
|
|
return endpoint, nil, err
|
|
}
|
|
if added {
|
|
created = append(created, response{"board_id": relation.Parent, "frame_id": relation.Frame, "port_id": id})
|
|
}
|
|
endpoint = canvasConnectionEndpoint{Board: relation.Parent, Node: relation.Frame, Handle: "port:" + id}
|
|
}
|
|
return endpoint, created, nil
|
|
}
|
|
|
|
func (s *service) createCanvasConnection(w http.ResponseWriter, r *http.Request, project, account string) error {
|
|
if r.Method != http.MethodPost {
|
|
reject(w, 405, "method_not_allowed")
|
|
return nil
|
|
}
|
|
if !acceptsCanvasV2(r) {
|
|
reject(w, 409, "canvas_schema_upgrade_required")
|
|
return nil
|
|
}
|
|
key := canvasIdempotencyKey(r)
|
|
if !uuidSlugPattern.MatchString(key) {
|
|
reject(w, 400, "invalid_idempotency_key")
|
|
return nil
|
|
}
|
|
var input canvasConnectionInput
|
|
if !decode(w, r, &input) {
|
|
return nil
|
|
}
|
|
for _, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
|
|
if !uuidSlugPattern.MatchString(endpoint.Board) || !canvasID.MatchString(endpoint.Node) || !canvasChoice(endpoint.Handle, "top", "right", "bottom", "left") {
|
|
reject(w, 400, "invalid_connection_request")
|
|
return nil
|
|
}
|
|
}
|
|
if !canvasID.MatchString(input.Edge.ID) {
|
|
reject(w, 400, "invalid_edge")
|
|
return nil
|
|
}
|
|
limits, _ := json.Marshal(response{"edges": []any{input.Edge}})
|
|
if canvasExceededGraphLimits(limits) {
|
|
reject(w, 413, "graph_limit_exceeded")
|
|
return nil
|
|
}
|
|
if input.Edge.Data != nil && !validCanvasEdgeData(input.Edge.Data) {
|
|
reject(w, 400, "invalid_edge")
|
|
return nil
|
|
}
|
|
if len(input.Expected) == 0 {
|
|
reject(w, 428, "version_required")
|
|
return nil
|
|
}
|
|
workspace, coordinator := r.PathValue("workspace"), r.PathValue("material")
|
|
tx, err := s.db.BeginTx(r.Context(), nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
kind, role, err := lockMaterial(r.Context(), tx, project, workspace, coordinator, account)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if kind != "board" {
|
|
reject(w, 400, "invalid_material")
|
|
return nil
|
|
}
|
|
if role == "viewer" {
|
|
reject(w, 403, "forbidden")
|
|
return nil
|
|
}
|
|
action, hash := "create_connection", payloadDigest(input)
|
|
done, err := transactionRetry(w, r, tx, project, account, coordinator, action, key, hash)
|
|
if err != nil || done {
|
|
return err
|
|
}
|
|
participants := map[string]bool{coordinator: true}
|
|
paths := [][]subgraphRelation{}
|
|
for _, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
|
|
kind, role, err := lockMaterial(r.Context(), tx, project, workspace, endpoint.Board, account)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if role == "viewer" {
|
|
reject(w, 403, "forbidden")
|
|
return nil
|
|
}
|
|
if kind != "board" {
|
|
reject(w, 400, "invalid_material")
|
|
return nil
|
|
}
|
|
path, err := connectionPath(r, tx, project, workspace, coordinator, endpoint.Board)
|
|
if err == errGraphOwnership {
|
|
reject(w, 409, "ownership_conflict")
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
paths = append(paths, path)
|
|
participants[endpoint.Board] = true
|
|
for _, relation := range path {
|
|
participants[relation.Parent] = true
|
|
}
|
|
}
|
|
// Require the lowest common ancestor, so the logical edge has one owner.
|
|
left := map[string]bool{}
|
|
for _, relation := range paths[0] {
|
|
left[relation.Child] = true
|
|
}
|
|
for _, relation := range paths[1] {
|
|
if left[relation.Child] {
|
|
reject(w, 409, "connection_owner_mismatch")
|
|
return nil
|
|
}
|
|
}
|
|
ids := []string{}
|
|
for id := range participants {
|
|
ids = append(ids, id)
|
|
}
|
|
sort.Strings(ids)
|
|
before := []canvasSnapshot{}
|
|
snapshots := map[string]canvasSnapshot{}
|
|
documents := map[string]map[string]any{}
|
|
for _, id := range ids {
|
|
_, role, err := lockMaterial(r.Context(), tx, project, workspace, id, account)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if role == "viewer" {
|
|
reject(w, 403, "forbidden")
|
|
return nil
|
|
}
|
|
snapshot, err := loadCanvas(r, tx, project, workspace, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
snapshots[id] = snapshot
|
|
before = append(before, snapshot)
|
|
var doc map[string]any
|
|
if err = json.Unmarshal(snapshot.Document, &doc); err != nil {
|
|
return err
|
|
}
|
|
documents[id] = doc
|
|
}
|
|
if len(input.Expected) != len(ids) {
|
|
reply(w, 428, response{"error": "version_required", "details": response{"required_board_ids": ids}})
|
|
return nil
|
|
}
|
|
for _, id := range ids {
|
|
expected, exists := input.Expected[id]
|
|
if !exists {
|
|
reply(w, 428, response{"error": "version_required", "details": response{"required_board_ids": ids}})
|
|
return nil
|
|
}
|
|
if expected != snapshots[id].Version {
|
|
reject(w, 412, "version_conflict")
|
|
return nil
|
|
}
|
|
}
|
|
for _, value := range documents[coordinator]["edges"].([]any) {
|
|
if value.(map[string]any)["id"] == input.Edge.ID {
|
|
reject(w, 409, "edge_id_conflict")
|
|
return nil
|
|
}
|
|
}
|
|
mapped := []canvasConnectionEndpoint{}
|
|
created := []any{}
|
|
for i, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
|
|
direction := "output"
|
|
if i == 1 {
|
|
direction = "input"
|
|
}
|
|
exposed, ports, err := exposeConnectionEndpoint(endpoint, direction, paths[i], documents)
|
|
if err == errGraphOwnership {
|
|
reject(w, 409, "ownership_conflict")
|
|
return nil
|
|
}
|
|
if err == errGraphCapacity {
|
|
reject(w, 413, "graph_limit_exceeded")
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
mapped = append(mapped, exposed)
|
|
created = append(created, ports...)
|
|
}
|
|
edge := map[string]any{"id": input.Edge.ID, "source": mapped[0].Node, "sourceHandle": mapped[0].Handle, "target": mapped[1].Node, "targetHandle": mapped[1].Handle}
|
|
if input.Edge.Data != nil {
|
|
edge["data"] = input.Edge.Data
|
|
}
|
|
documents[coordinator]["edges"] = append(documents[coordinator]["edges"].([]any), edge)
|
|
updated := []canvasSnapshot{}
|
|
versions := map[string]int64{}
|
|
for _, id := range ids {
|
|
documents[id]["schema_version"] = 2
|
|
raw, _ := json.Marshal(documents[id])
|
|
if len(raw) > canvasLimit {
|
|
reject(w, 413, "canvas_too_large")
|
|
return nil
|
|
}
|
|
if canvasExceededGraphLimits(raw) {
|
|
reject(w, 413, "graph_limit_exceeded")
|
|
return nil
|
|
}
|
|
normalized, ok := normalizeCanvas(raw)
|
|
if !ok {
|
|
reject(w, 400, "invalid_canvas")
|
|
return nil
|
|
}
|
|
snapshot := snapshots[id]
|
|
snapshot.Document = normalized
|
|
snapshot.Version++
|
|
updated = append(updated, snapshot)
|
|
snapshots[id] = snapshot
|
|
versions[id] = snapshot.Version
|
|
}
|
|
// Nothing is persisted until both complete port chains and all snapshots validate.
|
|
for _, snapshot := range updated {
|
|
if err = s.saveCanvas(r, tx, project, workspace, snapshot); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
var id string
|
|
var expiry time.Time
|
|
if err = tx.QueryRowContext(r.Context(), `SELECT gen_random_uuid()::text,now()+interval '7 days'`).Scan(&id, &expiry); err != nil {
|
|
return err
|
|
}
|
|
result := response{"transaction_id": id, "parent": snapshots[coordinator], "snapshots": updated, "connection": response{"board_id": coordinator, "edge": edge, "source": input.Source, "target": input.Target}, "created_ports": created, "undo_receipt": response{"transaction_id": id, "valid_for_seconds": 604800, "expires_at": expiry}}
|
|
encoded, _ := json.Marshal(result)
|
|
prior, _ := json.Marshal(before)
|
|
after, _ := json.Marshal(versions)
|
|
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_transactions(project,workspace_id,board_id,id,account_id,action,idempotency_key,payload_hash,response_status,response_body,before_state,after_versions) VALUES($1,$2,$3,$4,$5,$6,$7,$8,201,$9,$10,$11)`, project, workspace, coordinator, id, account, action, key, hash, encoded, prior, after)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
return err
|
|
}
|
|
setVersion(w, versions[coordinator])
|
|
reply(w, 201, result)
|
|
return nil
|
|
}
|