fix: register public work schema update without migration collision

This commit is contained in:
2026-10-06 16:56:29 +08:00
parent 9573e0cbcf
commit 0b4a7e12a1
3 changed files with 83 additions and 1 deletions
@@ -0,0 +1,14 @@
ALTER TABLE creator_work
ADD COLUMN is_public BOOLEAN NOT NULL DEFAULT false;
UPDATE creator_work
SET is_public = true
WHERE source_type = 'competitor';
ALTER TABLE creator_collection_checkpoint
ADD COLUMN public_work_count BIGINT,
ADD COLUMN public_work_count_in_progress BIGINT NOT NULL DEFAULT 0,
ADD CONSTRAINT creator_collection_checkpoint_public_work_count_nonnegative
CHECK (public_work_count IS NULL OR public_work_count >= 0),
ADD CONSTRAINT creator_collection_checkpoint_public_work_count_in_progress_nonnegative
CHECK (public_work_count_in_progress >= 0);
@@ -0,0 +1,65 @@
package environment
import (
"context"
"database/sql"
"os"
"testing"
)
func TestOwnedPublicSchemaUpgrade(t *testing.T) {
databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL")
if databaseURL == "" {
t.Skip("set CREATORHUB_POSTGRES_TEST_URL")
}
for _, upgrade := range []bool{false, true} {
name := "fresh and repeated startup"
if upgrade {
name = "already applied listener migration preserves data"
}
t.Run(name, func(t *testing.T) {
ctx := context.Background()
testURL := isolatedDatabaseURL(t, databaseURL)
store := openFullyMigratedHub(t, ctx, testURL)
store.Close()
db, err := sql.Open("pgx", testURL)
if err != nil {
t.Fatal(err)
}
defer db.Close()
if upgrade {
// Reproduce the deployed schema: 1052 belongs to listener alignment,
// not the public-work fields needed by the account-list query.
_, err := db.Exec(`
DELETE FROM schema_migration WHERE version=1053;
ALTER TABLE creator_work DROP COLUMN IF EXISTS is_public;
ALTER TABLE creator_collection_checkpoint DROP COLUMN IF EXISTS public_work_count;
ALTER TABLE creator_collection_checkpoint DROP COLUMN IF EXISTS public_work_count_in_progress;
INSERT INTO social_account(account_id,name,credential_provider,credential_key,platform,platform_account_key)
VALUES('public-upgrade-account','保留的账号','os_keyring','creatorhub/public-upgrade','douyin','public-upgrade-uid');
INSERT INTO creator_work(work_id,platform,work_key,source_type,source_id,title)
VALUES('public-upgrade-owned','douyin','public-upgrade-owned','owned','public-upgrade-account','保留的自有作品'),
('public-upgrade-competitor','douyin','public-upgrade-competitor','competitor','public-upgrade-competitor','保留的竞品作品');
`)
if err != nil {
t.Fatal(err)
}
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version=1052`, 1)
}
for range 2 {
store = openFullyMigratedHub(t, ctx, testURL)
store.Close()
assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version IN (1052,1053)`, 2)
assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema=current_schema()
AND ((table_name='creator_work' AND column_name='is_public') OR
(table_name='creator_collection_checkpoint' AND column_name IN ('public_work_count','public_work_count_in_progress')))`, 3)
assertListenerSchema(t, db)
if upgrade {
assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE account_id='public-upgrade-account' AND name='保留的账号'`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM creator_work WHERE work_id='public-upgrade-owned' AND title='保留的自有作品' AND NOT is_public`, 1)
assertDatabaseCount(t, db, `SELECT count(*) FROM creator_work WHERE work_id='public-upgrade-competitor' AND title='保留的竞品作品' AND is_public`, 1)
}
}
})
}
}
+4 -1
View File
@@ -197,6 +197,9 @@ var migration1051 string
//go:embed migrations/1052_listener_schema_alignment.sql
var migration1052 string
//go:embed migrations/1053_owned_public_works.sql
var migration1053 string
var (
ErrConflict = errors.New("resource conflicts with existing state")
ErrInvalid = errors.New("invalid environment input")
@@ -338,7 +341,7 @@ func (s *Store) migrate(ctx context.Context) error {
{1029, migration1029}, {1030, migration1030}, {1031, migration1031}, {1032, migration1032}, {1033, migration1033}, {1034, migration1034},
{1035, migration1035}, {1036, migration1036}, {1037, migration1037}, {1038, migration1038}, {1039, migration1039}, {1040, migration1040},
{1041, migration1041}, {1042, migration1042}, {1043, migration1043}, {1044, migration1044}, {1045, migration1045}, {1046, migration1046},
{43, migration043}, {44, migration044}, {1047, migration1047}, {1048, migration1048}, {1049, migration1049}, {1050, migration1050}, {1051, migration1051}, {1052, migration1052}} {
{43, migration043}, {44, migration044}, {1047, migration1047}, {1048, migration1048}, {1049, migration1049}, {1050, migration1050}, {1051, migration1051}, {1052, migration1052}, {1053, migration1053}} {
var applied bool
if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM schema_migration WHERE version = $1)`, migration.version).Scan(&applied); err != nil {
return errors.New("read environment schema migration state")