package environment import ( "context" "crypto/rand" "database/sql" _ "embed" "encoding/hex" "encoding/json" "errors" "fmt" "net/url" "regexp" "sort" "strings" "time" "unicode/utf8" "github.com/jackc/pgx/v5/pgconn" _ "github.com/jackc/pgx/v5/stdlib" ) //go:embed migrations/002_hub.sql var migration002 string //go:embed migrations/003_unified_accounts.sql var migration003 string //go:embed migrations/004_environment_actions.sql var migration004 string //go:embed migrations/005_sanitize_legacy_proxy.sql var migration005 string //go:embed migrations/006_runtime_cleanup.sql var migration006 string //go:embed migrations/007_runtime_binding_version.sql var migration007 string //go:embed migrations/008_runtime_cleanup_generation.sql var migration008 string //go:embed migrations/009_runtime_cleanup_compatibility.sql var migration009 string //go:embed migrations/010_runtime_network_generation.sql var migration010 string //go:embed migrations/011_task_recovery.sql var migration011 string //go:embed migrations/012_task_recovery_compatibility.sql var migration012 string //go:embed migrations/013_account_creation.sql var migration013 string //go:embed migrations/014_account_creation_compatibility.sql var migration014 string //go:embed migrations/015_gateway_rename_cascade.sql var migration015 string //go:embed migrations/016_network_exit_plain_credentials.sql var migration016 string //go:embed migrations/017_native_browser_versions.sql var migration017 string //go:embed migrations/033_unique_fingerprint_seed.sql var migration033 string //go:embed migrations/034_gateway_browser_default.sql var migration034 string //go:embed migrations/037_browser_env_image_version_repair.sql var migration037 string //go:embed migrations/001_phase_a.sql var migration001 string //go:embed migrations/038_creator_registration_backfill.sql var migration038 string //go:embed migrations/043_schema_consolidation.sql var migration043 string //go:embed migrations/044_numeric_primary_keys.sql var migration044 string //go:embed migrations/1017_creator.sql var migration1017 string //go:embed migrations/1018_collection_checkpoints.sql var migration1018 string //go:embed migrations/1019_metric_plans.sql var migration1019 string //go:embed migrations/1020_competitor_sync_leases.sql var migration1020 string //go:embed migrations/1021_work_sources.sql var migration1021 string //go:embed migrations/1022_collection_lease_tokens.sql var migration1022 string //go:embed migrations/1023_password_references.sql var migration1023 string //go:embed migrations/1024_event_processing_times.sql var migration1024 string //go:embed migrations/1025_competitor_sync_tokens.sql var migration1025 string //go:embed migrations/1026_event_message_text.sql var migration1026 string //go:embed migrations/1027_creator_listener_state.sql var migration1027 string //go:embed migrations/1028_creator_event_message_type.sql var migration1028 string //go:embed migrations/1029_creator_source_sync_lease.sql var migration1029 string //go:embed migrations/1030_creator_event_gateway_time.sql var migration1030 string //go:embed migrations/1031_raw_payloads.sql var migration1031 string //go:embed migrations/1032_douyin_release_remediation.sql var migration1032 string //go:embed migrations/1033_competitor_tags.sql var migration1033 string //go:embed migrations/1034_competitor_tags_repair.sql var migration1034 string //go:embed migrations/1035_account_deletion.sql var migration1035 string //go:embed migrations/1036_competitor_unique_id.sql var migration1036 string //go:embed migrations/1037_competitor_share_jobs.sql var migration1037 string //go:embed migrations/1038_douyin_only_competitor_profile.sql var migration1038 string //go:embed migrations/1039_work_collect_count.sql var migration1039 string //go:embed migrations/1040_work_play_count.sql var migration1040 string //go:embed migrations/1041_creator_account_metric.sql var migration1041 string //go:embed migrations/1042_work_cover_cache.sql var migration1042 string //go:embed migrations/1043_gateway_health.sql var migration1043 string //go:embed migrations/1044_creator_account_metric_friend_count.sql var migration1044 string //go:embed migrations/1045_drop_authorization_status.sql var migration1045 string //go:embed migrations/1046_competitor_profile_signature.sql var migration1046 string var ( ErrConflict = errors.New("resource conflicts with existing state") ErrInvalid = errors.New("invalid environment input") ErrNotFound = errors.New("resource not found") ErrReconcileRequired = errors.New("runtime cleanup generation is unknown; manual reconciliation required") ) func ValidNetworkExitID(id string) bool { return exitIDPattern.MatchString(id) } var ( aliasPattern = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,31}$`) gatewayNamePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`) tokenPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{15,127}$`) ) type Store struct { db *sql.DB lockAdmission chan struct{} } // Gateway 是平台注册的 native browser gateway 节点;Token 由平台生成,明文存储供页面复制(开发阶段约定)。 // 健康字段由控制面后台定时探活写入,只保留最近一次结果,不保留历史。 type Gateway struct { Name string `json:"name"` Endpoint string `json:"endpoint"` Token string `json:"token"` HealthStatus string `json:"health_status"` LastCheckReason string `json:"last_check_reason"` LastCheckedAt *time.Time `json:"last_checked_at"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` } // Env 是一个浏览器环境;浏览器安装和默认运行时由 gateway 宿主机配置。 type Env struct { Alias string `json:"alias"` Name string `json:"name"` Gateway string `json:"gateway"` BrowserVersion string `json:"-"` // legacy database test snapshot; runtime selection is gateway-owned Fingerprint Fingerprint `json:"fingerprint"` CreatedAt time.Time `json:"created_at"` } func Open(ctx context.Context, databaseURL string) (*Store, error) { db, err := sql.Open("pgx", databaseURL) if err != nil { return nil, errors.New("open environment database") } db.SetMaxOpenConns(10) db.SetMaxIdleConns(2) db.SetConnMaxIdleTime(5 * time.Minute) if err := db.PingContext(ctx); err != nil { db.Close() return nil, errors.New("connect to environment database") } // LockResources keeps one connection until the lifecycle operation finishes. // Admit at most half the pool so those operations can still open nested DB calls. store := &Store{db: db, lockAdmission: make(chan struct{}, 5)} if err := store.migrate(ctx); err != nil { db.Close() return nil, err } return store, nil } func (s *Store) Close() error { return s.db.Close() } func (s *Store) Ping(ctx context.Context) error { return s.db.PingContext(ctx) } // LockResources serializes lifecycle state across control-plane replicas. The // transaction carries no data changes; rolling it back only releases the locks. func (s *Store) LockResources(ctx context.Context, aliases, exitIDs []string, _ ...[]string) (func(), error) { for _, alias := range aliases { if !aliasPattern.MatchString(alias) { return nil, ErrInvalid } } for _, id := range exitIDs { if !exitIDPattern.MatchString(id) { return nil, ErrInvalid } } if len(aliases)+len(exitIDs) == 0 { return func() {}, nil } select { case s.lockAdmission <- struct{}{}: case <-ctx.Done(): return nil, ctx.Err() } tx, err := s.db.BeginTx(ctx, nil) if err != nil { <-s.lockAdmission return nil, errors.New("begin resource lock") } resources := []struct { namespace int keys []string }{{1542738013, aliases}, {1542738015, exitIDs}} for _, resource := range resources { keys := append([]string(nil), resource.keys...) sort.Strings(keys) for _, key := range keys { if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock($1, hashtext($2))`, resource.namespace, key); err != nil { _ = tx.Rollback() <-s.lockAdmission return nil, errors.New("lock resource") } } } return func() { _ = tx.Rollback() <-s.lockAdmission }, nil } func (s *Store) migrate(ctx context.Context) error { tx, err := s.db.BeginTx(ctx, nil) if err != nil { return errors.New("begin environment schema migration") } defer tx.Rollback() if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(1542738012)`); err != nil { return errors.New("lock environment schema migration") } if _, err := tx.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS schema_migration (version integer PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`); err != nil { return errors.New("create schema migration table") } for _, migration := range []struct { version int sql string }{{1, migration001}, {2, migration002}, {3, migration003}, {4, migration004}, {5, migration005}, {6, migration006}, {7, migration007}, {8, migration008}, {9, migration009}, {10, migration010}, {11, migration011}, {12, migration012}, {13, migration013}, {14, migration014}, {15, migration015}, {16, migration016}, {17, migration017}, {33, migration033}, {34, migration034}, {37, migration037}, {38, migration038}, {1017, migration1017}, {1018, migration1018}, {1019, migration1019}, {1020, migration1020}, {1021, migration1021}, {1022, migration1022}, {1023, migration1023}, {1024, migration1024}, {1025, migration1025}, {1026, migration1026}, {1027, migration1027}, {1028, migration1028}, {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}} { 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") } if applied { continue } if _, err := tx.ExecContext(ctx, migration.sql); err != nil { return fmt.Errorf("apply environment schema migration %d: %w", migration.version, err) } if _, err := tx.ExecContext(ctx, `INSERT INTO schema_migration (version) VALUES ($1)`, migration.version); err != nil { return fmt.Errorf("record environment schema migration %d", migration.version) } } if err := tx.Commit(); err != nil { return errors.New("commit environment schema migration") } return nil } func scanGateway(row rowScanner) (Gateway, error) { var gateway Gateway var checked sql.NullTime if err := row.Scan(&gateway.Name, &gateway.Endpoint, &gateway.Token, &gateway.HealthStatus, &gateway.LastCheckReason, &checked, &gateway.CreatedAt, &gateway.UpdatedAt); err != nil { return Gateway{}, rowError(err) } if checked.Valid { gateway.LastCheckedAt = &checked.Time } return gateway, nil } func (s *Store) CreateGateway(ctx context.Context, name, endpoint, token string) (Gateway, error) { name, endpoint, token = strings.TrimSpace(name), strings.TrimSpace(endpoint), strings.TrimSpace(token) if !gatewayNamePattern.MatchString(name) || !validHTTPURL(endpoint) { return Gateway{}, ErrInvalid } // 未显式提供 token 时由平台生成;提供则与网关进程的 GATEWAY_TOKEN 保持一致。 if token != "" { if !tokenPattern.MatchString(token) { return Gateway{}, ErrInvalid } } else { token = newToken() } result, err := scanGateway(s.db.QueryRowContext(ctx, ` INSERT INTO gateway (name, endpoint, token) VALUES ($1, $2, $3) RETURNING name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at`, name, endpoint, token)) if err != nil { return Gateway{}, err } return result, nil } // UpdateGateway 修改网关名称、Endpoint 和令牌。名称变更由数据库外键 ON UPDATE CASCADE // 原子同步 browser_env 引用;空令牌表示保留当前令牌,避免只改地址时意外轮换凭证。 func (s *Store) UpdateGateway(ctx context.Context, currentName, name, endpoint, token string) (Gateway, error) { currentName, name = strings.TrimSpace(currentName), strings.TrimSpace(name) endpoint, token = strings.TrimSpace(endpoint), strings.TrimSpace(token) if !gatewayNamePattern.MatchString(currentName) || !gatewayNamePattern.MatchString(name) || !validHTTPURL(endpoint) || (token != "" && !tokenPattern.MatchString(token)) { return Gateway{}, ErrInvalid } return scanGateway(s.db.QueryRowContext(ctx, ` UPDATE gateway SET name = $1, endpoint = $2, token = CASE WHEN $3 = '' THEN token ELSE $3 END, updated_at = now() WHERE name = $4 RETURNING name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at`, name, endpoint, token, currentName)) } func (s *Store) ListGateways(ctx context.Context) ([]Gateway, error) { rows, err := s.db.QueryContext(ctx, ` SELECT name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at FROM gateway ORDER BY created_at, name`) if err != nil { return nil, errors.New("read gateways") } defer rows.Close() gateways := []Gateway{} for rows.Next() { gateway, err := scanGateway(rows) if err != nil { return nil, err } gateways = append(gateways, gateway) } return gateways, rows.Err() } func (s *Store) GetGateway(ctx context.Context, name string) (Gateway, error) { if !gatewayNamePattern.MatchString(name) { return Gateway{}, ErrInvalid } return scanGateway(s.db.QueryRowContext(ctx, ` SELECT name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at FROM gateway WHERE name = $1`, name)) } // RecordGatewayCheck 覆盖写入最近一次探活结果;不保留历史。status 只允许 healthy/unhealthy。 func (s *Store) RecordGatewayCheck(ctx context.Context, name, status, reason string) (Gateway, error) { if !gatewayNamePattern.MatchString(name) || (status != "healthy" && status != "unhealthy") { return Gateway{}, ErrInvalid } return scanGateway(s.db.QueryRowContext(ctx, ` UPDATE gateway SET health_status = $2, last_check_reason = $3, last_checked_at = now(), updated_at = now() WHERE name = $1 RETURNING name, endpoint, token, health_status, last_check_reason, last_checked_at, created_at, updated_at`, name, status, reason)) } func (s *Store) DeleteGateway(ctx context.Context, name string) error { if !gatewayNamePattern.MatchString(name) { return ErrInvalid } var deleted string if err := s.db.QueryRowContext(ctx, `DELETE FROM gateway WHERE name = $1 RETURNING name`, name).Scan(&deleted); err != nil { return rowError(err) } return nil } func (s *Store) ListEnvs(ctx context.Context) ([]Env, error) { rows, err := s.db.QueryContext(ctx, ` SELECT environment.alias, environment.name, gateway.name, environment.fingerprint, environment.created_at FROM browser_env environment JOIN gateway ON gateway.id = environment.gateway_id ORDER BY environment.created_at, environment.alias`) if err != nil { return nil, errors.New("read browser envs") } defer rows.Close() envs := []Env{} for rows.Next() { env, err := scanEnv(rows) if err != nil { return nil, err } envs = append(envs, env) } return envs, rows.Err() } func (s *Store) GetEnv(ctx context.Context, alias string) (Env, error) { if !aliasPattern.MatchString(alias) { return Env{}, ErrInvalid } rows, err := s.db.QueryContext(ctx, ` SELECT environment.alias, environment.name, gateway.name, environment.fingerprint, environment.created_at FROM browser_env environment JOIN gateway ON gateway.id = environment.gateway_id WHERE environment.alias = $1`, alias) if err != nil { return Env{}, errors.New("read browser env") } defer rows.Close() if !rows.Next() { return Env{}, ErrNotFound } env, err := scanEnv(rows) if err != nil { return Env{}, err } return env, rows.Err() } func (s *Store) DeleteEnv(ctx context.Context, alias string) error { if !aliasPattern.MatchString(alias) { return ErrInvalid } var deleted string if err := s.db.QueryRowContext(ctx, `DELETE FROM browser_env WHERE alias = $1 RETURNING alias`, alias).Scan(&deleted); err != nil { return rowError(err) } return nil } // DeleteAccountEnvironment removes the account binding and its private browser environment. // Shared gateways, images, and network exits are intentionally preserved. func (s *Store) DeleteAccountEnvironment(ctx context.Context, accountID string) error { if strings.TrimSpace(accountID) == "" { return ErrInvalid } tx, err := s.db.BeginTx(ctx, nil) if err != nil { return errors.New("begin account environment deletion") } defer tx.Rollback() var alias string if err := tx.QueryRowContext(ctx, ` SELECT environment.alias FROM browser_env environment JOIN social_account account ON account.id = environment.account_id WHERE account.account_id = $1 FOR UPDATE OF environment`, accountID).Scan(&alias); errors.Is(err, sql.ErrNoRows) { return nil } else if err != nil { return errors.New("read account environment binding") } if _, err := tx.ExecContext(ctx, `DELETE FROM browser_env WHERE alias = $1`, alias); err != nil { return errors.New("delete account browser environment") } return commitHub(tx) } func scanEnv(rows *sql.Rows) (Env, error) { var env Env var encoded []byte if err := rows.Scan(&env.Alias, &env.Name, &env.Gateway, &encoded, &env.CreatedAt); err != nil { return Env{}, errors.New("decode browser env") } if len(encoded) > 0 { if err := json.Unmarshal(encoded, &env.Fingerprint); err != nil { return Env{}, errors.New("decode env fingerprint") } } env.Fingerprint.ProxyServer = "" env.Fingerprint.DisableNonProxiedUDP = false return env, nil } func validDisplayName(name string) bool { if name == "" || utf8.RuneCountInString(name) > 64 { return false } for _, r := range name { if r < 0x20 || r == 0x7f { return false } } return true } func validHTTPURL(raw string) bool { parsed, err := url.Parse(raw) return err == nil && parsed.Host != "" && (parsed.Scheme == "http" || parsed.Scheme == "https") } func newToken() string { var value [24]byte _, _ = rand.Read(value[:]) return hex.EncodeToString(value[:]) } func commitHub(tx *sql.Tx) error { if err := tx.Commit(); err != nil { return errors.New("commit environment transaction") } return nil } func rowError(err error) error { if errors.Is(err, sql.ErrNoRows) { return ErrNotFound } return publicDatabaseError(err) } func publicDatabaseError(err error) error { if err == nil { return nil } var postgresError *pgconn.PgError if errors.As(err, &postgresError) && (postgresError.Code == "23505" || postgresError.Code == "23503" || postgresError.Code == "23514") { return ErrConflict } return errors.New("environment persistence operation failed") }