Add canonical canvas rows, stable identity migration and model operations
This commit is contained in:
@@ -0,0 +1,479 @@
|
||||
package platform
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
const modelMigrateAction = "model.migrate.v1"
|
||||
const modelEditAction = "model.operations.v2"
|
||||
|
||||
func newModelID() string {
|
||||
var value [16]byte
|
||||
if _, err := rand.Read(value[:]); err != nil {
|
||||
return ""
|
||||
}
|
||||
value[6], value[8] = value[6]&0x0f|0x40, value[8]&0x3f|0x80
|
||||
text := hex.EncodeToString(value[:])
|
||||
return text[:8] + "-" + text[8:12] + "-" + text[12:16] + "-" + text[16:20] + "-" + text[20:]
|
||||
}
|
||||
func legacyModelRoot(r *http.Request, tx *sql.Tx, project, workspace, board string) (string, error) {
|
||||
seen := map[string]bool{}
|
||||
for len(seen) <= 8 {
|
||||
if seen[board] {
|
||||
return "", errCanvasModel
|
||||
}
|
||||
seen[board] = true
|
||||
relation, err := readSubgraphRelation(r, tx, project, workspace, board)
|
||||
if err == sql.ErrNoRows {
|
||||
return board, nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if relation.Archived {
|
||||
return "", sql.ErrNoRows
|
||||
}
|
||||
board = relation.Parent
|
||||
}
|
||||
return "", errCanvasModel
|
||||
}
|
||||
func collectLegacyModel(r *http.Request, tx *sql.Tx, project, workspace, root string) (map[string]canvasSnapshot, []subgraphRelation, error) {
|
||||
rows, err := tx.QueryContext(r.Context(), `WITH RECURSIVE tree(id,path) AS (SELECT $3::uuid,ARRAY[$3::uuid] UNION ALL SELECT s.child_board_id,t.path||s.child_board_id FROM tree t JOIN canvas_subgraphs s ON s.project=$1 AND s.workspace_id=$2 AND s.parent_board_id=t.id WHERE s.archived_at IS NULL AND NOT s.child_board_id=ANY(t.path) AND cardinality(t.path)<=8) SELECT m.id::text,m.canvas_version,COALESCE(m.canvas_document,'{"schema_version":2,"nodes":[],"edges":[],"viewport":{"x":0,"y":0,"zoom":1}}'::jsonb) FROM tree t JOIN materials m ON m.project=$1 AND m.workspace_id=$2 AND m.id=t.id WHERE m.kind='board' ORDER BY m.id LIMIT 257`, project, workspace, root)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
snapshots := map[string]canvasSnapshot{}
|
||||
for rows.Next() {
|
||||
var snapshot canvasSnapshot
|
||||
if err = rows.Scan(&snapshot.BoardID, &snapshot.Version, &snapshot.Document); err != nil {
|
||||
rows.Close()
|
||||
return nil, nil, err
|
||||
}
|
||||
snapshots[snapshot.BoardID] = snapshot
|
||||
}
|
||||
err = rows.Err()
|
||||
rows.Close()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if len(snapshots) == 0 || len(snapshots) > 256 {
|
||||
return nil, nil, errGraphCapacity
|
||||
}
|
||||
rows, err = tx.QueryContext(r.Context(), `SELECT parent_board_id::text,frame_id,child_board_id::text FROM canvas_subgraphs WHERE project=$1 AND workspace_id=$2 AND archived_at IS NULL ORDER BY child_board_id`, project, workspace)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
relations := []subgraphRelation{}
|
||||
for rows.Next() {
|
||||
var relation subgraphRelation
|
||||
if err = rows.Scan(&relation.Parent, &relation.Frame, &relation.Child); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
if _, exists := snapshots[relation.Parent]; exists {
|
||||
relations = append(relations, relation)
|
||||
}
|
||||
}
|
||||
return snapshots, relations, rows.Err()
|
||||
}
|
||||
func modelRequestRetry(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account, root, action, key, hash string) (bool, error) {
|
||||
var previousRoot, previousAction, previousHash string
|
||||
var status int
|
||||
var body json.RawMessage
|
||||
err := tx.QueryRowContext(r.Context(), `SELECT root_id::text,action,payload_hash,response_status,response_body FROM canvas_model_requests WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND request_id=$4`, project, workspace, account, key).Scan(&previousRoot, &previousAction, &previousHash, &status, &body)
|
||||
if err == sql.ErrNoRows {
|
||||
return false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if previousRoot != root || previousAction != action || previousHash != hash {
|
||||
reject(w, 409, "idempotency_conflict")
|
||||
return true, nil
|
||||
}
|
||||
reply(w, status, body)
|
||||
return true, nil
|
||||
}
|
||||
func commitModelReceipt(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account, root, action, key, hash string, status int, body response) error {
|
||||
body["request_id"], body["protocol_version"] = key, canvasModelProtocol
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_model_requests(project,workspace_id,account_id,request_id,root_id,action,protocol_version,payload_hash,response_status,response_body) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)`, project, workspace, account, key, root, action, canvasModelProtocol, hash, status, raw)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err = tx.Commit(); err != nil {
|
||||
return err
|
||||
}
|
||||
reply(w, status, body)
|
||||
return nil
|
||||
}
|
||||
func decodeModelRequest(w http.ResponseWriter, r *http.Request, input any) bool {
|
||||
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, canvasLimit))
|
||||
decoder.DisallowUnknownFields()
|
||||
if decoder.Decode(input) != nil || decoder.Decode(new(any)) != io.EOF {
|
||||
reject(w, 400, "invalid_model_request")
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func (s *service) getCanvasModel(w http.ResponseWriter, r *http.Request, project, account string) error {
|
||||
if r.Method != http.MethodGet {
|
||||
reject(w, 405, "method_not_allowed")
|
||||
return nil
|
||||
}
|
||||
tx, err := s.db.BeginTx(r.Context(), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
workspace, board := r.PathValue("workspace"), r.PathValue("material")
|
||||
kind, _, err := lockMaterial(r.Context(), tx, project, workspace, board, account)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if kind != "board" {
|
||||
reject(w, 400, "invalid_material")
|
||||
return nil
|
||||
}
|
||||
root, err := canonicalRoot(r.Context(), tx, project, workspace, board)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if root == "" {
|
||||
root, err = legacyModelRoot(r, tx, project, workspace, board)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
snapshots, _, err := collectLegacyModel(r, tx, project, workspace, root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
versions := map[string]int64{}
|
||||
for id, snapshot := range snapshots {
|
||||
versions[id] = snapshot.Version
|
||||
}
|
||||
reply(w, 409, response{"error": "canonical_migration_required", "root_id": root, "source_versions": versions, "document_format_version": canvasModelFormat})
|
||||
return nil
|
||||
}
|
||||
model, err := loadCanvasModel(r.Context(), tx, project, workspace, root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err = tx.Rollback(); err != nil {
|
||||
return err
|
||||
}
|
||||
reply(w, 200, response{"model": model.view(nil)})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *service) migrateCanonicalCanvas(w http.ResponseWriter, r *http.Request, project, account string) error {
|
||||
if r.Method != http.MethodPost {
|
||||
reject(w, 405, "method_not_allowed")
|
||||
return nil
|
||||
}
|
||||
var input struct {
|
||||
ProtocolVersion int `json:"protocol_version"`
|
||||
RequestID string `json:"request_id"`
|
||||
Expected map[string]int64 `json:"expected_versions"`
|
||||
}
|
||||
if !decodeModelRequest(w, r, &input) {
|
||||
return nil
|
||||
}
|
||||
key := canvasIdempotencyKey(r)
|
||||
if input.ProtocolVersion != canvasModelProtocol || !uuidSlugPattern.MatchString(key) || input.RequestID != key || len(input.Expected) == 0 || len(input.Expected) > 256 {
|
||||
reject(w, 400, "invalid_model_migration")
|
||||
return nil
|
||||
}
|
||||
tx, err := s.db.BeginTx(r.Context(), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
workspace, board := r.PathValue("workspace"), r.PathValue("material")
|
||||
kind, role, err := lockMaterial(r.Context(), tx, project, workspace, board, 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
|
||||
}
|
||||
root, err := canonicalRoot(r.Context(), tx, project, workspace, board)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
already := root != ""
|
||||
if !already {
|
||||
root, err = legacyModelRoot(r, tx, project, workspace, board)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
done, err := modelRequestRetry(w, r, tx, project, workspace, account, root, modelMigrateAction, key, payloadDigest(input))
|
||||
if err != nil || done {
|
||||
return err
|
||||
}
|
||||
rejectMigration := func(status int, code string) error {
|
||||
return commitModelReceipt(w, r, tx, project, workspace, account, root, modelMigrateAction, key, payloadDigest(input), status, response{"status": "not-applied", "terminal": true, "error": code})
|
||||
}
|
||||
if already {
|
||||
return rejectMigration(409, "canonical_migration_already_applied")
|
||||
}
|
||||
snapshots, relations, err := collectLegacyModel(r, tx, project, workspace, root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(snapshots) != len(input.Expected) {
|
||||
return rejectMigration(412, "migration_version_conflict")
|
||||
}
|
||||
for id, snapshot := range snapshots {
|
||||
if input.Expected[id] != snapshot.Version {
|
||||
return rejectMigration(412, "migration_version_conflict")
|
||||
}
|
||||
}
|
||||
// Active or outcome-unknown paid work must settle before the identity barrier.
|
||||
for id := range snapshots {
|
||||
var busy bool
|
||||
if err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND status IN('queued','running'))`, project, workspace, id).Scan(&busy); err != nil {
|
||||
return err
|
||||
}
|
||||
if busy {
|
||||
return rejectMigration(409, "migration_waiting_for_runs")
|
||||
}
|
||||
}
|
||||
migration, err := migrateCanvasModel(root, snapshots, relations, newModelID)
|
||||
if err != nil {
|
||||
return rejectMigration(422, "model_migration_failed")
|
||||
}
|
||||
versionsRaw, _ := json.Marshal(input.Expected)
|
||||
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_model_projects(project,workspace_id,root_id,migration_id,source_versions) VALUES($1,$2,$3,$4,$5)`, project, workspace, root, key, versionsRaw)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
before := emptyModel(root)
|
||||
for id, snapshot := range snapshots {
|
||||
before.Versions[id] = snapshot.Version
|
||||
}
|
||||
participants := map[string]bool{}
|
||||
for id := range migration.Model.Canvases {
|
||||
participants[id] = true
|
||||
}
|
||||
if err = s.saveCanvasModel(r, tx, project, workspace, account, before, migration.Model, participants); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, identity := range migration.Identities {
|
||||
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_model_identity_map(project,workspace_id,root_id,source_canvas_id,source_structure,source_element_id,element_id) VALUES($1,$2,$3,$4,$5,$6,$7)`, project, workspace, root, identity.SourceCanvasID, identity.Structure, identity.SourceID, identity.ElementID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return commitModelReceipt(w, r, tx, project, workspace, account, root, modelMigrateAction, key, payloadDigest(input), 200, response{"status": "applied", "model": migration.Model.view(nil), "identity_map": migration.Identities, "source_versions": input.Expected})
|
||||
}
|
||||
|
||||
func (s *service) submitCanvasModelOperations(w http.ResponseWriter, r *http.Request, project, account string) error {
|
||||
if r.Method != http.MethodPost {
|
||||
reject(w, 405, "method_not_allowed")
|
||||
return nil
|
||||
}
|
||||
var input modelBatch
|
||||
if !decodeModelRequest(w, r, &input) {
|
||||
return nil
|
||||
}
|
||||
key := canvasIdempotencyKey(r)
|
||||
if input.ProtocolVersion != canvasModelProtocol || !uuidSlugPattern.MatchString(key) || input.RequestID != key || len(input.Expected) == 0 || len(input.Expected) > 256 || len(input.Operations) == 0 || len(input.Operations) > 2500 {
|
||||
reject(w, 400, "invalid_model_request")
|
||||
return nil
|
||||
}
|
||||
for id, version := range input.Expected {
|
||||
if !uuidSlugPattern.MatchString(id) || version < 0 || version > 9007199254740990 {
|
||||
reject(w, 400, "invalid_model_version")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
tx, err := s.db.BeginTx(r.Context(), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
workspace, board := r.PathValue("workspace"), r.PathValue("material")
|
||||
kind, role, err := lockMaterial(r.Context(), tx, project, workspace, board, 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
|
||||
}
|
||||
root, err := canonicalRoot(r.Context(), tx, project, workspace, board)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if root == "" {
|
||||
reject(w, 409, "canonical_migration_required")
|
||||
return nil
|
||||
}
|
||||
done, err := modelRequestRetry(w, r, tx, project, workspace, account, root, modelEditAction, key, payloadDigest(input))
|
||||
if err != nil || done {
|
||||
return err
|
||||
}
|
||||
rejectBatch := func(status int, code string) error {
|
||||
return commitModelReceipt(w, r, tx, project, workspace, account, root, modelEditAction, key, payloadDigest(input), status, response{"status": "not-applied", "terminal": true, "error": code})
|
||||
}
|
||||
before, err := loadCanvasModel(r.Context(), tx, project, workspace, root)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for id, expected := range input.Expected {
|
||||
if before.Versions[id] != expected {
|
||||
return rejectBatch(412, "version_conflict")
|
||||
}
|
||||
}
|
||||
for _, operation := range input.Operations {
|
||||
if !uuidSlugPattern.MatchString(operation.ID) {
|
||||
return rejectBatch(400, "invalid_model_operation")
|
||||
}
|
||||
var used bool
|
||||
if err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_model_operations WHERE project=$1 AND workspace_id=$2 AND operation_id=$3)`, project, workspace, operation.ID).Scan(&used); err != nil {
|
||||
return err
|
||||
}
|
||||
if used {
|
||||
return rejectBatch(409, "operation_already_applied")
|
||||
}
|
||||
}
|
||||
after, touched, err := applyModelOperations(before, input.Operations)
|
||||
if err != nil {
|
||||
code := "invalid_model_operation"
|
||||
if err == errModelConflict {
|
||||
code = "canvas_model_field_conflict"
|
||||
}
|
||||
return rejectBatch(409, code)
|
||||
}
|
||||
for id := range touched {
|
||||
if _, exists := input.Expected[id]; !exists {
|
||||
return rejectBatch(428, "version_required")
|
||||
}
|
||||
}
|
||||
for id := range input.Expected {
|
||||
if _, existed := before.Versions[id]; !existed {
|
||||
if _, created := after.Canvases[id]; !created {
|
||||
return rejectBatch(409, "invalid_model_participant")
|
||||
}
|
||||
var exists bool
|
||||
if err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM materials WHERE project=$1 AND workspace_id=$2 AND id=$3)`, project, workspace, id).Scan(&exists); err != nil {
|
||||
return err
|
||||
}
|
||||
if exists {
|
||||
return rejectBatch(409, "canvas_identity_conflict")
|
||||
}
|
||||
}
|
||||
if !touched[id] {
|
||||
touched[id] = true
|
||||
after.Versions[id] = before.Versions[id] + 1
|
||||
}
|
||||
}
|
||||
for id, element := range after.Elements {
|
||||
if before.Elements[id] == nil {
|
||||
var foreign bool
|
||||
if err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_model_elements WHERE project=$1 AND element_id=$2 AND (workspace_id<>$3 OR root_id<>$4))`, project, id, workspace, root).Scan(&foreign); err != nil {
|
||||
return err
|
||||
}
|
||||
if foreign {
|
||||
return rejectBatch(409, "element_identity_conflict")
|
||||
}
|
||||
}
|
||||
if element["structure"] == "node" {
|
||||
data := element["data"].(map[string]any)
|
||||
if asset := modelString(data["assetId"]); asset != "" {
|
||||
prior := before.Elements[id]
|
||||
if prior == nil || !modelEqual(prior["data"].(map[string]any)["assetId"], data["assetId"]) {
|
||||
readable, e := assetReadable(r, tx, project, workspace, asset, account)
|
||||
if e != nil {
|
||||
return e
|
||||
}
|
||||
if !readable {
|
||||
return rejectBatch(400, "invalid_asset_reference")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if err = s.saveCanvasModel(r, tx, project, workspace, account, before, after, touched); err != nil {
|
||||
return err
|
||||
}
|
||||
ids := []string{}
|
||||
for _, operation := range input.Operations {
|
||||
ids = append(ids, operation.ID)
|
||||
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_model_operations(project,workspace_id,operation_id,request_id,account_id) VALUES($1,$2,$3,$4,$5)`, project, workspace, operation.ID, key, account)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return commitModelReceipt(w, r, tx, project, workspace, account, root, modelEditAction, key, payloadDigest(input), 200, response{"status": "applied", "operation_ids": ids, "model": after.view(input.Expected)})
|
||||
}
|
||||
func (s *service) getCanvasModelOperation(w http.ResponseWriter, r *http.Request, project, account string) error {
|
||||
if r.Method != http.MethodGet {
|
||||
reject(w, 405, "method_not_allowed")
|
||||
return nil
|
||||
}
|
||||
key := r.PathValue("request")
|
||||
if !uuidSlugPattern.MatchString(key) {
|
||||
reject(w, 400, "invalid_request_id")
|
||||
return nil
|
||||
}
|
||||
tx, err := s.db.BeginTx(r.Context(), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
workspace, board := r.PathValue("workspace"), r.PathValue("material")
|
||||
if _, _, err = lockMaterial(r.Context(), tx, project, workspace, board, account); err != nil {
|
||||
return err
|
||||
}
|
||||
root, err := canonicalRoot(r.Context(), tx, project, workspace, board)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if root == "" {
|
||||
root, err = legacyModelRoot(r, tx, project, workspace, board)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
var body json.RawMessage
|
||||
err = tx.QueryRowContext(r.Context(), `SELECT response_body FROM canvas_model_requests WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND request_id=$4 AND root_id=$5`, project, workspace, account, key, root).Scan(&body)
|
||||
if err == sql.ErrNoRows {
|
||||
reply(w, 404, response{"status": "unknown", "error": "operation_receipt_unknown", "request_id": key})
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err = tx.Rollback(); err != nil {
|
||||
return err
|
||||
}
|
||||
reply(w, 200, body)
|
||||
return nil
|
||||
}
|
||||
|
||||
func modelExists(ctx context.Context, tx *sql.Tx, project, workspace, board string) (bool, error) {
|
||||
root, err := canonicalRoot(ctx, tx, project, workspace, board)
|
||||
return root != "", err
|
||||
}
|
||||
Reference in New Issue
Block a user