creator-hub/internal/environment/migration_test.go

package environment

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"net/url"
	"os"
	"slices"
	"strings"
	"testing"
	"time"

	"git.ipao.vip/rogee/creator-hub/internal/account"
	"git.ipao.vip/rogee/creator-hub/internal/taskstate"
)

func TestUnifiedAccountMigration(t *testing.T) {
	databaseURL := os.Getenv("CREATORHUB_POSTGRES_TEST_URL")
	if databaseURL == "" {
		t.Skip("set CREATORHUB_POSTGRES_TEST_URL to run PostgreSQL integration coverage")
	}

	t.Run("fresh database", 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()
		assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version BETWEEN 1 AND 17`, 17)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name IN ('social_account', 'browser_env', 'browser_version', 'network_exit', 'environment_binding')`, 5)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.tables WHERE table_schema = current_schema() AND table_name = 'browser_image'`, 0)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'browser_version' AND column_name = 'browser_path'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'social_account' AND column_name IN ('name', 'tags')`, 2)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'social_account' AND column_name = 'cookies'`, 0)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'environment_binding' AND column_name = 'runtime_cleanup_pending'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'network_exit' AND column_name IN ('username', 'password')`, 2)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND table_name = 'network_exit' AND column_name = 'credential_reference_id'`, 0)

		store = openFullyMigratedHub(t, ctx, testURL)
		store.Close()
		assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version BETWEEN 1 AND 17`, 17)
	})

	t.Run("legacy migration 013 without account secrets is repaired forward", func(t *testing.T) {
		ctx := context.Background()
		testURL := isolatedDatabaseURL(t, databaseURL)
		db := openLegacyAccountCreationSchema(t, ctx, testURL)
		defer db.Close()

		store, err := Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		store.Close()
		assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version = 14`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema()
			AND table_name = 'social_account' AND column_name = 'cookies'`, 0)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema()
			AND table_name = 'social_account' AND column_name = 'credential_reference_id' AND is_nullable = 'NO'`, 1)
	})

	t.Run("legacy migration 013 with account secrets blocks upgrade", func(t *testing.T) {
		ctx := context.Background()
		testURL := isolatedDatabaseURL(t, databaseURL)
		db := openLegacyAccountCreationSchema(t, ctx, testURL)
		defer db.Close()
		if _, err := db.Exec(`
			INSERT INTO social_account
				(id, name, platform, platform_account_key, tags, cookies, authorization_kind, authorization_status, status)
			VALUES ('legacy-secret', 'Legacy', 'douyin', 'legacy-secret', ARRAY[]::text[],
				'sessionid=migration-secret', 'owned', 'authorized', 'paused')`); err != nil {
			t.Fatal(err)
		}

		_, err := Open(ctx, testURL)
		if err == nil || !strings.Contains(err.Error(), "apply environment schema migration 14") || strings.Contains(err.Error(), "migration-secret") {
			t.Fatalf("unsafe legacy migration was not blocked safely: %v", err)
		}
		assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version = 14`, 0)
	})

	t.Run("previous migration 011 already applied", 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 _, err := db.Exec(`
			INSERT INTO credential_reference (id, provider, reference_key)
			VALUES ('credential-recovery-upgrade', 'os_keyring', 'creatorhub/recovery-upgrade');
			INSERT INTO social_account
				(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status)
			VALUES ('recovery-upgrade', 'credential-recovery-upgrade', 'mock', 'recovery-upgrade', 'owned', 'authorized');
			INSERT INTO content_draft (id, account_id, version, content)
			VALUES ('draft-recovery-upgrade', 'recovery-upgrade', 1, 'legacy');
			INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
			VALUES ('confirmation-recovery-upgrade', 'recovery-upgrade', 1, 'draft-recovery-upgrade', 1, 1);
			INSERT INTO operation_task
				(id, idempotency_key, account_id, account_version, draft_id, draft_version,
				 confirmation_id, confirmation_version, state, hold_reason)
			VALUES ('task-recovery-upgrade', 'task-recovery-upgrade-key', 'recovery-upgrade', 1,
				'draft-recovery-upgrade', 1, 'confirmation-recovery-upgrade', 1,
				'needs_confirmation', 'legacy_confirmation_required');
			DELETE FROM schema_migration WHERE version = 12`); err != nil {
			t.Fatal(err)
		}

		store, err = Open(ctx, testURL)
		if err != nil {
			t.Fatalf("apply task recovery compatibility migration: %v", err)
		}
		store.Close()
		assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version = 12`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task
			WHERE id = 'task-recovery-upgrade' AND state = 'needs_confirmation'
			AND hold_reason = 'task_result_uncertain'`, 1)
		phaseAStore, err := account.Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		defer phaseAStore.Close()
		detail, err := phaseAStore.GetTaskDetail(ctx, "task-recovery-upgrade")
		if err != nil || detail.AllowedAction != "verify" {
			t.Fatalf("migrated unknown result did not require manual verification: detail=%+v err=%v", detail, err)
		}
	})

	t.Run("previous migration 008 already applied", 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 _, err := db.Exec(`
			INSERT INTO credential_reference (id, provider, reference_key)
			VALUES ('credential-upgrade', 'os_keyring', 'creatorhub/upgrade');
			INSERT INTO social_account
				(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status, status)
			VALUES ('upgrade', 'credential-upgrade', 'mock', 'upgrade', 'owned', 'authorized', 'paused');
			INSERT INTO gateway (name, endpoint, token)
			VALUES ('upgrade-gateway', 'http://127.0.0.1:8081', 'upgrade-gateway-token');
			INSERT INTO browser_version (version, browser_path) VALUES ('1', '/opt/creatorhub/browsers/1');
			INSERT INTO browser_env (alias, name, gateway_name, browser_version, fingerprint)
			VALUES ('upgrade', 'Upgrade', 'upgrade-gateway', '1', '{"seed":1}');
			INSERT INTO environment_binding (id, account_id, browser_env_alias)
			VALUES ('upgrade', 'upgrade', 'upgrade');
			UPDATE environment_binding SET runtime_cleanup_pending = true,
				runtime_cleanup_binding_version = version WHERE id = 'upgrade';
			DELETE FROM schema_migration WHERE version = 9;
			ALTER TABLE environment_binding ADD CONSTRAINT runtime_cleanup_generation_consistent CHECK (
				(runtime_cleanup_pending AND runtime_cleanup_binding_version IS NOT NULL AND runtime_cleanup_binding_version > 0)
				OR
				(NOT runtime_cleanup_pending AND runtime_cleanup_binding_version IS NULL
					AND runtime_cleanup_instance_id IS NULL AND runtime_cleanup_runtime_id IS NULL)
			)`); err != nil {
			t.Fatal(err)
		}
		if _, err := db.Exec(`UPDATE environment_binding SET runtime_cleanup_pending = false WHERE id = 'upgrade'`); err == nil {
			t.Fatal("old migration 008 constraint unexpectedly allowed a legacy writer to clear pending")
		}

		store, err = Open(ctx, testURL)
		if err != nil {
			t.Fatalf("apply forward compatibility migration: %v", err)
		}
		defer store.Close()
		if _, err := db.Exec(`UPDATE environment_binding SET runtime_cleanup_pending = false WHERE id = 'upgrade'`); err != nil {
			t.Fatalf("migration 009 did not unblock the old writer: %v", err)
		}
		context, err := store.GetEnvironmentContext(ctx, "upgrade")
		if err != nil || context.RuntimeCleanupPending || context.RuntimeCleanupBindingVersion != 0 || context.RuntimeCleanupRuntimeID != "" {
			t.Fatalf("new runner reused stale cleanup generation after old writer clear: %#v err=%v", context, err)
		}
		if _, err := db.Exec(`UPDATE environment_binding SET runtime_cleanup_pending = true WHERE id = 'upgrade'`); err != nil {
			t.Fatalf("migration 009 did not allow the old writer to set pending: %v", err)
		}
		context, err = store.GetEnvironmentContext(ctx, "upgrade")
		if err != nil || !context.RuntimeCleanupPending || context.RuntimeCleanupBindingVersion != 1 || context.RuntimeCleanupRuntimeID != "" {
			t.Fatalf("new runner did not expose legacy pending for manual reconcile: %#v err=%v", context, err)
		}
		assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version IN (9, 10)`, 2)
		assertDatabaseCount(t, db, `SELECT count(*) FROM information_schema.columns WHERE table_schema = current_schema() AND
			((table_name = 'runtime_instance' AND column_name = 'network_id') OR
			 (table_name = 'environment_binding' AND column_name = 'runtime_cleanup_network_id'))`, 2)
		assertDatabaseCount(t, db, `SELECT count(*) FROM pg_constraint WHERE conname = 'runtime_cleanup_generation_consistent'`, 0)
	})

	t.Run("v1 and v2 data", func(t *testing.T) {
		ctx := context.Background()
		testURL := isolatedDatabaseURL(t, databaseURL)
		phaseAStore, err := account.Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		phaseAStore.Close()

		db, err := sql.Open("pgx", testURL)
		if err != nil {
			t.Fatal(err)
		}
		defer db.Close()
		if _, err := db.Exec(migration002); err != nil {
			t.Fatal(err)
		}
		if _, err := db.Exec(`
			INSERT INTO schema_migration (version) VALUES (2);
			ALTER TABLE browser_image DROP CONSTRAINT browser_image_image_ref_check;
			INSERT INTO credential_reference (id, provider, reference_key) VALUES
				('credential-mapped', 'os_keyring', 'creatorhub/mapped'),
				('credential-unbound', 'secret_manager', 'creatorhub/unbound');
			INSERT INTO social_account (id, credential_reference_id, profile_id, status) VALUES
				('mapped', 'credential-mapped', 'legacy-profile-mapped', 'active'),
				('unbound', 'credential-unbound', 'legacy-profile-unbound', 'active');
			INSERT INTO content_draft (id, account_id, version, content) VALUES ('legacy-draft', 'mapped', 1, 'legacy');
			INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
			VALUES ('legacy-confirmation', 'mapped', 1, 'legacy-draft', 1, 1);
			INSERT INTO operation_task
				(id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version, state)
			VALUES
				('legacy-task', 'legacy-task-key', 'mapped', 1, 'legacy-draft', 1, 'legacy-confirmation', 1, 'queued'),
				('legacy-unknown-task', 'legacy-unknown-task-key', 'mapped', 1, 'legacy-draft', 1, 'legacy-confirmation', 1, 'needs_confirmation');
			INSERT INTO gateway (name, endpoint, token) VALUES ('legacy-gateway', 'http://127.0.0.1:8081', 'legacy-gateway-token');
			INSERT INTO browser_image (version, image_ref) VALUES ('1', '/opt/creatorhub/browsers/1');
			INSERT INTO browser_env (alias, name, gateway_name, image_version, fingerprint) VALUES
				('mapped', 'Mapped', 'legacy-gateway', '1', '{"seed":1,"proxy_server":"http://legacy:secret@proxy.example:8080","disable_non_proxied_udp":true}'),
				('orphan-env', 'Orphan', 'legacy-gateway', '1', '{"seed":2}');
			INSERT INTO runtime_instance (id, account_id, runtime_id, lease_until) VALUES
				('instance-mapped', 'mapped', 'runtime-mapped', now() + interval '1 hour'),
				('instance-unbound', 'unbound', 'runtime-unbound', now() + interval '1 hour');
			INSERT INTO audit_event (event_type, account_id) VALUES ('legacy_event', 'mapped')`); err != nil {
			t.Fatal(err)
		}

		store, err := Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		store.Close()

		assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE platform = 'mock' AND platform_account_key = id AND status = 'paused'`, 2)
		assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE profile_id LIKE 'legacy-profile-%'`, 2)
		assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE account_id = 'mapped' AND browser_env_alias = 'mapped' AND network_exit_id IS NULL`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE account_id = 'mapped' AND NOT runtime_cleanup_pending`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE account_id = 'unbound'`, 0)
		assertDatabaseCount(t, db, `SELECT count(*) FROM runtime_instance WHERE id = 'instance-mapped' AND binding_id = 'mapped'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM runtime_instance WHERE id = 'instance-mapped' AND binding_version = 1`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM runtime_instance WHERE id = 'instance-unbound' AND binding_id IS NULL`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM audit_event WHERE event_type = 'legacy_event'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'legacy-task' AND state = 'policy_hold'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'legacy-unknown-task' AND state = 'needs_confirmation' AND hold_reason = 'task_result_uncertain'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM browser_env WHERE alias = 'mapped' AND NOT (fingerprint ?| ARRAY['proxy_server', 'disable_non_proxied_udp'])`, 1)
		if _, err := db.Exec(`UPDATE environment_binding SET runtime_cleanup_pending = true WHERE id = 'mapped'`); err != nil {
			t.Fatalf("migration 008 blocked an old writer setting cleanup pending: %v", err)
		}
		assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE id = 'mapped' AND runtime_cleanup_pending AND runtime_cleanup_binding_version IS NULL AND runtime_cleanup_runtime_id IS NULL`, 1)
		if _, err := db.Exec(`UPDATE environment_binding SET runtime_cleanup_pending = false WHERE id = 'mapped'`); err != nil {
			t.Fatalf("migration 008 blocked an old writer clearing cleanup pending: %v", err)
		}
		store, err = Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		legacy, err := store.GetEnvironmentContext(ctx, "mapped")
		store.Close()
		if err != nil || legacy.Exit.ID != "" || legacy.Fingerprint.ProxyServer != "" || legacy.Fingerprint.DisableNonProxiedUDP {
			t.Fatalf("legacy NULL binding must remain visible without persisted proxy credentials: %#v err=%v", legacy, err)
		}

		if _, err := db.Exec(`
			INSERT INTO credential_reference (id, provider, reference_key) VALUES ('credential-duplicate', 'os_keyring', 'creatorhub/duplicate');
			INSERT INTO social_account
				(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status)
			VALUES ('duplicate', 'credential-duplicate', 'mock', 'mapped', 'owned', 'authorized')`); err == nil {
			t.Fatal("duplicate platform account must fail")
		}

		if _, err := db.Exec(`
			INSERT INTO network_exit (id, protocol, host, port, health_status) VALUES ('shared-exit', 'socks5', '127.0.0.1', 1080, 'healthy');
			UPDATE environment_binding SET network_exit_id = 'shared-exit' WHERE id = 'mapped';
			INSERT INTO environment_binding (id, account_id, browser_env_alias, network_exit_id)
			VALUES ('unbound-binding', 'unbound', 'orphan-env', 'shared-exit')`); err != nil {
			t.Fatal(err)
		}
		assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE network_exit_id = 'shared-exit'`, 2)

		if _, err := db.Exec(`
			INSERT INTO browser_env (alias, name, gateway_name, image_version, fingerprint)
			VALUES ('extra-env', 'Extra', 'legacy-gateway', '1', '{"seed":3}');
			INSERT INTO environment_binding (id, account_id, browser_env_alias)
			VALUES ('duplicate-account-binding', 'mapped', 'extra-env')`); err == nil {
			t.Fatal("one account must not bind two environments")
		}
		if _, err := db.Exec(`
			INSERT INTO credential_reference (id, provider, reference_key) VALUES ('credential-extra', 'os_keyring', 'creatorhub/extra');
			INSERT INTO social_account
				(id, credential_reference_id, platform, platform_account_key, authorization_kind, authorization_status)
			VALUES ('extra-account', 'credential-extra', 'mock', 'extra-account', 'owned', 'authorized');
			INSERT INTO environment_binding (id, account_id, browser_env_alias)
			VALUES ('duplicate-env-binding', 'extra-account', 'mapped')`); err == nil {
			t.Fatal("one environment must not bind two accounts")
		}
		if _, err := db.Exec(`
			INSERT INTO runtime_instance (id, account_id, binding_id, runtime_id, lease_until)
			VALUES ('instance-conflict', 'mapped', 'mapped', 'runtime-new', now() + interval '1 hour')`); err == nil {
			t.Fatal("one binding must not have two active runtime instances")
		}
		if _, err := db.Exec(`
			INSERT INTO browser_version (version, browser_path) VALUES ('2', '/opt/creatorhub/browsers/2');
			UPDATE social_account SET status = 'active' WHERE id = 'mapped';
			INSERT INTO content_draft (id, account_id, version, content) VALUES ('upgrade-draft', 'mapped', 1, 'test');
			INSERT INTO confirmation (id, account_id, account_version, draft_id, draft_version, version)
			VALUES ('upgrade-confirmation', 'mapped', 1, 'upgrade-draft', 1, 1);
			INSERT INTO operation_task
				(id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version)
			VALUES ('upgrade-task', 'upgrade-task-key', 'mapped', 1, 'upgrade-draft', 1, 'upgrade-confirmation', 1);
			INSERT INTO operation_task
				(id, idempotency_key, account_id, account_version, draft_id, draft_version, confirmation_id, confirmation_version, state, lease_owner, lease_until)
			VALUES ('upgrade-executing', 'upgrade-executing-key', 'mapped', 1, 'upgrade-draft', 1, 'upgrade-confirmation', 1,
				'executing', 'worker-old', now() + interval '1 minute');
			INSERT INTO execution_attempt (id, task_id) VALUES ('upgrade-attempt', 'upgrade-executing');
			UPDATE operation_task SET current_attempt_id = 'upgrade-attempt' WHERE id = 'upgrade-executing'`); err != nil {
			t.Fatal(err)
		}
		store, err = Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		var notifications []taskstate.Transition
		store.SetTaskNotifier(func(transition taskstate.Transition) { notifications = append(notifications, transition) })
		if err := store.UpgradeEnv(ctx, "mapped", "2"); err != nil {
			t.Fatal(err)
		}
		store.Close()
		assertDatabaseCount(t, db, `SELECT count(*) FROM browser_env WHERE alias = 'mapped' AND version = 2 AND browser_version = '2'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM environment_binding WHERE id = 'mapped' AND version = 2`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM social_account WHERE id = 'mapped' AND version = 2 AND status = 'paused'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-task' AND state = 'policy_hold' AND hold_reason = 'binding_version_changed'`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-executing' AND state = 'needs_confirmation' AND hold_reason = 'task_result_uncertain' AND lease_owner IS NULL`, 1)
		assertDatabaseCount(t, db, `SELECT count(*) FROM execution_attempt WHERE id = 'upgrade-attempt' AND outcome = 'uncertain' AND finished_at IS NOT NULL`, 1)
		for _, want := range []taskstate.Transition{
			{State: "policy_hold", ReasonCode: "binding_version_changed", AccountID: "mapped", TaskID: "upgrade-task"},
			{State: "needs_confirmation", ReasonCode: "task_result_uncertain", AccountID: "mapped", TaskID: "upgrade-executing"},
		} {
			if !slices.Contains(notifications, want) {
				t.Fatalf("missing upgrade transition %+v in %+v", want, notifications)
			}
		}
		phaseAStore, err = account.Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		defer phaseAStore.Close()
		detail, err := phaseAStore.GetTaskDetail(ctx, "upgrade-executing")
		if err != nil || detail.AllowedAction != "verify" {
			t.Fatalf("upgraded executing task did not require manual verification: detail=%+v err=%v", detail, err)
		}
		if err := phaseAStore.VerifyTask(ctx, "upgrade-executing", "failed"); err != nil {
			t.Fatalf("verify upgraded executing task: %v", err)
		}
		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE id = 'upgrade-executing' AND verification_result = 'failed' AND verified_by = 'local-user'`, 1)
		if err := phaseAStore.VerifyTask(ctx, "upgrade-executing", "succeeded"); !errors.Is(err, account.ErrConflict) {
			t.Fatalf("repeated verification changed the recorded conclusion: %v", err)
		}
		assertDatabaseCount(t, db, `SELECT count(*) FROM operation_task WHERE confirmation_id = 'upgrade-confirmation'`, 2)

		store, err = Open(ctx, testURL)
		if err != nil {
			t.Fatal(err)
		}
		store.Close()
		assertDatabaseCount(t, db, `SELECT count(*) FROM schema_migration WHERE version IN (3, 4, 5, 6, 7, 8, 9)`, 7)
	})
}

