diff --git a/internal/environment/migrations/1053_owned_public_works.sql b/internal/environment/migrations/1053_owned_public_works.sql new file mode 100644 index 0000000..0bdac3f --- /dev/null +++ b/internal/environment/migrations/1053_owned_public_works.sql @@ -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); diff --git a/internal/environment/owned_public_schema_upgrade_test.go b/internal/environment/owned_public_schema_upgrade_test.go new file mode 100644 index 0000000..87f1c7a --- /dev/null +++ b/internal/environment/owned_public_schema_upgrade_test.go @@ -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) + } + } + }) + } +} diff --git a/internal/environment/store.go b/internal/environment/store.go index ed55226..fdbf6df 100644 --- a/internal/environment/store.go +++ b/internal/environment/store.go @@ -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")