216 lines
7.7 KiB
Go
216 lines
7.7 KiB
Go
package platform
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"net/http"
|
|
"time"
|
|
)
|
|
|
|
func transactionRetry(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, account, board, action, key, hash string) (bool, error) {
|
|
var previousHash string
|
|
var status int
|
|
var previous json.RawMessage
|
|
err := tx.QueryRowContext(r.Context(), `SELECT payload_hash,response_status,response_body FROM canvas_transactions WHERE project=$1 AND account_id=$2 AND board_id=$3 AND action=$4 AND idempotency_key=$5 AND workspace_id=$6`, project, account, board, action, key, r.PathValue("workspace")).Scan(&previousHash, &status, &previous)
|
|
if err == sql.ErrNoRows {
|
|
return false, nil
|
|
}
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if hash != previousHash {
|
|
reject(w, 409, "idempotency_conflict")
|
|
return true, nil
|
|
}
|
|
var result struct {
|
|
Parent canvasSnapshot `json:"parent"`
|
|
}
|
|
_ = json.Unmarshal(previous, &result)
|
|
if result.Parent.Version > 0 {
|
|
setVersion(w, result.Parent.Version)
|
|
}
|
|
reply(w, status, previous)
|
|
return true, nil
|
|
}
|
|
func payloadDigest(v any) string {
|
|
raw, _ := json.Marshal(v)
|
|
h := sha256.Sum256(raw)
|
|
return hex.EncodeToString(h[:])
|
|
}
|
|
func (s *service) saveCanvas(r *http.Request, tx *sql.Tx, project, workspace string, snapshot canvasSnapshot) error {
|
|
var before json.RawMessage
|
|
if err := tx.QueryRowContext(r.Context(), `SELECT COALESCE(canvas_document,'{"schema_version":1,"nodes":[],"edges":[],"viewport":{"x":0,"y":0,"zoom":1}}'::jsonb) FROM materials WHERE project=$1 AND workspace_id=$2 AND id=$3 FOR UPDATE`, project, workspace, snapshot.BoardID).Scan(&before); err != nil {
|
|
return err
|
|
}
|
|
if err := invalidateCanvasAITargets(r, tx, project, workspace, snapshot.BoardID, before, snapshot.Document); err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := s.attachCanvasAssets(r, tx, project, workspace, snapshot.BoardID, snapshot.Document); err != nil {
|
|
return err
|
|
}
|
|
_, err := tx.ExecContext(r.Context(), `UPDATE materials SET canvas_document=$4,canvas_version=$5 WHERE project=$1 AND workspace_id=$2 AND id=$3`, project, workspace, snapshot.BoardID, snapshot.Document, snapshot.Version)
|
|
return err
|
|
}
|
|
|
|
func (s *service) migrateSubgraphs(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
|
|
}
|
|
var input struct {
|
|
DryRun *bool `json:"dry_run"`
|
|
PlanID string `json:"plan_id"`
|
|
}
|
|
if !decode(w, r, &input) {
|
|
return nil
|
|
}
|
|
if input.DryRun == nil {
|
|
reject(w, 400, "invalid_migration_request")
|
|
return nil
|
|
}
|
|
key := canvasIdempotencyKey(r)
|
|
var expected int64
|
|
if !*input.DryRun {
|
|
if !uuidSlugPattern.MatchString(key) {
|
|
reject(w, 400, "invalid_idempotency_key")
|
|
return nil
|
|
}
|
|
var ok bool
|
|
expected, ok = expectedVersion(w, r)
|
|
if !ok {
|
|
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
|
|
}
|
|
if !*input.DryRun {
|
|
done, err := transactionRetry(w, r, tx, project, account, board, "migrate_subgraphs", key, payloadDigest(input))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if done {
|
|
return nil
|
|
}
|
|
}
|
|
parent, err := loadCanvas(r, tx, project, workspace, board)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
plan := legacyPlan(project, workspace, parent)
|
|
// Preflight runs the actual transform with placeholder canonical IDs. It
|
|
// never inserts materials or relation records.
|
|
placeholders := map[string]string{}
|
|
for _, f := range plan.Frames {
|
|
placeholders[f.FrameID] = "00000000-0000-0000-0000-000000000000"
|
|
}
|
|
if len(placeholders) > 0 {
|
|
if _, _, err := migrateLegacyDocument(parent.Document, placeholders); err != nil {
|
|
plan.Issues = append(plan.Issues, "invalid_geometry_or_capacity")
|
|
}
|
|
}
|
|
var depth int
|
|
err = tx.QueryRowContext(r.Context(), `WITH RECURSIVE ancestors AS (SELECT $3::uuid AS id,0 AS depth UNION ALL SELECT s.parent_board_id,a.depth+1 FROM ancestors a JOIN canvas_subgraphs s ON s.project=$1 AND s.workspace_id=$2 AND s.child_board_id=a.id WHERE a.depth<8 AND s.archived_at IS NULL) SELECT max(depth) FROM ancestors`, project, workspace, board).Scan(&depth)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if depth >= 8 && plan.ChildCount > 0 {
|
|
plan.Issues = append(plan.Issues, "subgraph_depth_exceeded")
|
|
}
|
|
if *input.DryRun {
|
|
setVersion(w, parent.Version)
|
|
reply(w, 200, plan)
|
|
return nil
|
|
}
|
|
if expected != parent.Version || input.PlanID != plan.PlanID {
|
|
reject(w, 412, "version_conflict")
|
|
return nil
|
|
}
|
|
if len(plan.Issues) > 0 {
|
|
reply(w, 413, response{"error": "graph_limit_exceeded", "details": plan.Issues})
|
|
return nil
|
|
}
|
|
before := parent
|
|
children := map[string]string{}
|
|
childInfo := []any{}
|
|
nodeData := canvasNodeData(parent.Document)
|
|
for _, frame := range plan.Frames {
|
|
var child, slug string
|
|
err = tx.QueryRowContext(r.Context(), `INSERT INTO materials(project,workspace_id,kind,title,created_by) VALUES($1,$2,'board',$3,$4) RETURNING id::text,slug`, project, workspace, nodeData[frame.FrameID]["title"], account).Scan(&child, &slug)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
source, _ := json.Marshal(response{"frame_id": frame.FrameID, "source_digest": plan.SourceDigest})
|
|
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_subgraphs(project,workspace_id,parent_board_id,frame_id,child_board_id,migration_source) VALUES($1,$2,$3,$4,$5,$6)`, project, workspace, board, frame.FrameID, child, source)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
children[frame.FrameID] = child
|
|
childInfo = append(childInfo, response{"frame_id": frame.FrameID, "id": child, "slug": slug, "workspace_id": workspace})
|
|
}
|
|
snapshots := []canvasSnapshot{}
|
|
versions := map[string]int64{board: parent.Version}
|
|
if len(children) > 0 {
|
|
migrated, documents, err := migrateLegacyDocument(parent.Document, children)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
parent.Document = migrated
|
|
parent.Version++
|
|
if err = s.saveCanvas(r, tx, project, workspace, parent); err != nil {
|
|
return err
|
|
}
|
|
versions[board] = parent.Version
|
|
for _, f := range plan.Frames {
|
|
child := canvasSnapshot{BoardID: children[f.FrameID], Version: 1, Document: documents[f.FrameID]}
|
|
if err = s.saveCanvas(r, tx, project, workspace, child); err != nil {
|
|
return err
|
|
}
|
|
snapshots = append(snapshots, child)
|
|
versions[child.BoardID] = 1
|
|
}
|
|
}
|
|
var id string
|
|
var receiptExpiry time.Time
|
|
err = tx.QueryRowContext(r.Context(), `SELECT gen_random_uuid()::text,now()+interval '7 days'`).Scan(&id, &receiptExpiry)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
result := response{"transaction_id": id, "parent": parent, "children": snapshots, "child_boards": childInfo, "plan": plan, "migrated_count": len(children), "undo_receipt": response{"transaction_id": id, "valid_for_seconds": 604800, "expires_at": receiptExpiry}}
|
|
responseJSON, _ := json.Marshal(result)
|
|
beforeJSON, _ := json.Marshal([]canvasSnapshot{before})
|
|
versionsJSON, _ := 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,'migrate_subgraphs',$6,$7,200,$8,$9,$10)`, project, workspace, board, id, account, key, payloadDigest(input), responseJSON, beforeJSON, versionsJSON)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
return err
|
|
}
|
|
setVersion(w, parent.Version)
|
|
reply(w, 200, result)
|
|
return nil
|
|
}
|