105 lines
3.1 KiB
Go
105 lines
3.1 KiB
Go
package platform
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"net/http"
|
|
)
|
|
|
|
const modelStreamType = "application/vnd.foundry.canvas-model+ndjson;version=1"
|
|
|
|
// The caller captures a coherent model while holding the workspace lock, then
|
|
// releases its transaction before writing. A slow reader must not hold a writer.
|
|
// This streams transfer/encoding; it does not stream database row acquisition.
|
|
func writeCanvasModelStream(ctx context.Context, w http.ResponseWriter, model canvasModel) {
|
|
token := newModelID()
|
|
if token == "" {
|
|
reject(w, 500, "model_stream_unavailable")
|
|
return
|
|
}
|
|
view := model.view(nil)
|
|
// A count covers direct members only; an empty canvas is explicit too.
|
|
counts := map[string]int{}
|
|
for _, canvas := range view.Canvases {
|
|
counts[canvas.ID] = 0
|
|
}
|
|
for _, element := range view.Elements {
|
|
owner := modelString(element["ownerCanvasId"])
|
|
if _, ok := counts[owner]; !ok {
|
|
reject(w, 500, "invalid_model_coverage")
|
|
return
|
|
}
|
|
counts[owner]++
|
|
}
|
|
header := response{"streamVersion": 1, "kind": "header", "sequence": 0, "snapshotToken": token,
|
|
"documentFormatVersion": canvasModelFormat, "projectId": model.Root, "canvases": view.Canvases,
|
|
"versions": view.Versions, "tombstones": view.Tombstones, "removedCanvasIds": view.Removed, "elementCount": len(view.Elements), "canvasElementCounts": counts}
|
|
const maxLine = 8 * 1024 * 1024
|
|
const maxBytes = 64 * 1024 * 1024
|
|
sequence, total := 0, 0
|
|
started := false
|
|
write := func(value response) bool {
|
|
if ctx.Err() != nil {
|
|
return false
|
|
}
|
|
raw, err := json.Marshal(value)
|
|
if err != nil || len(raw) > maxLine || total+len(raw)+1 > maxBytes || sequence >= 4096 {
|
|
if !started {
|
|
reject(w, 413, "model_stream_budget")
|
|
}
|
|
return false
|
|
}
|
|
if !started {
|
|
w.Header().Set("Content-Type", modelStreamType)
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
w.Header().Add("Vary", "Accept")
|
|
w.Header().Set("X-Content-Type-Options", "nosniff")
|
|
w.WriteHeader(http.StatusOK)
|
|
started = true
|
|
}
|
|
total += len(raw) + 1
|
|
if n, writeErr := w.Write(append(raw, '\n')); writeErr != nil || n != len(raw)+1 {
|
|
return false
|
|
}
|
|
sequence++
|
|
// A missing terminal record is an explicit incomplete stream at the client.
|
|
// Never append an unrelated JSON error after streaming has begun.
|
|
return http.NewResponseController(w).Flush() == nil
|
|
}
|
|
if !write(header) {
|
|
return
|
|
}
|
|
batch := []map[string]any{}
|
|
batchBytes := 0
|
|
flush := func() bool {
|
|
if len(batch) == 0 {
|
|
return true
|
|
}
|
|
if !write(response{"streamVersion": 1, "kind": "elements", "sequence": sequence, "snapshotToken": token, "elements": batch}) {
|
|
return false
|
|
}
|
|
batch, batchBytes = []map[string]any{}, 0
|
|
return true
|
|
}
|
|
for _, element := range view.Elements {
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
raw, err := json.Marshal(element)
|
|
if err != nil {
|
|
return
|
|
}
|
|
if len(batch) > 0 && (len(batch) == 64 || batchBytes+len(raw) > 512*1024) {
|
|
if !flush() {
|
|
return
|
|
}
|
|
}
|
|
batch = append(batch, element)
|
|
batchBytes += len(raw)
|
|
}
|
|
if !flush() {
|
|
return
|
|
}
|
|
write(response{"streamVersion": 1, "kind": "complete", "sequence": sequence, "snapshotToken": token, "elementCount": len(view.Elements)})
|
|
}
|