268 lines
10 KiB
Go
268 lines
10 KiB
Go
package platform
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/http"
|
|
)
|
|
|
|
var errAITargetChanged = errors.New("target_changed")
|
|
var errAIInvalidPatch = errors.New("invalid_ai_patch")
|
|
|
|
// The snapshot is an AI reader projection only. Canonical changes retain the
|
|
// original rows, including unknown envelopes, geometry and reference metadata.
|
|
type canvasAIChange struct {
|
|
Snapshot canvasSnapshot
|
|
Before *canvasModel
|
|
After *canvasModel
|
|
Incarnation int64
|
|
}
|
|
|
|
// Until a narrower read-set protocol exists, admission binds every active
|
|
// canvas in this canonical project. Archived versions are not graph inputs.
|
|
// Workspace locking keeps validation, plan freezing and admission atomic.
|
|
func validateCanvasAIVersions(model canvasModel, board string, header int64, expected map[string]int64) string {
|
|
if len(expected) != len(model.Canvases) || expected[board] != header {
|
|
return "invalid_ai_versions"
|
|
}
|
|
for id := range model.Canvases {
|
|
version, ok := expected[id]
|
|
if !ok || version < 1 || version > 9007199254740991 {
|
|
return "invalid_ai_versions"
|
|
}
|
|
if version != model.Versions[id] {
|
|
return "version_conflict"
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func prepareModelAIPatch(before canvasModel, board, node string, patch response) (canvasModel, error) {
|
|
element := before.Elements[node]
|
|
if element == nil || element["structure"] != "node" || element["ownerCanvasId"] != board || modelInt(element["dataVersion"]) != 1 || !canvasChoice(element["componentId"], "text", "script", "plain-text", "image", "plain-image") {
|
|
return canvasModel{}, errAITargetChanged
|
|
}
|
|
after := cloneModel(before)
|
|
data := after.Elements[node]["data"].(map[string]any)
|
|
for key, value := range patch {
|
|
if !canvasChoice(key, "ai", "text", "src", "assetId") {
|
|
return canvasModel{}, errAIInvalidPatch
|
|
}
|
|
data[key] = value
|
|
}
|
|
if before.Versions[board] < 1 || before.Versions[board] >= 9007199254740991 {
|
|
return canvasModel{}, errModelConflict
|
|
}
|
|
after.Versions[board]++
|
|
if err := validateCanvasModel(after); err != nil {
|
|
return canvasModel{}, err
|
|
}
|
|
return after, nil
|
|
}
|
|
func prepareCanvasAIPatch(r *http.Request, tx *sql.Tx, project, workspace string, snapshot canvasSnapshot, node string, patch response) (canvasAIChange, error) {
|
|
change := canvasAIChange{Snapshot: snapshot}
|
|
root, err := canonicalRoot(r.Context(), tx, project, workspace, snapshot.BoardID)
|
|
if err != nil {
|
|
return change, err
|
|
}
|
|
if root != "" {
|
|
before, err := loadCanvasModel(r.Context(), tx, project, workspace, root)
|
|
if err != nil {
|
|
return change, err
|
|
}
|
|
if before.Versions[snapshot.BoardID] != snapshot.Version {
|
|
return change, errModelConflict
|
|
}
|
|
after, err := prepareModelAIPatch(before, snapshot.BoardID, node, patch)
|
|
if err != nil {
|
|
return change, err
|
|
}
|
|
projection, err := projectCanvasModel(after)
|
|
if err != nil {
|
|
return change, err
|
|
}
|
|
change.Before, change.After = &before, &after
|
|
change.Incarnation = modelInt(before.Elements[node]["incarnation"])
|
|
change.Snapshot.Document, change.Snapshot.Version = projection[snapshot.BoardID], after.Versions[snapshot.BoardID]
|
|
return change, nil
|
|
}
|
|
var document map[string]any
|
|
if err = json.Unmarshal(snapshot.Document, &document); err != nil {
|
|
return change, err
|
|
}
|
|
target := canvasNodeData(snapshot.Document)[node]
|
|
if target == nil {
|
|
return change, errAITargetChanged
|
|
}
|
|
// The data lookup owns a separate decode. Update the staged document itself.
|
|
for _, raw := range document["nodes"].([]any) {
|
|
element := raw.(map[string]any)
|
|
if element["id"] != node {
|
|
continue
|
|
}
|
|
data := element["data"].(map[string]any)
|
|
for key, value := range patch {
|
|
if !canvasChoice(key, "ai", "text", "src", "assetId") {
|
|
return change, errAIInvalidPatch
|
|
}
|
|
data[key] = value
|
|
}
|
|
}
|
|
document["schema_version"] = float64(2)
|
|
encoded, err := json.Marshal(document)
|
|
if err != nil {
|
|
return change, err
|
|
}
|
|
normalized, ok := normalizeCanvas(encoded)
|
|
if !ok {
|
|
return change, errAIInvalidPatch
|
|
}
|
|
change.Snapshot.Document = normalized
|
|
change.Snapshot.Version++
|
|
return change, nil
|
|
}
|
|
func (s *service) saveCanvasAIChange(r *http.Request, tx *sql.Tx, project, workspace, account string, change canvasAIChange) error {
|
|
if change.After == nil {
|
|
return s.saveCanvas(r, tx, project, workspace, change.Snapshot)
|
|
}
|
|
return s.saveCanvasModel(r, tx, project, workspace, account, *change.Before, *change.After, map[string]bool{change.Snapshot.BoardID: true})
|
|
}
|
|
func (change canvasAIChange) addCanonical(result response) {
|
|
if change.After == nil {
|
|
return
|
|
}
|
|
board := change.Snapshot.BoardID
|
|
result["document_format_version"], result["target_incarnation"] = canvasModelFormat, change.Incarnation
|
|
result["model"] = change.After.view(map[string]int64{board: change.Snapshot.Version})
|
|
}
|
|
|
|
func canvasAIIdentity(ctx context.Context, tx *sql.Tx, project, workspace, board, node string) (string, int64, string, error) {
|
|
root, err := canonicalRoot(ctx, tx, project, workspace, board)
|
|
if err != nil || root == "" {
|
|
return root, 0, "", err
|
|
}
|
|
var incarnation, version int64
|
|
var kind string
|
|
err = tx.QueryRowContext(ctx, `SELECT incarnation,record->>'componentId',(record->>'dataVersion')::bigint FROM canvas_model_elements WHERE project=$1 AND workspace_id=$2 AND root_id=$3 AND element_id=$4 AND owner_canvas_id=$5 AND NOT deleted AND record->>'structure'='node'`, project, workspace, root, node, board).Scan(&incarnation, &kind, &version)
|
|
if err == sql.ErrNoRows {
|
|
return root, 0, "", nil
|
|
}
|
|
if err != nil {
|
|
return root, 0, "", err
|
|
}
|
|
if version != 1 || !canvasChoice(kind, "text", "script", "plain-text", "image", "plain-image") {
|
|
return root, 0, "", nil
|
|
}
|
|
return root, incarnation, kind, nil
|
|
}
|
|
func bindCanvasAIIdentities(ctx context.Context, tx *sql.Tx, project, workspace string, plan *canvasAIPlan) error {
|
|
for i, step := range plan.Steps {
|
|
root, incarnation, kind, err := canvasAIIdentity(ctx, tx, project, workspace, step.BoardID, step.NodeID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if root != "" && (incarnation < 1 || kind != step.Kind) {
|
|
return errAITargetChanged
|
|
}
|
|
plan.Steps[i].CanonicalRoot, plan.Steps[i].TargetIncarnation = root, incarnation
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Run URLs remain anchored at their admission board, even after that partition
|
|
// is archived. Canonical ownership grants are inherited from the same root.
|
|
func lockCanvasAIRunRoute(ctx context.Context, tx *sql.Tx, project, workspace, board, account string) (string, string, error) {
|
|
root, err := canonicalRoot(ctx, tx, project, workspace, board)
|
|
if err != nil {
|
|
return "", "", err
|
|
}
|
|
if root != "" {
|
|
board = root
|
|
}
|
|
return lockMaterial(ctx, tx, project, workspace, board, account)
|
|
}
|
|
func lockCanvasAIWorkspace(ctx context.Context, tx *sql.Tx, project, workspace string) error {
|
|
var id string
|
|
return tx.QueryRowContext(ctx, `SELECT id::text FROM workspaces WHERE project=$1 AND id=$2 FOR UPDATE`, project, workspace).Scan(&id)
|
|
}
|
|
|
|
// Caller holds the workspace lock. Frozen identity is stable; owner is resolved
|
|
// afresh at execution, hydration and write time, never written into the plan.
|
|
func resolveCanvasAITarget(ctx context.Context, tx *sql.Tx, project, workspace string, step canvasAIStepPlan) (string, bool, error) {
|
|
if step.CanonicalRoot == "" {
|
|
root, err := canonicalRoot(ctx, tx, project, workspace, step.BoardID)
|
|
return step.BoardID, err == nil && root == "" && step.TargetIncarnation == 0, err
|
|
}
|
|
var board, kind string
|
|
var incarnation, version int64
|
|
err := tx.QueryRowContext(ctx, `SELECT e.owner_canvas_id::text,e.incarnation,e.record->>'componentId',(e.record->>'dataVersion')::bigint FROM canvas_model_elements e JOIN canvas_model_canvases c ON c.project=e.project AND c.workspace_id=e.workspace_id AND c.canvas_id=e.owner_canvas_id AND c.root_id=e.root_id WHERE e.project=$1 AND e.workspace_id=$2 AND e.root_id=$3 AND e.element_id=$4 AND NOT e.deleted AND NOT c.archived AND e.record->>'structure'='node'`, project, workspace, step.CanonicalRoot, step.NodeID).Scan(&board, &incarnation, &kind, &version)
|
|
if err == sql.ErrNoRows {
|
|
return "", false, nil
|
|
}
|
|
if err != nil {
|
|
return "", false, err
|
|
}
|
|
return board, incarnation == step.TargetIncarnation && incarnation > 0 && kind == step.Kind && version == 1, nil
|
|
}
|
|
func canvasAIIdentityMatches(ctx context.Context, tx *sql.Tx, project, workspace string, step canvasAIStepPlan) (bool, error) {
|
|
_, ok, err := resolveCanvasAITarget(ctx, tx, project, workspace, step)
|
|
return ok, err
|
|
}
|
|
func (change canvasAIChange) addTargetContext(result response, step canvasAIStepPlan) {
|
|
change.addCanonical(result)
|
|
if step.CanonicalRoot != "" {
|
|
result["origin_board_id"], result["canonical_root"] = step.BoardID, step.CanonicalRoot
|
|
if change.After != nil && change.Snapshot.BoardID != step.BoardID {
|
|
result["model"] = change.After.view(nil)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Whole identity transitions cancel old jobs even when delete/restore happens
|
|
// in a single atomic batch and the compatibility projection looks identical.
|
|
func invalidateCanonicalAIIdentities(r *http.Request, tx *sql.Tx, project, workspace string, before, after canvasModel) error {
|
|
removed := map[string][]any{}
|
|
for id, element := range before.Elements {
|
|
if element["structure"] != "node" {
|
|
continue
|
|
}
|
|
next := after.Elements[id]
|
|
if next != nil && modelInt(next["incarnation"]) == modelInt(element["incarnation"]) && next["componentId"] == element["componentId"] && modelInt(next["dataVersion"]) == modelInt(element["dataVersion"]) {
|
|
continue
|
|
}
|
|
rows, err := tx.QueryContext(r.Context(), `SELECT DISTINCT board_id::text FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND frozen_input->>'canonical_root'=$3 AND node_id=$4 AND frozen_input->>'target_incarnation'=$5::bigint::text AND status IN('queued','running')`, project, workspace, before.Root, id, modelInt(element["incarnation"]))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
boards := []string{}
|
|
for rows.Next() {
|
|
var board string
|
|
if err = rows.Scan(&board); err != nil {
|
|
rows.Close()
|
|
return err
|
|
}
|
|
boards = append(boards, board)
|
|
}
|
|
err = rows.Err()
|
|
rows.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, board := range boards {
|
|
removed[board] = append(removed[board], map[string]any{"id": id, "data": map[string]any{"kind": element["componentId"]}})
|
|
}
|
|
}
|
|
for _, board := range sortedKeys(removed) {
|
|
raw, err := json.Marshal(map[string]any{"nodes": removed[board]})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = invalidateCanvasAITargets(r, tx, project, workspace, board, raw, json.RawMessage(`{"nodes":[]}`)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|