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 }