修复了一点bug,然后增加了图层拆分更多的功能
This commit is contained in:
1 parent
1305a9cdd8
commit
ce8372e878
31 files changed
+2567
-391
No files matched your search
@@ -12,6 +12,7 @@ WITH required_table_privileges(table_name, privilege_name) AS (
|
||||
VALUES
|
||||
('assets', 'SELECT'), ('assets', 'INSERT'), ('assets', 'DELETE'),
|
||||
('generation_jobs', 'SELECT'), ('generation_jobs', 'INSERT'), ('generation_jobs', 'UPDATE'), ('generation_jobs', 'DELETE'),
|
||||
('seedream_layer_compositions', 'SELECT'), ('seedream_layer_compositions', 'INSERT'), ('seedream_layer_compositions', 'UPDATE'), ('seedream_layer_compositions', 'DELETE'),
|
||||
('usage_events', 'SELECT'), ('usage_events', 'INSERT'), ('usage_events', 'UPDATE'),
|
||||
('projects', 'SELECT'), ('projects', 'UPDATE'),
|
||||
('image_templates', 'SELECT'), ('image_templates', 'INSERT'), ('image_templates', 'UPDATE'), ('image_templates', 'DELETE'),
|
||||
|
||||
@@ -266,6 +266,8 @@ func (rows *jobRows) Scan(dest ...any) error {
|
||||
*target = value.(int)
|
||||
case *int64:
|
||||
*target = value.(int64)
|
||||
case *bool:
|
||||
*target = value.(bool)
|
||||
case *time.Time:
|
||||
if value != nil {
|
||||
*target = value.(time.Time)
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/layercompositions"
|
||||
)
|
||||
|
||||
const FindLayerCompositionSQL = `SELECT job_id, owner_id, base_asset_id, base_visible, layers, version, created_at, updated_at
|
||||
FROM public.seedream_layer_compositions
|
||||
WHERE owner_id = $1::text AND job_id = $2::text
|
||||
LIMIT 1`
|
||||
|
||||
const SaveLayerCompositionSQL = `WITH updated AS (
|
||||
UPDATE public.seedream_layer_compositions SET
|
||||
base_asset_id = $3::text,
|
||||
base_visible = $4::boolean,
|
||||
layers = $5::jsonb,
|
||||
version = seedream_layer_compositions.version + 1,
|
||||
updated_at = now()
|
||||
WHERE job_id = $1::text
|
||||
AND owner_id = $2::text
|
||||
AND version = $6::bigint
|
||||
AND $6::bigint > 0
|
||||
RETURNING job_id, owner_id, base_asset_id, base_visible, layers, version, created_at, updated_at
|
||||
), inserted AS (
|
||||
INSERT INTO public.seedream_layer_compositions (
|
||||
job_id, owner_id, base_asset_id, base_visible, layers, version, created_at, updated_at
|
||||
)
|
||||
SELECT $1::text, $2::text, $3::text, $4::boolean, $5::jsonb, 1, now(), now()
|
||||
WHERE $6::bigint = 0
|
||||
ON CONFLICT (job_id) DO NOTHING
|
||||
RETURNING job_id, owner_id, base_asset_id, base_visible, layers, version, created_at, updated_at
|
||||
)
|
||||
SELECT job_id, owner_id, base_asset_id, base_visible, layers, version, created_at, updated_at FROM updated
|
||||
UNION ALL
|
||||
SELECT job_id, owner_id, base_asset_id, base_visible, layers, version, created_at, updated_at FROM inserted
|
||||
LIMIT 1`
|
||||
|
||||
func (db *Database) FindLayerComposition(ctx context.Context, ownerID, jobID string) (layercompositions.Composition, bool, error) {
|
||||
if err := db.available(); err != nil {
|
||||
return layercompositions.Composition{}, false, err
|
||||
}
|
||||
rows, err := db.querier.Query(ctx, FindLayerCompositionSQL, ownerID, jobID)
|
||||
if err != nil {
|
||||
return layercompositions.Composition{}, false, fmt.Errorf("find layer composition: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
if !rows.Next() {
|
||||
if err := rows.Err(); err != nil {
|
||||
return layercompositions.Composition{}, false, fmt.Errorf("read layer composition: %w", err)
|
||||
}
|
||||
return layercompositions.Composition{}, false, nil
|
||||
}
|
||||
composition, err := scanLayerComposition(rows)
|
||||
if err != nil {
|
||||
return layercompositions.Composition{}, false, err
|
||||
}
|
||||
return composition, true, rows.Err()
|
||||
}
|
||||
|
||||
func (db *Database) SaveLayerComposition(ctx context.Context, composition layercompositions.Composition, expectedVersion int64) (layercompositions.Composition, error) {
|
||||
if err := db.available(); err != nil {
|
||||
return layercompositions.Composition{}, err
|
||||
}
|
||||
layers, err := json.Marshal(composition.Layers)
|
||||
if err != nil {
|
||||
return layercompositions.Composition{}, fmt.Errorf("encode layer composition: %w", err)
|
||||
}
|
||||
rows, err := db.querier.Query(ctx, SaveLayerCompositionSQL,
|
||||
composition.JobID, composition.OwnerID, composition.BaseAssetID,
|
||||
composition.BaseVisible, layers, expectedVersion,
|
||||
)
|
||||
if err != nil {
|
||||
return layercompositions.Composition{}, fmt.Errorf("save layer composition: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
if !rows.Next() {
|
||||
if err := rows.Err(); err != nil {
|
||||
return layercompositions.Composition{}, fmt.Errorf("read saved layer composition: %w", err)
|
||||
}
|
||||
return layercompositions.Composition{}, layercompositions.ErrConflict
|
||||
}
|
||||
saved, err := scanLayerComposition(rows)
|
||||
if err != nil {
|
||||
return layercompositions.Composition{}, err
|
||||
}
|
||||
return saved, rows.Err()
|
||||
}
|
||||
|
||||
func scanLayerComposition(rows Rows) (layercompositions.Composition, error) {
|
||||
var composition layercompositions.Composition
|
||||
var rawLayers []byte
|
||||
if err := rows.Scan(
|
||||
&composition.JobID, &composition.OwnerID, &composition.BaseAssetID,
|
||||
&composition.BaseVisible, &rawLayers, &composition.Version,
|
||||
&composition.CreatedAt, &composition.UpdatedAt,
|
||||
); err != nil {
|
||||
return layercompositions.Composition{}, fmt.Errorf("scan layer composition: %w", err)
|
||||
}
|
||||
if err := json.Unmarshal(rawLayers, &composition.Layers); err != nil {
|
||||
return layercompositions.Composition{}, fmt.Errorf("decode layer composition: %w", err)
|
||||
}
|
||||
if composition.Layers == nil {
|
||||
composition.Layers = []layercompositions.Layer{}
|
||||
}
|
||||
return composition, nil
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
package postgres
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.nianxx.cn/wangxuming/NianAIGC/backend/internal/layercompositions"
|
||||
)
|
||||
|
||||
func TestLayerCompositionAdapterScopesReadsAndUsesOptimisticSave(t *testing.T) {
|
||||
now := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC)
|
||||
row := []any{"job-1", "owner", "base", true, []byte(`[{"id":"layer-1","assetId":"asset-1","name":"主体","visible":true,"box":[100,100,900,900],"rotation":0,"source":"provider"}]`), int64(2), now, now}
|
||||
querier := &jobQuerier{rows: &jobRows{rows: [][]any{row}}}
|
||||
database := NewDatabase(Config{Backend: BackendPostgres}, querier)
|
||||
|
||||
composition, found, err := database.FindLayerComposition(context.Background(), "owner", "job-1")
|
||||
if err != nil || !found || composition.Version != 2 || len(composition.Layers) != 1 {
|
||||
t.Fatalf("composition=%#v found=%v err=%v", composition, found, err)
|
||||
}
|
||||
if querier.query != FindLayerCompositionSQL || !reflect.DeepEqual(querier.args, []any{"owner", "job-1"}) {
|
||||
t.Fatalf("query=%q args=%#v", querier.query, querier.args)
|
||||
}
|
||||
|
||||
savedRow := append([]any(nil), row...)
|
||||
savedRow[5] = int64(3)
|
||||
querier.rows = &jobRows{rows: [][]any{savedRow}}
|
||||
saved, err := database.SaveLayerComposition(context.Background(), layercompositions.Composition{
|
||||
JobID: "job-1", OwnerID: "owner", BaseAssetID: "base", BaseVisible: true,
|
||||
Layers: []layercompositions.Layer{{ID: "layer-1", AssetID: "asset-1", Name: "主体", Visible: true, Box: layercompositions.Box{100, 100, 900, 900}, Source: "provider"}},
|
||||
}, 2)
|
||||
if err != nil || saved.Version != 3 {
|
||||
t.Fatalf("saved=%#v err=%v", saved, err)
|
||||
}
|
||||
if querier.query != SaveLayerCompositionSQL || len(querier.args) != 6 || querier.args[5] != int64(2) {
|
||||
t.Fatalf("query=%q args=%#v", querier.query, querier.args)
|
||||
}
|
||||
for _, fragment := range []string{"WITH updated AS", "WHERE $6::bigint = 0", "version = seedream_layer_compositions.version + 1", "AND version = $6::bigint", "ON CONFLICT (job_id) DO NOTHING"} {
|
||||
if !strings.Contains(SaveLayerCompositionSQL, fragment) {
|
||||
t.Fatalf("upsert omits %q", fragment)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLayerCompositionSaveSQLKeepsExistingVersionedUpdatesReachable(t *testing.T) {
|
||||
// A top-level INSERT ... SELECT ... WHERE expectedVersion = 0 filters the
|
||||
// source row before ON CONFLICT can run. In that shape every normal update
|
||||
// (expectedVersion > 0) returns no row and is reported as a false conflict.
|
||||
if strings.Contains(SaveLayerCompositionSQL, "SELECT $1::text, $2::text, $3::text, $4::boolean, $5::jsonb, 1, now(), now()\nWHERE $6::bigint = 0\nON CONFLICT") {
|
||||
t.Fatal("versioned updates are unreachable because the INSERT source is filtered before ON CONFLICT")
|
||||
}
|
||||
for _, fragment := range []string{
|
||||
"WITH updated AS",
|
||||
"version = $6::bigint",
|
||||
"$6::bigint > 0",
|
||||
"WHERE $6::bigint = 0",
|
||||
"UNION ALL",
|
||||
} {
|
||||
if !strings.Contains(SaveLayerCompositionSQL, fragment) {
|
||||
t.Fatalf("save query does not preserve both update and create paths; missing %q", fragment)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLayerCompositionAdapterReturnsConflictWhenVersionUpdateMatchesNoRow(t *testing.T) {
|
||||
querier := &jobQuerier{rows: &jobRows{}}
|
||||
database := NewDatabase(Config{Backend: BackendPostgres}, querier)
|
||||
_, err := database.SaveLayerComposition(context.Background(), layercompositions.Composition{JobID: "job-1", OwnerID: "owner", BaseAssetID: "base", Layers: []layercompositions.Layer{}}, 4)
|
||||
if !errors.Is(err, layercompositions.ErrConflict) {
|
||||
t.Fatalf("err=%v", err)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user