func openFullyMigratedHub(t *testing.T, ctx context.Context, databaseURL string) *Store {
	t.Helper()
	phaseAStore, err := account.Open(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	phaseAStore.Close()
	store, err := Open(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	return store
}

func isolatedDatabaseURL(t *testing.T, databaseURL string) string {
	t.Helper()
	admin, err := sql.Open("pgx", databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { admin.Close() })
	schema := fmt.Sprintf("creatorhub_hh804_%d", time.Now().UnixNano())
	if _, err := admin.Exec("CREATE SCHEMA " + schema); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() {
		if _, err := admin.Exec("DROP SCHEMA " + schema + " CASCADE"); err != nil {
			t.Errorf("drop test schema: %v", err)
		}
	})
	parsed, err := url.Parse(databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	query := parsed.Query()
	query.Set("search_path", schema)
	parsed.RawQuery = query.Encode()
	return parsed.String()
}

func openLegacyAccountCreationSchema(t *testing.T, ctx context.Context, databaseURL string) *sql.DB {
	t.Helper()
	phaseAStore, err := account.Open(ctx, databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	phaseAStore.Close()
	db, err := sql.Open("pgx", databaseURL)
	if err != nil {
		t.Fatal(err)
	}
	for _, migration := range []struct {
		version int
		sql     string
	}{{2, migration002}, {3, migration003}, {4, migration004}, {5, migration005}, {6, migration006}, {7, migration007},
		{8, migration008}, {9, migration009}, {10, migration010}, {11, migration011}, {12, migration012}} {
		if _, err := db.Exec(migration.sql); err != nil {
			t.Fatal(err)
		}
		if _, err := db.Exec(`INSERT INTO schema_migration (version) VALUES ($1)`, migration.version); err != nil {
			t.Fatal(err)
		}
	}
	if _, err := db.Exec(`
		ALTER TABLE social_account
			ADD COLUMN name text,
			ADD COLUMN tags text[],
			ADD COLUMN cookies text;
		UPDATE social_account SET name = platform_account_key, tags = ARRAY[]::text[], cookies = '';
		ALTER TABLE social_account
			ALTER COLUMN credential_reference_id DROP NOT NULL,
			ALTER COLUMN name SET NOT NULL,
			ALTER COLUMN name SET DEFAULT '未命名账号',
			ALTER COLUMN tags SET NOT NULL,
			ALTER COLUMN tags SET DEFAULT ARRAY[]::text[],
			ALTER COLUMN cookies SET NOT NULL,
			ALTER COLUMN cookies SET DEFAULT '',
			ADD CONSTRAINT social_account_name_check CHECK (name = btrim(name) AND length(name) BETWEEN 1 AND 128),
			ADD CONSTRAINT social_account_tags_check CHECK (cardinality(tags) <= 20),
			ADD CONSTRAINT social_account_cookies_length_check CHECK (length(cookies) <= 8192);
		INSERT INTO schema_migration (version) VALUES (13)`); err != nil {
		t.Fatal(err)
	}
	return db
}

func assertDatabaseCount(t *testing.T, db *sql.DB, query string, want int, args ...any) {
	t.Helper()
	var got int
	if err := db.QueryRow(query, args...).Scan(&got); err != nil || got != want {
		t.Fatalf("count mismatch: got=%d want=%d err=%v query=%s", got, want, err, query)
	}
}