perf(*): 创作工作台 - 优化首页按需加载与画布交互并修复历史迁移冲突

This commit is contained in:
ddCat committed 2026-09-05 13:24:10 +08:00
1 parent 165b4443ab
commit 58f01d1eb6
107 files changed
+3025 -1089

No files matched your search

+54 -3
View File
@@ -10,7 +10,7 @@ import (
"gorm.io/gorm"
)
const CurrentSchemaVersion int64 = 7
const CurrentSchemaVersion int64 = 8
const baselineSchemaChecksum = "sha256:open-ai-canvas-schema-v1-20260830"
const schemaMigrationAppliedAtIndexChecksum = "sha256:schema-migrations-applied-at-index-v2-20260830"
@@ -19,6 +19,7 @@ const resourceUploadKeyChecksum = "sha256:resource-upload-key-v4-20260901"
const paymentTopupChecksum = "sha256:payment-topup-v5-20260902"
const resourcePlaybackChecksum = "sha256:resource-playback-v6-20260902"
const assetLibraryFoldersChecksum = "sha256:asset-library-folders-v6-20260902"
const logicalModelActiveCodeChecksum = "sha256:logical-model-active-code-v8-20260905"
const postgresSchemaMigrationLockID int64 = 73123910420260830
@@ -52,6 +53,35 @@ var schemaMigrations = []migration{
{version: 5, name: "payment_topup", checksum: paymentTopupChecksum, apply: migrateSchemaV5},
{version: 6, name: "resource_playback_variant", checksum: resourcePlaybackChecksum, apply: migrateSchemaV6},
{version: 7, name: "asset_library_folders", checksum: assetLibraryFoldersChecksum, apply: migrateSchemaV7},
{version: 8, name: "logical_model_active_code", checksum: logicalModelActiveCodeChecksum, apply: migrateSchemaV8},
}
func migrationsForDatabase(db *gorm.DB) ([]migration, error) {
var applied schemaMigration
err := db.First(&applied, "version = ?", 6).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
return schemaMigrations, nil
}
if err != nil {
return nil, fmt.Errorf("读取数据库迁移 6:%w", err)
}
if applied.Name != "asset_library_folders" {
return schemaMigrations, nil
}
legacy := migration{version: 6, name: "asset_library_folders", checksum: assetLibraryFoldersChecksum, apply: migrateSchemaV7}
if err := validateMigrationRecord(applied, legacy); err != nil {
return nil, err
}
plan := append([]migration(nil), schemaMigrations...)
for index, item := range plan {
switch item.version {
case 6:
plan[index] = legacy
case 7:
plan[index] = migration{version: 7, name: "resource_playback_variant", checksum: resourcePlaybackChecksum, apply: migrateSchemaV6}
}
}
return plan, nil
}
func migrateSchemaV2(tx *gorm.DB) error {
@@ -155,6 +185,19 @@ func migrateSchemaV7(tx *gorm.DB) error {
return nil
}
func migrateSchemaV8(tx *gorm.DB) error {
if !tx.Migrator().HasTable(&model.LogicalModel{}) {
return nil
}
if err := tx.Exec("DROP INDEX IF EXISTS idx_logical_models_code").Error; err != nil {
return fmt.Errorf("移除前台模型旧 code 唯一索引:%w", err)
}
if err := tx.Exec("CREATE UNIQUE INDEX idx_logical_models_code ON logical_models(code) WHERE archived_at IS NULL").Error; err != nil {
return fmt.Errorf("创建前台模型活动 code 唯一索引:%w", err)
}
return nil
}
func MigrateSchema(db *gorm.DB) error {
return db.Transaction(func(tx *gorm.DB) error {
if tx.Dialector.Name() == "postgres" {
@@ -165,7 +208,11 @@ func MigrateSchema(db *gorm.DB) error {
if err := tx.AutoMigrate(&schemaMigration{}); err != nil {
return fmt.Errorf("初始化数据库迁移记录:%w", err)
}
for _, item := range schemaMigrations {
plan, err := migrationsForDatabase(tx)
if err != nil {
return err
}
for _, item := range plan {
var applied schemaMigration
err := tx.First(&applied, "version = ?", item.version).Error
if err == nil {
@@ -208,7 +255,11 @@ func ReadSchemaStatus(db *gorm.DB) (SchemaStatus, error) {
}
func validateMigrationRecords(db *gorm.DB) error {
for _, item := range schemaMigrations {
plan, err := migrationsForDatabase(db)
if err != nil {
return err
}
for _, item := range plan {
var applied schemaMigration
if err := db.First(&applied, "version = ?", item.version).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
@@ -0,0 +1,119 @@
package database
import (
"strings"
"testing"
"time"
"infinite-canvas/backend/internal/model"
"gorm.io/gorm"
)
func legacyFolderMigrationDatabase(t *testing.T) *gorm.DB {
t.Helper()
db, err := Open(Config{Driver: "sqlite", DSN: "file:" + t.Name() + "?mode=memory&cache=shared"})
if err != nil {
t.Fatal(err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = sqlDB.Close() })
if err := db.AutoMigrate(&schemaMigration{}); err != nil {
t.Fatal(err)
}
for _, item := range schemaMigrations[:5] {
if err := item.apply(db); err != nil {
t.Fatal(err)
}
if err := db.Create(&schemaMigration{Version: item.version, Name: item.name, Checksum: item.checksum, AppliedAt: time.Now().UTC()}).Error; err != nil {
t.Fatal(err)
}
}
if err := migrateSchemaV7(db); err != nil {
t.Fatal(err)
}
for _, column := range []string{"playback_status", "playback_object_key", "playback_error"} {
if db.Migrator().HasColumn(&model.Resource{}, column) {
if err := db.Migrator().DropColumn(&model.Resource{}, column); err != nil {
t.Fatal(err)
}
}
}
if err := db.Create(&schemaMigration{Version: 6, Name: "asset_library_folders", Checksum: assetLibraryFoldersChecksum, AppliedAt: time.Date(2026, 9, 2, 0, 0, 0, 0, time.UTC)}).Error; err != nil {
t.Fatal(err)
}
return db
}
func TestMigrateSchemaPreservesLegacyFolderMigration(t *testing.T) {
db := legacyFolderMigrationDatabase(t)
var before schemaMigration
if err := db.First(&before, "version = 6").Error; err != nil {
t.Fatal(err)
}
if err := db.Exec("INSERT INTO asset_folders (id, user_id, name) VALUES ('kept-folder', 'owner', 'Keep me')").Error; err != nil {
t.Fatal(err)
}
for attempt := 0; attempt < 2; attempt++ {
if err := MigrateSchema(db); err != nil {
t.Fatal(err)
}
if err := RequireSchemaVersion(db); err != nil {
t.Fatal(err)
}
}
var after schemaMigration
if err := db.First(&after, "version = 6").Error; err != nil {
t.Fatal(err)
}
if before.Name != after.Name || before.Checksum != after.Checksum || !before.AppliedAt.Equal(after.AppliedAt) {
t.Fatalf("historical record changed: before=%+v after=%+v", before, after)
}
var playback schemaMigration
if err := db.First(&playback, "version = 7").Error; err != nil {
t.Fatal(err)
}
if playback.Name != "resource_playback_variant" || playback.Checksum != resourcePlaybackChecksum {
t.Fatalf("unexpected playback migration: %+v", playback)
}
for _, column := range []string{"playback_status", "playback_object_key", "playback_error"} {
if !db.Migrator().HasColumn(&model.Resource{}, column) {
t.Fatalf("missing playback column %s", column)
}
}
var name string
if err := db.Raw("SELECT name FROM asset_folders WHERE id = 'kept-folder'").Scan(&name).Error; err != nil || name != "Keep me" {
t.Fatalf("folder data changed: %q %v", name, err)
}
}
func TestMigrateSchemaRejectsUnknownLegacyLineage(t *testing.T) {
for _, scenario := range []string{"checksum", "name", "version-seven"} {
t.Run(scenario, func(t *testing.T) {
db := legacyFolderMigrationDatabase(t)
switch scenario {
case "checksum":
if err := db.Model(&schemaMigration{}).Where("version = 6").Update("checksum", "unknown").Error; err != nil {
t.Fatal(err)
}
case "name":
if err := db.Model(&schemaMigration{}).Where("version = 6").Update("name", "unknown").Error; err != nil {
t.Fatal(err)
}
case "version-seven":
if err := db.Create(&schemaMigration{Version: 7, Name: "asset_library_folders", Checksum: assetLibraryFoldersChecksum, AppliedAt: time.Now().UTC()}).Error; err != nil {
t.Fatal(err)
}
}
if err := MigrateSchema(db); err == nil || !strings.Contains(err.Error(), "不一致") {
t.Fatalf("expected lineage rejection, got %v", err)
}
if db.Migrator().HasColumn(&model.Resource{}, "playback_status") {
t.Fatal("rejected migration changed resource schema")
}
})
}
}
@@ -37,6 +37,31 @@ func TestMigrateSchemaRecordsAndValidatesVersion(t *testing.T) {
}
}
func TestMigrateSchemaV8AllowsReusingArchivedLogicalModelCode(t *testing.T) {
db, err := Open(Config{Driver: "sqlite", DSN: "file:migration-logical-model-active-code?mode=memory&cache=shared"})
if err != nil {
t.Fatal(err)
}
if err := db.Exec(`CREATE TABLE logical_models (id text PRIMARY KEY, code text NOT NULL, archived_at datetime)`).Error; err != nil {
t.Fatal(err)
}
if err := db.Exec(`CREATE UNIQUE INDEX idx_logical_models_code ON logical_models(code)`).Error; err != nil {
t.Fatal(err)
}
if err := db.Exec(`INSERT INTO logical_models(id, code, archived_at) VALUES ('archived', 'gpt-image-2', CURRENT_TIMESTAMP)`).Error; err != nil {
t.Fatal(err)
}
if err := migrateSchemaV8(db); err != nil {
t.Fatal(err)
}
if err := db.Exec(`INSERT INTO logical_models(id, code, archived_at) VALUES ('active', 'gpt-image-2', NULL)`).Error; err != nil {
t.Fatalf("reusing archived code after migration: %v", err)
}
if err := db.Exec(`INSERT INTO logical_models(id, code, archived_at) VALUES ('duplicate', 'gpt-image-2', NULL)`).Error; err == nil {
t.Fatal("active logical model code must remain unique")
}
}
func TestMigrateSchemaRejectsChecksumMismatch(t *testing.T) {
db, err := Open(Config{Driver: "sqlite", DSN: "file:migration-checksum?mode=memory&cache=shared"})
if err != nil {
+32
View File
@@ -17,6 +17,27 @@ import (
)
func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) {
r.POST("/assets/batch", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {
failService(c, err)
return
}
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 16<<10)
var req struct {
IDs []string `json:"ids"`
}
if err := c.ShouldBindJSON(&req); err != nil {
fail(c, http.StatusBadRequest, err)
return
}
assets, err := svc.UserAssetsByIDs(user.ID, req.IDs)
if err != nil {
failService(c, err)
return
}
ok(c, gin.H{"assets": assets})
})
r.GET("/settings/prompt-templates", func(c *gin.Context) {
user, err := currentUser(c, svc)
if err != nil {
@@ -571,6 +592,17 @@ func RegisterUserDataRoutes(r *gin.RouterGroup, svc *service.Service) {
failService(c, err)
return
}
if c.Query("page") != "" {
page, _ := strconv.Atoi(c.Query("page"))
pageSize, _ := strconv.Atoi(c.Query("page_size"))
result, pageErr := svc.UserCanvasProjectsPage(user.ID, page, pageSize, c.Query("project_id"), c.Query("q"), c.Query("sort"))
if pageErr != nil {
failService(c, pageErr)
return
}
ok(c, result)
return
}
projects, err := svc.UserCanvasProjectSummaries(user.ID)
if err != nil {
failService(c, err)
@@ -0,0 +1,31 @@
package handler
import (
"testing"
"infinite-canvas/backend/internal/service"
"github.com/gin-gonic/gin"
)
func TestUserDataIncrementalRoutesAreRegistered(t *testing.T) {
gin.SetMode(gin.TestMode)
router := gin.New()
RegisterUserDataRoutes(router.Group("/api"), &service.Service{})
wanted := map[string]bool{
"GET /api/canvas-projects": false,
"GET /api/canvas-projects/:id": false,
"POST /api/assets/batch": false,
}
for _, route := range router.Routes() {
key := route.Method + " " + route.Path
if _, exists := wanted[key]; exists {
wanted[key] = true
}
}
for route, found := range wanted {
if !found {
t.Errorf("missing route: %s", route)
}
}
}
@@ -12,7 +12,7 @@ type IDSequence struct {
type LogicalModel struct {
ID string `json:"id" gorm:"primaryKey;size:36"`
Code string `json:"code" gorm:"size:80;uniqueIndex"`
Code string `json:"code" gorm:"size:80;uniqueIndex:idx_logical_models_code,where:archived_at IS NULL"`
Name string `json:"name" gorm:"size:120"`
Icon string `json:"icon" gorm:"size:80"`
Description string `json:"description" gorm:"size:500"`
@@ -89,6 +89,12 @@ func TestArchiveLogicalModelPreservesPublishedHistory(t *testing.T) {
t.Fatalf("%s history was not preserved: %v", label, err)
}
}
replacement := &model.LogicalModel{ID: "LMODEL_REPLACEMENT", Code: item.Code, Name: "Replacement", Capability: "text", Enabled: true, PricePolicy: "unified", BillingMode: "fixed_request", CreatedAt: time.Now(), UpdatedAt: time.Now()}
replacementRevision := &model.LogicalModelRevision{ID: "REVISION_REPLACEMENT", LogicalModelID: replacement.ID, CapabilitySpecJSON: `{"version":1,"capability":"text"}`, DefaultOptionsJSON: `{}`, CreatedAt: time.Now()}
if err := repo.SaveLogicalModelBundle(replacement, replacementRevision, nil, true); err != nil {
t.Fatalf("reusing archived logical model code: %v", err)
}
}
func TestArchiveLogicalModelRejectsActiveTask(t *testing.T) {
@@ -168,6 +168,34 @@ func (r *Repository) ProjectCanvasSummariesPage(userID string, projectID string,
return canvases, total, err
}
func (r *Repository) UserCanvasProjectsPage(userID string, page int, pageSize int, projectID string, search string, sort string) ([]model.CanvasProject, int64, error) {
var projects []model.CanvasProject
var total int64
query := r.db.Model(&model.CanvasProject{}).Where("user_id = ?", userID)
if projectID == "independent" {
query = query.Where("project_id = '' OR project_id IS NULL")
} else if projectID != "" && projectID != "all" {
query = query.Where("project_id = ?", projectID)
}
if search = strings.TrimSpace(search); search != "" {
query = query.Where("LOWER(title) LIKE ?", "%"+strings.ToLower(search)+"%")
}
if err := query.Count(&total).Error; err != nil {
return nil, 0, err
}
order := "updated_at desc, id asc"
if sort == "name" {
order = "title asc, id asc"
} else if sort == "nodes" {
order = "COALESCE(json_array_length(payload_json, '$.nodes'), 0) desc, id asc"
if r.db.Dialector.Name() == "postgres" {
order = "COALESCE(jsonb_array_length(payload_json::jsonb->'nodes'), 0) desc, id asc"
}
}
err := query.Order(order).Offset((page - 1) * pageSize).Limit(pageSize).Find(&projects).Error
return projects, total, err
}
func (r *Repository) ProjectCanvasUnitLinksForCanvases(projectID string, canvasIDs []string) ([]model.CanvasUnitLink, error) {
if len(canvasIDs) == 0 {
return []model.CanvasUnitLink{}, nil
+83 -42
View File
@@ -46,6 +46,7 @@ type canvasGenerationInput struct {
Mask *providerMedia `json:"mask"`
Metadata map[string]interface{} `json:"metadata"`
AgentRequests *agentToolRequests `json:"agentRequests"`
TextOptions canvasTextOptions `json:"textOptions"`
ImageCapability *ImageCapabilityConfig `json:"-"`
StreamText bool `json:"-"` // 分镜请求使用上游 SSE 保活;最终结构仍在流结束后统一校验。
MaxOutputTokens int `json:"-"`
@@ -53,6 +54,11 @@ type canvasGenerationInput struct {
VideoCapability *VideoCapabilityConfig `json:"-"`
}
type canvasTextOptions struct {
Stream *bool `json:"stream"`
Thinking bool `json:"thinking"`
}
type agentToolRequests struct {
Responses map[string]interface{} `json:"responses"`
ChatCompletion map[string]interface{} `json:"chatCompletion"`
@@ -335,8 +341,10 @@ func (s *Service) processCanvasGenerationTask(ctx context.Context, userID string
ctx = withProviderOutboundPolicy(ctx, input.Config)
var textPublisher *taskTextStreamPublisher
if input.Mode == "text" && strings.HasPrefix(taskType, "canvas_text") {
input.StreamText = input.TextOptions.Stream == nil || *input.TextOptions.Stream
}
if input.Mode == "text" && strings.HasPrefix(taskType, "canvas_text") && input.StreamText {
textPublisher = newTaskTextStreamPublisher(s, userID, taskExecutionID(ctx))
input.StreamText = true
input.OnTextDelta = textPublisher.Publish
defer textPublisher.Close()
}
@@ -673,6 +681,8 @@ func parseAgentToolPayload(payload map[string]interface{}, protocol string) (map
switch stringField(item, "type") {
case "text":
result["text"] = result["text"].(string) + stringField(item, "text")
case "thinking":
result["reasoning"] = resultString(result, "reasoning") + firstNonEmptyString(stringField(item, "thinking"), stringField(item, "text"))
case "tool_use":
arguments, err := json.Marshal(item["input"])
if err != nil {
@@ -885,6 +895,8 @@ func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}
switch stringField(block, "type") {
case "text":
p.appendText(stringField(block, "text"))
case "thinking":
p.reasoning.WriteString(firstNonEmptyString(stringField(block, "thinking"), stringField(block, "text")))
case "tool_use":
arguments := ""
if input := block["input"]; input != nil {
@@ -900,6 +912,9 @@ func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}
if stringField(delta, "type") == "text_delta" {
p.appendText(stringField(delta, "text"))
}
if stringField(delta, "type") == "thinking_delta" {
p.reasoning.WriteString(firstNonEmptyString(stringField(delta, "thinking"), stringField(delta, "text")))
}
if stringField(delta, "type") == "input_json_delta" {
p.toolCall(index).arguments += stringField(delta, "partial_json")
}
@@ -909,6 +924,11 @@ func (p *streamingAgentParser) consumeClaudeEvent(payload map[string]interface{}
}
}
func resultString(value map[string]interface{}, key string) string {
text, _ := value[key].(string)
return text
}
func (p *streamingAgentParser) appendText(delta string) {
if delta == "" {
return
@@ -2006,8 +2026,9 @@ func runLegacyTextTask(ctx context.Context, input canvasGenerationInput) (map[st
return nil, err
}
body := map[string]interface{}{"model": input.Config.Model, "input": responseInput}
applyTextThinking(body, input, "responses")
applyTextOutputLimit(body, input.MaxOutputTokens, "max_output_tokens")
text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText, input.OnTextDelta)
result, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText, input.OnTextDelta)
if err != nil {
if !shouldFallbackTextToChat(err) {
return nil, err
@@ -2018,7 +2039,7 @@ func runLegacyTextTask(ctx context.Context, input canvasGenerationInput) (map[st
}
return nil, fmt.Errorf("文本接口请求失败:Responses API %v;Chat Completions %v", err, chatErr)
}
return map[string]interface{}{"mode": "text", "text": text}, nil
return providerTextTaskResult(result), nil
}
func runResponsesTextTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) {
@@ -2027,12 +2048,13 @@ func runResponsesTextTask(ctx context.Context, input canvasGenerationInput) (map
return nil, err
}
body := map[string]interface{}{"model": input.Config.Model, "input": responseInput}
applyTextThinking(body, input, "responses")
applyTextOutputLimit(body, input.MaxOutputTokens, "max_output_tokens")
text, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText, input.OnTextDelta)
result, err := requestTextProvider(ctx, input.Config, "/responses", body, "responses", input.StreamText, input.OnTextDelta)
if err != nil {
return nil, err
}
return map[string]interface{}{"mode": "text", "text": text}, nil
return providerTextTaskResult(result), nil
}
func runChatCompletionsTextTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) {
@@ -2047,12 +2069,13 @@ func runChatCompletionsTextTask(ctx context.Context, input canvasGenerationInput
}
messages = append(messages, map[string]interface{}{"role": "user", "content": userContent})
body := map[string]interface{}{"model": input.Config.Model, "messages": messages}
applyTextThinking(body, input, "chat-completion")
applyTextOutputLimit(body, input.MaxOutputTokens, "max_tokens")
text, err := requestTextProvider(ctx, input.Config, "/chat/completions", body, "chat-completion", input.StreamText, input.OnTextDelta)
result, err := requestTextProvider(ctx, input.Config, "/chat/completions", body, "chat-completion", input.StreamText, input.OnTextDelta)
if err != nil {
return nil, err
}
return map[string]interface{}{"mode": "text", "text": text}, nil
return providerTextTaskResult(result), nil
}
func runClaudeTextTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) {
@@ -2073,14 +2096,42 @@ func runClaudeTextTask(ctx context.Context, input canvasGenerationInput) (map[st
maxTokens = input.MaxOutputTokens
}
body := map[string]interface{}{"model": input.Config.Model, "max_tokens": maxTokens, "messages": messages}
applyTextThinking(body, input, "claude-api")
if systemPrompt := strings.TrimSpace(input.Config.SystemPrompt); systemPrompt != "" {
body["system"] = systemPrompt
}
text, err := requestTextProvider(ctx, input.Config, "/messages", body, "claude-api", input.StreamText, input.OnTextDelta)
result, err := requestTextProvider(ctx, input.Config, "/messages", body, "claude-api", input.StreamText, input.OnTextDelta)
if err != nil {
return nil, err
}
return map[string]interface{}{"mode": "text", "text": text}, nil
return providerTextTaskResult(result), nil
}
func applyTextThinking(body map[string]interface{}, input canvasGenerationInput, protocol string) {
if !input.TextOptions.Thinking {
return
}
switch protocol {
case "responses":
body["reasoning"] = map[string]interface{}{"effort": "medium", "summary": "auto"}
case "chat-completion":
body["reasoning_effort"] = "medium"
case "claude-api":
body["thinking"] = map[string]interface{}{"type": "enabled", "budget_tokens": 1024}
}
}
type providerTextResult struct {
Text string
Reasoning string
}
func providerTextTaskResult(result providerTextResult) map[string]interface{} {
payload := map[string]interface{}{"mode": "text", "text": result.Text}
if strings.TrimSpace(result.Reasoning) != "" {
payload["reasoning"] = result.Reasoning
}
return payload
}
func applyTextOutputLimit(body map[string]interface{}, limit int, field string) {
@@ -4564,51 +4615,41 @@ func runSeedanceAgentPlanVideoTask(ctx context.Context, input canvasGenerationIn
return nil, fmt.Errorf("%s视频生成超时", providerName)
}
func requestTextProvider(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, stream bool, onDelta func(string)) (string, error) {
func requestTextProvider(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, stream bool, onDelta func(string)) (providerTextResult, error) {
if stream {
metadata, _ := ctx.Value(providerAnalyticsKey{}).(providerAnalyticsContext)
if protocol == "chat-completion" && metadata.BillingMode == "token" {
if err := ensureChatCompletionStreamUsage(body); err != nil {
return "", err
}
}
return postStreamingText(ctx, config, path, body, protocol, onDelta)
return postStreamingTextResult(ctx, config, path, body, protocol, onDelta)
}
var payload map[string]interface{}
if err := postJSON(ctx, config, path, body, &payload); err != nil {
return "", err
return providerTextResult{}, err
}
text := extractTextPayload(payload, protocol)
if text == "" {
return "", errors.New("文本接口没有返回内容")
parsed, err := parseAgentToolPayload(payload, protocol)
if err != nil {
return providerTextResult{}, err
}
return text, nil
result := providerTextResult{Text: stringField(parsed, "text"), Reasoning: stringField(parsed, "reasoning")}
if result.Text == "" {
return providerTextResult{}, errors.New("文本接口没有返回内容")
}
return result, nil
}
func postStreamingText(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, onDelta func(string)) (string, error) {
// 只把分镜规划/修复切到上游 SSE,完整 JSON 仍在流结束后校验,避免半截结构污染画布。
body["stream"] = true
parser := newStreamingTextDeltaParser(protocol, onDelta)
data, mimeType, err := postStreamingBinary(ctx, config, path, body, parser.consume)
result, err := postStreamingTextResult(ctx, config, path, body, protocol, onDelta)
return result.Text, err
}
func postStreamingTextResult(ctx context.Context, config providerConfig, path string, body map[string]interface{}, protocol string, onDelta func(string)) (providerTextResult, error) {
// 文本创作与 Agent 共用同一套 SSE 解析,确保正文、推理摘要和供应商错误语义一致。
parsed, err := postStreamingAgent(ctx, config, path, body, protocol, onDelta)
if err != nil {
return "", err
return providerTextResult{}, err
}
parser.flush()
if !strings.Contains(strings.ToLower(mimeType), "event-stream") {
var payload map[string]interface{}
if err := json.Unmarshal(data, &payload); err != nil {
return "", fmt.Errorf("流式文本接口返回格式无效:%w", err)
}
if err := validateTextPayload(payload); err != nil {
return "", err
}
text := extractTextPayload(payload, protocol)
if text == "" {
return "", errors.New("文本接口没有返回内容")
}
return text, nil
result := providerTextResult{Text: stringField(parsed, "text"), Reasoning: stringField(parsed, "reasoning")}
if result.Text == "" {
return providerTextResult{}, errors.New("流式文本接口没有返回内容")
}
return parseTextEventStream(data, protocol)
return result, nil
}
func extractTextPayload(payload map[string]interface{}, protocol string) string {
+19
View File
@@ -489,6 +489,25 @@ data: [DONE]
}
}
func TestTextThinkingOptionsAndReasoningParsing(t *testing.T) {
input := canvasGenerationInput{TextOptions: canvasTextOptions{Thinking: true}}
responsesBody := map[string]interface{}{}
applyTextThinking(responsesBody, input, "responses")
reasoning, _ := responsesBody["reasoning"].(map[string]interface{})
if reasoning["effort"] != "medium" || reasoning["summary"] != "auto" {
t.Fatalf("responses reasoning options = %#v", reasoning)
}
chatBody := map[string]interface{}{}
applyTextThinking(chatBody, input, "chat-completion")
if chatBody["reasoning_effort"] != "medium" {
t.Fatalf("chat reasoning options = %#v", chatBody)
}
result, err := parseAgentToolPayload(map[string]interface{}{"choices": []interface{}{map[string]interface{}{"message": map[string]interface{}{"content": "正文", "reasoning_content": "推理摘要"}}}}, "chat-completion")
if err != nil || result["text"] != "正文" || result["reasoning"] != "推理摘要" {
t.Fatalf("parsed reasoning result = %#v, err = %v", result, err)
}
}
func TestStreamingAgentParserReassemblesChatToolCallsAcrossChunks(t *testing.T) {
var deltas strings.Builder
parser := newStreamingAgentParser("chat-completion", func(delta string) {
+43 -4
View File
@@ -31,6 +31,7 @@ type TaskSummary struct {
ErrorCode string `json:"errorCode,omitempty"`
PreviewURL string `json:"previewUrl,omitempty"`
PreviewKind string `json:"previewKind,omitempty"`
PreviewPosterURL string `json:"previewPosterUrl,omitempty"`
Attempts int `json:"attempts"`
StartedAt *time.Time `json:"startedAt"`
CompletedAt *time.Time `json:"completedAt"`
@@ -98,7 +99,7 @@ func taskSummaryForOutput(task model.Task) TaskSummary {
if isContentModerationFailure(task.Error) {
errorCode = contentModerationErrorCode
}
previewURL, previewKind := taskMediaPreview(task.ResultJSON, task.Type)
previewURL, previewKind, previewPosterURL := taskMediaPreviewWithPoster(task.ResultJSON, task.Type)
return TaskSummary{
ID: task.ID,
SessionID: task.SessionID,
@@ -120,6 +121,7 @@ func taskSummaryForOutput(task model.Task) TaskSummary {
ErrorCode: errorCode,
PreviewURL: previewURL,
PreviewKind: previewKind,
PreviewPosterURL: previewPosterURL,
Attempts: task.Attempts,
StartedAt: task.StartedAt,
CompletedAt: task.CompletedAt,
@@ -179,18 +181,55 @@ func taskClientContext(raw string) *TaskClientContext {
// 列表只暴露首个可访问媒体地址,避免把完整生成结果和内嵌数据带回前端。
func taskMediaPreview(raw string, taskType string) (string, string) {
previewURL, previewKind, _ := taskMediaPreviewWithPoster(raw, taskType)
return previewURL, previewKind
}
func taskMediaPreviewWithPoster(raw string, taskType string) (string, string, string) {
if strings.TrimSpace(raw) == "" {
return "", ""
return "", "", ""
}
var payload any
if json.Unmarshal([]byte(raw), &payload) != nil {
return "", ""
return "", "", ""
}
defaultKind := "image"
if strings.Contains(strings.ToLower(taskType), "video") {
defaultKind = "video"
}
return findTaskMediaPreview(payload, defaultKind)
previewURL, previewKind := findTaskMediaPreview(payload, defaultKind)
posterURL := findTaskMediaPoster(payload)
if previewKind == "image" && posterURL == "" {
posterURL = previewURL
}
return previewURL, previewKind, posterURL
}
func findTaskMediaPoster(value any) string {
switch item := value.(type) {
case []any:
for _, child := range item {
if posterURL := findTaskMediaPoster(child); posterURL != "" {
return posterURL
}
}
case map[string]any:
for _, key := range []string{"posterUrl", "posterURL", "thumbnailUrl", "thumbnailURL", "coverUrl", "coverURL", "poster", "thumbnail", "cover"} {
child, exists := item[key]
if !exists {
continue
}
if previewURL, previewKind := findTaskMediaPreview(child, "image"); previewURL != "" && previewKind == "image" {
return previewURL
}
}
for _, child := range item {
if posterURL := findTaskMediaPoster(child); posterURL != "" {
return posterURL
}
}
}
return ""
}
func findTaskMediaPreview(value any, hint string) (string, string) {
@@ -41,6 +41,16 @@ func TestTaskMediaPreviewUsesSafeMediaURLs(t *testing.T) {
}
}
func TestTaskMediaPreviewSeparatesVideoPoster(t *testing.T) {
previewURL, previewKind, posterURL := taskMediaPreviewWithPoster(`{"video":{"url":"https://cdn.example.com/output.mp4","posterUrl":"https://cdn.example.com/poster.webp"}}`, "canvas_video")
if previewURL != "https://cdn.example.com/output.mp4" || previewKind != "video" || posterURL != "https://cdn.example.com/poster.webp" {
t.Fatalf("unexpected video preview: url=%q kind=%q poster=%q", previewURL, previewKind, posterURL)
}
if _, _, posterURL := taskMediaPreviewWithPoster(`{"video":{"url":"https://cdn.example.com/output.mp4","posterUrl":"file:///tmp/poster.png"}}`, "canvas_video"); posterURL != "" {
t.Fatalf("unsafe local poster was exposed: %q", posterURL)
}
}
func TestTaskClientContextRequiresCreatePageMetadata(t *testing.T) {
valid := taskClientContext(`{"metadata":{"source":"create-page","conversationId":"conversation-1","messageId":"message-1","batchIndex":2,"batchCount":4}}`)
if valid == nil || valid.ConversationID != "conversation-1" || valid.BatchIndex != 2 {
+110
View File
@@ -0,0 +1,110 @@
package service
import (
"encoding/json"
"strings"
"time"
)
func (s *Service) UserAssetsByIDs(userID string, ids []string) ([]json.RawMessage, error) {
if len(ids) > 100 {
return nil, BadAuthRequest("每次最多读取 100 个素材")
}
unique := make([]string, 0, len(ids))
seen := make(map[string]bool, len(ids))
for _, id := range ids {
id = strings.TrimSpace(id)
if id == "" || len(id) > 80 {
return nil, BadAuthRequest("素材 ID 无效")
}
if !seen[id] {
seen[id] = true
unique = append(unique, id)
}
}
assets, err := s.repo.AssetsForUserIDs(userID, unique)
if err != nil {
return nil, err
}
result := make([]json.RawMessage, 0, len(assets))
for _, asset := range assets {
result = append(result, json.RawMessage(asset.PayloadJSON))
}
return result, nil
}
type CanvasLibrarySummary struct {
ID string `json:"id"`
ProjectID string `json:"projectId,omitempty"`
Title string `json:"title"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
NodeCount int `json:"nodeCount"`
PreviewNodes []map[string]any `json:"previewNodes"`
}
type CanvasLibraryPage struct {
Projects []CanvasLibrarySummary `json:"projects"`
Page int `json:"page"`
PageSize int `json:"pageSize"`
Total int64 `json:"total"`
HasMore bool `json:"hasMore"`
}
func (s *Service) UserCanvasProjectsPage(userID string, page int, pageSize int, projectID string, search string, sort string) (CanvasLibraryPage, error) {
if page < 1 {
page = 1
}
if page > 1000000 {
return CanvasLibraryPage{}, BadAuthRequest("页码超出范围")
}
if pageSize < 1 {
pageSize = 40
}
if pageSize > 50 {
pageSize = 50
}
projects, total, err := s.repo.UserCanvasProjectsPage(userID, page, pageSize, projectID, search, sort)
if err != nil {
return CanvasLibraryPage{}, err
}
result := CanvasLibraryPage{Projects: make([]CanvasLibrarySummary, 0, len(projects)), Page: page, PageSize: pageSize, Total: total, HasMore: int64(page)*int64(pageSize) < total}
for _, project := range projects {
var document struct {
Nodes []map[string]any `json:"nodes"`
}
if err := json.Unmarshal([]byte(project.PayloadJSON), &document); err != nil {
return CanvasLibraryPage{}, err
}
preview := make([]map[string]any, 0, 4)
for _, node := range document.Nodes {
if len(preview) == 4 {
break
}
if node["type"] != "image" && node["type"] != "video" {
continue
}
item := map[string]any{"position": map[string]int{"x": 0, "y": 0}}
for _, key := range []string{"id", "type", "title"} {
if value, ok := node[key].(string); ok {
item[key] = string([]rune(value)[:min(len([]rune(value)), 256)])
}
}
for _, key := range []string{"width", "height"} {
if value, ok := node[key].(float64); ok {
item[key] = value
}
}
metadata := map[string]any{}
if original, ok := node["metadata"].(map[string]any); ok {
if key, ok := original["storageKey"].(string); ok && len(key) <= 512 {
metadata["storageKey"] = key
}
}
item["metadata"] = metadata
preview = append(preview, item)
}
result.Projects = append(result.Projects, CanvasLibrarySummary{ID: project.ID, ProjectID: project.ProjectID, Title: project.Title, CreatedAt: project.CreatedAt, UpdatedAt: project.UpdatedAt, NodeCount: len(document.Nodes), PreviewNodes: preview})
}
return result, nil
}
@@ -0,0 +1,68 @@
package service
import (
"encoding/json"
"fmt"
"strings"
"testing"
"infinite-canvas/backend/internal/model"
)
func TestUserCanvasLibraryPageIsBoundedAndScoped(t *testing.T) {
service, db := newProjectWorkbenchReadTestService(t)
for index := 0; index < 55; index++ {
project := model.CanvasProject{ID: fmt.Sprintf("canvas-%02d", index), UserID: "owner", Title: fmt.Sprintf("Title %02d", index), PayloadJSON: `{"nodes":[{"id":"image","type":"image","metadata":{"storageKey":"resource:preview","content":"secret-inline-media","prompt":"secret-prompt"}}],"chatSessions":["secret-chat"]}`}
if err := db.Create(&project).Error; err != nil {
t.Fatal(err)
}
}
if err := db.Create(&model.CanvasProject{ID: "foreign", UserID: "other", Title: "Foreign", PayloadJSON: `{}`}).Error; err != nil {
t.Fatal(err)
}
page, err := service.UserCanvasProjectsPage("owner", 1, 999, "", "", "name")
if err != nil {
t.Fatal(err)
}
if len(page.Projects) != 50 || page.Total != 55 || !page.HasMore || page.PageSize != 50 {
t.Fatalf("unexpected pagination: %+v", page)
}
encoded, err := json.Marshal(page)
if err != nil {
t.Fatal(err)
}
if strings.Contains(string(encoded), "secret") || strings.Contains(string(encoded), "foreign") {
t.Fatal("page exposed full payload or another account")
}
last, err := service.UserCanvasProjectsPage("owner", 2, 50, "", "", "name")
if err != nil || len(last.Projects) != 5 || last.HasMore {
t.Fatalf("last page: %+v %v", last, err)
}
filtered, err := service.UserCanvasProjectsPage("owner", 1, 40, "", "Title 04", "name")
if err != nil || filtered.Total != 1 || filtered.Projects[0].NodeCount != 1 {
t.Fatalf("filter: %+v %v", filtered, err)
}
}
func TestUserAssetsBatchIsBoundedAndScoped(t *testing.T) {
service, db := newProjectWorkbenchReadTestService(t)
for _, owner := range []string{"owner", "other"} {
if err := db.Create(&model.Asset{ID: owner, UserID: owner, PayloadJSON: fmt.Sprintf(`{"id":%q}`, owner)}).Error; err != nil {
t.Fatal(err)
}
}
assets, err := service.UserAssetsByIDs("owner", []string{"owner", "owner", "other", "missing"})
if err != nil || len(assets) != 1 || string(assets[0]) != `{"id":"owner"}` {
t.Fatalf("unexpected assets: %s %v", assets, err)
}
empty, err := service.UserAssetsByIDs("owner", nil)
if err != nil || len(empty) != 0 {
t.Fatalf("empty IDs must not list account: %s %v", empty, err)
}
if _, err := service.UserAssetsByIDs("owner", make([]string, 101)); err == nil {
t.Fatal("expected batch limit")
}
if _, err := service.UserAssetsByIDs("owner", []string{" "}); err == nil {
t.Fatal("expected invalid ID error")
}
}