gpt-plus-gpt/internal/task/runner.go

package task

import (
	"context"
	"crypto/rand"
	"errors"
	"fmt"
	"log"
	"math/big"
	"net/http"
	"strings"
	"sync/atomic"
	"time"

	"gpt-plus/config"
	"gpt-plus/internal/db"
	"gpt-plus/pkg/auth"
	"gpt-plus/pkg/captcha"
	"gpt-plus/pkg/chatgpt"
	"gpt-plus/pkg/httpclient"
	"gpt-plus/pkg/provider/card"
	"gpt-plus/pkg/provider/email"
	"gpt-plus/pkg/proxy"
	"gpt-plus/pkg/storage"
	"gpt-plus/pkg/stripe"

	"gorm.io/gorm"
)

type TaskRunner struct {
	taskID   string
	gormDB   *gorm.DB
	ctx      context.Context
	cancel   context.CancelFunc
	stopping atomic.Bool
	running  atomic.Bool
}

func NewTaskRunner(taskID string, d *gorm.DB) (*TaskRunner, error) {
	ctx, cancel := context.WithCancel(context.Background())
	return &TaskRunner{
		taskID: taskID,
		gormDB: d,
		ctx:    ctx,
		cancel: cancel,
	}, nil
}

func (r *TaskRunner) IsRunning() bool {
	return r.running.Load()
}

func (r *TaskRunner) GracefulStop() {
	r.stopping.Store(true)
	r.gormDB.Model(&db.Task{}).Where("id = ?", r.taskID).Update("status", StatusStopping)
	log.Printf("[task-runner] %s: graceful stop requested", r.taskID)
}

func (r *TaskRunner) ForceStop() {
	r.stopping.Store(true)
	r.cancel()
	log.Printf("[task-runner] %s: force stop (context cancelled)", r.taskID)
}

func (r *TaskRunner) Run() {
	r.running.Store(true)
	defer r.running.Store(false)

	var t db.Task
	if err := r.gormDB.First(&t, "id = ?", r.taskID).Error; err != nil {
		log.Printf("[task-runner] %s: load task failed: %v", r.taskID, err)
		return
	}

	// Load config from DB
	cfg, err := config.LoadFromDB(r.gormDB)
	if err != nil {
		log.Printf("[task-runner] %s: load config failed: %v", r.taskID, err)
		r.failTask(&t, "配置加载失败: "+err.Error())
		return
	}

	// Override task-specific settings based on task type
	switch t.Type {
	case TaskTypePlus:
		cfg.Team.Enabled = false
	case TaskTypeTeam:
		cfg.Team.Enabled = true
	case TaskTypeBoth:
		cfg.Team.Enabled = true
	}

	// Mark task as running
	now := time.Now()
	t.Status = StatusRunning
	t.StartedAt = &now
	r.gormDB.Save(&t)

	// Wire locale
	if cfg.Account.Locale != "" {
		lang := cfg.Account.Locale
		acceptLang := lang + ",en;q=0.9"
		if lang == "en-US" {
			acceptLang = "en-US,en;q=0.9"
		}
		auth.SetLocale(lang, acceptLang)
		chatgpt.SetDefaultLanguage(lang)
	}

	// Initialize components
	proxyHasSession := strings.Contains(cfg.Proxy.URL, "{SESSION}")
	b2proxyEnabled := cfg.Proxy.B2Proxy.Enabled
	var cardCountry string
	if b2proxyEnabled {
		cardCountry = cfg.Card.ExpectedCountry()
	}

	var client *httpclient.Client
	if !b2proxyEnabled && !proxyHasSession {
		client, err = httpclient.NewClient(cfg.Proxy.URL)
		if err != nil {
			r.failTask(&t, "创建 HTTP 客户端失败: "+err.Error())
			return
		}
	}

	// Email provider
	emailProv := email.NewMailGateway(
		cfg.Email.MailGateway.BaseURL,
		cfg.Email.MailGateway.APIKey,
		cfg.Email.MailGateway.Provider,
	)

	// Card provider — always use DB mode
	cardProv := card.NewDBCardProvider(card.DBCardProviderConfig{
		DB:                r.gormDB,
		DefaultName:       cfg.Card.API.DefaultName,
		DefaultCountry:    cfg.Card.API.DefaultCountry,
		DefaultCurrency:   cfg.Card.API.DefaultCurrency,
		DefaultAddress:    cfg.Card.API.DefaultAddress,
		DefaultCity:       cfg.Card.API.DefaultCity,
		DefaultState:      cfg.Card.API.DefaultState,
		DefaultPostalCode: cfg.Card.API.DefaultPostalCode,
		APIBaseURL:        cfg.Card.API.BaseURL,
	})

	// Captcha solver
	var solver *captcha.Solver
	if cfg.Captcha.APIKey != "" {
		captchaProxy := cfg.Captcha.Proxy
		if captchaProxy == "" {
			captchaProxy = cfg.Proxy.URL
		}
		solver = captcha.NewSolver(cfg.Captcha.APIKey, captchaProxy)
	}

	// Browser fingerprint pool
	var fpPool *stripe.BrowserFingerprintPool
	if cfg.Stripe.FingerprintDir != "" {
		fpPool, _ = stripe.NewBrowserFingerprintPool(cfg.Stripe.FingerprintDir, "en")
	}

	// Stripe constants
	stripe.SetFallbackConstants(cfg.Stripe.BuildHash, cfg.Stripe.TagVersion)
	stripe.FetchStripeConstants()

	log.Printf("[task-runner] %s: starting batch of %d", r.taskID, t.TotalCount)

	// Main loop
	for i := 1; i <= t.TotalCount; i++ {
		if r.stopping.Load() {
			break
		}
		if r.ctx.Err() != nil {
			break
		}

		log.Printf("[task-runner] %s: iteration %d/%d", r.taskID, i, t.TotalCount)

		// Per-iteration client (B2Proxy / session template)
		iterClient := client
		if b2proxyEnabled {
			r.logStep(i, "", "正在连接 B2Proxy...")
			iterClient = r.getB2ProxyClient(cfg, cardCountry, i, solver)
			if iterClient == nil {
				r.logResult(&t, i, "", LogStatusFailed, "", "B2Proxy 连接失败", 0)
				continue
			}
			r.logStep(i, "", "B2Proxy 连接成功")
		} else if proxyHasSession {
			r.logStep(i, "", "正在连接代理...")
			iterClient = r.getSessionClient(cfg, i, solver)
			if iterClient == nil {
				r.logResult(&t, i, "", LogStatusFailed, "", "代理连接失败", 0)
				continue
			}
			r.logStep(i, "", "代理连接成功")
		}

		start := time.Now()
		result := r.runOnce(r.ctx, i, t.Type, iterClient, emailProv, cardProv, solver, cfg, fpPool)
		dur := int(time.Since(start).Seconds())

		status := LogStatusSuccess
		errMsg := ""
		if result.Error != nil {
			status = LogStatusFailed
			errMsg = result.Error.Error()
		}

		r.logResult(&t, i, result.Email, status, result.Plan, errMsg, dur)

		if i < t.TotalCount {
			time.Sleep(3 * time.Second)
		}
	}

	// Finalize task status
	stopped := time.Now()
	t.StoppedAt = &stopped
	if r.stopping.Load() || r.ctx.Err() != nil {
		t.Status = StatusStopped
	} else {
		t.Status = StatusCompleted
	}
	r.gormDB.Save(&t)
	log.Printf("[task-runner] %s: finished (status=%s, success=%d, fail=%d)", r.taskID, t.Status, t.SuccessCount, t.FailCount)
}

type iterResult struct {
	Email string
	Plan  string
	Error error
}

func (r *TaskRunner) runOnce(ctx context.Context, index int, taskType string, client *httpclient.Client,
	emailProv email.EmailProvider, cardProv card.CardProvider, solver *captcha.Solver,
	cfg *config.Config, fpPool *stripe.BrowserFingerprintPool) iterResult {

	result := iterResult{Plan: "failed"}

	password := generatePassword(cfg.Account.PasswordLength)

	// Step: Register
	r.logStep(index, "", "正在注册账号...")
	regResult, err := auth.Register(ctx, client, emailProv, password)
	if err != nil {
		result.Error = fmt.Errorf("注册失败: %w", err)
		return result
	}
	result.Email = regResult.Email
	r.logStep(index, regResult.Email, "注册成功: "+regResult.Email)

	r.saveEmailRecord(regResult.Email, "in_use", "owner")

	// Step: Login
	sentinel := auth.NewSentinelGeneratorWithDeviceID(client, regResult.DeviceID)
	var loginResult *auth.LoginResult

	if regResult.Tokens != nil {
		loginResult = regResult.Tokens
		r.logStep(index, regResult.Email, "登录成功 (注册时已获取令牌)")
	} else {
		r.logStep(index, regResult.Email, "正在登录...")
		loginResult, err = auth.Login(ctx, client, regResult.Email, password,
			regResult.DeviceID, sentinel, regResult.MailboxID, emailProv)
		if err != nil {
			result.Error = fmt.Errorf("登录失败: %w", err)
			r.updateEmailRecord(regResult.Email, "used_failed")
			return result
		}
		r.logStep(index, regResult.Email, "登录成功")
	}

	session := chatgpt.NewSession(client, loginResult.AccessToken, loginResult.RefreshToken,
		loginResult.IDToken, regResult.DeviceID, loginResult.ChatGPTAccountID, loginResult.ChatGPTUserID)

	// Step: Activate Plus
	r.logStep(index, regResult.Email, "正在开通 Plus...")
	var browserFP *stripe.BrowserFingerprint
	if fpPool != nil {
		browserFP = fpPool.Get()
	}
	statusFn := func(f string, a ...interface{}) {
		msg := fmt.Sprintf(f, a...)
		log.Printf("[task-runner] %s #%d: %s", r.taskID, index, msg)
		r.logStep(index, regResult.Email, msg)
	}
	plusResult, plusErr := chatgpt.ActivatePlus(ctx, session, sentinel, cardProv, cfg.Stripe, regResult.Email, solver, statusFn, browserFP)

	plusSkipped := false
	if plusErr != nil {
		plusSkipped = true
		if errors.Is(plusErr, chatgpt.ErrPlusNotEligible) || errors.Is(plusErr, chatgpt.ErrCaptchaRequired) ||
			errors.Is(plusErr, stripe.ErrCardDeclined) || strings.Contains(plusErr.Error(), "challenge") {
			r.logStep(index, regResult.Email, "Plus 跳过: "+plusErr.Error())
			log.Printf("[task-runner] %s #%d: Plus skipped: %v", r.taskID, index, plusErr)
		} else {
			r.logStep(index, regResult.Email, "Plus 失败: "+plusErr.Error())
			log.Printf("[task-runner] %s #%d: Plus failed: %v", r.taskID, index, plusErr)
		}
	}

	buildAccountResult := func(fp *stripe.Fingerprint) *chatgpt.AccountResult {
		return &chatgpt.AccountResult{
			Email: regResult.Email, Password: password,
			AccessToken: session.AccessToken, RefreshToken: session.RefreshToken,
			IDToken: session.IDToken, ChatGPTAccountID: session.AccountID,
			ChatGPTUserID: session.UserID, GUID: fp.GUID, MUID: fp.MUID, SID: fp.SID,
			Proxy: cfg.Proxy.URL, CreatedAt: time.Now().UTC().Format(time.RFC3339),
		}
	}

	var fingerprint *stripe.Fingerprint
	if !plusSkipped {
		r.logStep(index, regResult.Email, "Plus 开通成功: "+plusResult.PlanType)
		result.Plan = plusResult.PlanType
		fingerprint = &stripe.Fingerprint{GUID: plusResult.GUID, MUID: plusResult.MUID, SID: plusResult.SID}

		session.RefreshSession()
		if session.AccountID == "" {
			if acctInfo, acctErr := chatgpt.CheckAccount(session.Client, session.AccessToken, session.DeviceID); acctErr == nil {
				session.AccountID = acctInfo.AccountID
			}
		}

		r.logStep(index, regResult.Email, "正在获取 Plus OAuth 授权...")
		plusCodexTokens, plusCodexErr := auth.ObtainCodexTokens(ctx, client, regResult.DeviceID, "")
		if plusCodexErr == nil {
			session.AccessToken = plusCodexTokens.AccessToken
			session.RefreshToken = plusCodexTokens.RefreshToken
			if plusCodexTokens.IDToken != "" {
				session.IDToken = plusCodexTokens.IDToken
			}
			if plusCodexTokens.ChatGPTAccountID != "" {
				session.AccountID = plusCodexTokens.ChatGPTAccountID
			}
			r.logStep(index, regResult.Email, "Plus OAuth 授权成功")
		} else {
			r.logStep(index, regResult.Email, "Plus OAuth 授权失败 (非致命)")
		}

		plusAccount := buildAccountResult(fingerprint)
		plusAccount.PlanType = plusResult.PlanType
		plusAccount.StripeSessionID = plusResult.StripeSessionID
		storage.SavePlusAuthFile(cfg.Output.Dir, plusAccount)

		r.saveAccountToDB(plusAccount, "plus", nil, r.taskID)
		r.updateEmailRecord(regResult.Email, "used")
		r.logStep(index, regResult.Email, "Plus 账号已保存")
	} else {
		sc := stripe.FetchStripeConstants()
		if browserFP != nil {
			fingerprint, _ = stripe.GetFingerprint(session.Client, chatgpt.GetDefaultUA(), "chatgpt.com", sc.TagVersion, browserFP)
		} else {
			fingerprint, _ = stripe.GetFingerprintAuto(ctx, session.Client, chatgpt.GetDefaultUA(), "chatgpt.com", sc.TagVersion)
		}
		if fingerprint == nil {
			fingerprint = &stripe.Fingerprint{}
		}
	}

	// Team activation
	var teamResult *chatgpt.TeamResult
	var teamErr error
	if cfg.Team.Enabled {
		r.logStep(index, regResult.Email, "正在开通 Team...")
		teamStatusFn := func(f string, a ...interface{}) {
			msg := fmt.Sprintf(f, a...)
			log.Printf("[task-runner] %s #%d: %s", r.taskID, index, msg)
			r.logStep(index, regResult.Email, msg)
		}
		teamResult, teamErr = chatgpt.ActivateTeam(ctx, session, sentinel, cardProv,
			cfg.Stripe, cfg.Team, fingerprint, regResult.Email, solver,
			emailProv, regResult.MailboxID, teamStatusFn)
		if teamErr != nil {
			r.logStep(index, regResult.Email, "Team 失败: "+teamErr.Error())
			log.Printf("[task-runner] %s #%d: Team failed: %v", r.taskID, index, teamErr)
		}
		if teamResult != nil {
			r.logStep(index, regResult.Email, "Team 开通成功: "+teamResult.TeamAccountID)
			if result.Plan == "failed" {
				result.Plan = "team"
			}
			savedAccessToken := session.AccessToken
			r.logStep(index, regResult.Email, "正在获取 Team OAuth 授权...")
			codexTokens, codexErr := auth.ObtainCodexTokens(ctx, client, regResult.DeviceID, teamResult.TeamAccountID)

			var wsAccessToken string
			if codexErr != nil {
				wsAccessToken, _ = chatgpt.GetWorkspaceAccessToken(client, teamResult.TeamAccountID)
				if wsAccessToken == "" {
					wsAccessToken = savedAccessToken
				}
			} else {
				wsAccessToken = codexTokens.AccessToken
			}

			teamAccount := buildAccountResult(fingerprint)
			teamAccount.AccessToken = wsAccessToken
			teamAccount.ChatGPTAccountID = teamResult.TeamAccountID
			teamAccount.TeamAccountID = teamResult.TeamAccountID
			teamAccount.WorkspaceToken = wsAccessToken
			if codexTokens != nil {
				teamAccount.RefreshToken = codexTokens.RefreshToken
				if codexTokens.IDToken != "" {
					teamAccount.IDToken = codexTokens.IDToken
				}
			}
			teamAccount.PlanType = "team"
			storage.SaveTeamAuthFile(cfg.Output.Dir, teamAccount)

			parentID := r.saveAccountToDB(teamAccount, "team_owner", nil, r.taskID)
			r.updateEmailRecord(regResult.Email, "used")
			r.logStep(index, regResult.Email, "Team 账号已保存")

			if cfg.Team.InviteCount > 0 && teamResult.TeamAccountID != "" {
				r.logStep(index, regResult.Email, fmt.Sprintf("正在邀请 %d 个成员...", cfg.Team.InviteCount))
				r.inviteMembers(ctx, cfg, client, emailProv, session, regResult, teamResult, fingerprint, parentID)
				r.logStep(index, regResult.Email, "成员邀请完成")
			}
		}
	}

	if plusResult == nil && teamResult == nil && taskType == TaskTypePlus {
		result.Plan = "free"
		freeAccount := buildAccountResult(fingerprint)
		freeAccount.PlanType = "free"
		storage.SaveFreeAuthFile(cfg.Output.Dir, freeAccount)
		r.saveAccountToDB(freeAccount, "plus", nil, r.taskID)
	}

	membershipState, verifyErr := r.verifyTaskMembership(ctx, taskType, session, teamAccountIDFromResult(teamResult), statusFn)
	if membershipState != nil {
		result.Plan = membershipState.resultPlanForTask(taskType)
	}
	if verifyErr != nil {
		reasons := []string{verifyErr.Error()}
		if plusErr != nil {
			reasons = append(reasons, "plus="+plusErr.Error())
		}
		if teamErr != nil {
			reasons = append(reasons, "team="+teamErr.Error())
		}
		result.Error = fmt.Errorf("最终会员校验失败: %s", strings.Join(reasons, "; "))
		r.updateEmailRecord(regResult.Email, "used_failed")
		return result
	}

	r.updateEmailRecord(regResult.Email, "used")
	return result
}

func (r *TaskRunner) inviteMembers(ctx context.Context, cfg *config.Config,
	client *httpclient.Client, emailProv email.EmailProvider,
	session *chatgpt.Session, regResult *auth.RegisterResult,
	teamResult *chatgpt.TeamResult, fingerprint *stripe.Fingerprint, parentID *uint) {

	invWsToken, invWsErr := chatgpt.GetWorkspaceAccessToken(client, teamResult.TeamAccountID)
	if invWsErr != nil {
		invWsToken = session.AccessToken
	}
	if invWsToken == "" {
		return
	}

	for invIdx := 1; invIdx <= cfg.Team.InviteCount; invIdx++ {
		if r.stopping.Load() || r.ctx.Err() != nil {
			break
		}

		r.logStep(0, regResult.Email, fmt.Sprintf("成员 %d/%d: 创建代理客户端...", invIdx, cfg.Team.InviteCount))

		// Use B2Proxy for member too, not just cfg.Proxy.URL (which may be empty in B2Proxy mode)
		var memberClient *httpclient.Client
		var memberClientErr error
		if cfg.Proxy.B2Proxy.Enabled {
			cardCountry := cfg.Card.ExpectedCountry()
			for try := 1; try <= 3; try++ {
				proxyURL, fetchErr := proxy.FetchB2Proxy(cfg.Proxy.B2Proxy, cardCountry)
				if fetchErr != nil {
					continue
				}
				memberClient, memberClientErr = httpclient.NewClient(proxyURL)
				if memberClientErr == nil {
					break
				}
			}
		} else if strings.Contains(cfg.Proxy.URL, "{SESSION}") {
			memberProxyURL := strings.ReplaceAll(cfg.Proxy.URL, "{SESSION}", fmt.Sprintf("inv%s%d", randomSessionID(), invIdx))
			memberClient, memberClientErr = httpclient.NewClient(memberProxyURL)
		} else {
			memberClient, memberClientErr = httpclient.NewClient(cfg.Proxy.URL)
		}
		_ = memberClientErr
		if memberClient == nil {
			r.logStep(0, regResult.Email, fmt.Sprintf("成员 %d: 代理连接失败，跳过", invIdx))
			continue
		}

		r.logStep(0, regResult.Email, fmt.Sprintf("成员 %d: 注册中...", invIdx))
		memberPassword := auth.GenerateRandomPassword(cfg.Account.PasswordLength)
		memberReg, err := auth.Register(ctx, memberClient, emailProv, memberPassword)
		if err != nil {
			r.logStep(0, regResult.Email, fmt.Sprintf("成员 %d: 注册失败: %v", invIdx, err))
			continue
		}
		r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 注册成功: %s", invIdx, memberReg.Email))

		r.saveEmailRecord(memberReg.Email, "in_use", "member")

		r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 发送邀请...", invIdx))
		if err := chatgpt.InviteToTeam(client, invWsToken, teamResult.TeamAccountID, session.DeviceID, memberReg.Email); err != nil {
			r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 邀请失败: %v", invIdx, err))
			r.updateEmailRecord(memberReg.Email, "used_failed")
			continue
		}
		r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 邀请成功，等待传播...", invIdx))

		time.Sleep(3 * time.Second)

		var memberLoginResult *auth.LoginResult
		if memberReg.Tokens != nil {
			memberLoginResult = memberReg.Tokens
			r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 使用注册令牌", invIdx))
		} else {
			r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 登录中...", invIdx))
			memberSentinel := auth.NewSentinelGeneratorWithDeviceID(memberClient, memberReg.DeviceID)
			memberLoginResult, err = auth.Login(ctx, memberClient, memberReg.Email, memberPassword,
				memberReg.DeviceID, memberSentinel, memberReg.MailboxID, emailProv)
			if err != nil {
				r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 登录失败: %v", invIdx, err))
				r.updateEmailRecord(memberReg.Email, "used_failed")
				continue
			}
			r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 登录成功", invIdx))
		}

		var memberWsToken, memberRefreshToken, memberIDToken, memberAccountID string
		memberWsToken = memberLoginResult.AccessToken
		memberRefreshToken = memberLoginResult.RefreshToken
		memberIDToken = memberLoginResult.IDToken
		memberAccountID = memberLoginResult.ChatGPTAccountID

		r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 获取 Team OAuth 授权...", invIdx))
		for codexTry := 1; codexTry <= 3; codexTry++ {
			memberCodex, err := auth.ObtainCodexTokens(ctx, memberClient, memberReg.DeviceID, teamResult.TeamAccountID)
			if err != nil {
				if codexTry < 3 {
					time.Sleep(3 * time.Second)
				}
				continue
			}
			if memberCodex.ChatGPTAccountID == teamResult.TeamAccountID {
				memberWsToken = memberCodex.AccessToken
				memberRefreshToken = memberCodex.RefreshToken
				if memberCodex.IDToken != "" {
					memberIDToken = memberCodex.IDToken
				}
				memberAccountID = memberCodex.ChatGPTAccountID
				r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: Team OAuth 成功", invIdx))
				break
			}
			if codexTry < 3 {
				time.Sleep(3 * time.Second)
			}
		}

		memberAuth := &chatgpt.AccountResult{
			Email: memberReg.Email, Password: memberPassword,
			AccessToken: memberWsToken, RefreshToken: memberRefreshToken,
			IDToken: memberIDToken, ChatGPTAccountID: memberAccountID,
			ChatGPTUserID: memberLoginResult.ChatGPTUserID,
			TeamAccountID: teamResult.TeamAccountID, WorkspaceToken: memberWsToken,
			PlanType: "team", Proxy: cfg.Proxy.URL,
			CreatedAt: time.Now().UTC().Format(time.RFC3339),
		}
		storage.SaveTeamAuthFile(cfg.Output.Dir, memberAuth)
		r.saveAccountToDB(memberAuth, "team_member", parentID, r.taskID)
		r.updateEmailRecord(memberReg.Email, "used_member")
		r.logStep(0, memberReg.Email, fmt.Sprintf("成员 %d: 已保存 (account=%s)", invIdx, memberAccountID))
	}
}

func (r *TaskRunner) logStep(index int, email, message string) {
	r.gormDB.Create(&db.TaskLog{
		TaskID:  r.taskID,
		Index:   index,
		Email:   email,
		Status:  "step",
		Message: message,
	})
}

func (r *TaskRunner) logResult(t *db.Task, index int, email, status, plan, errMsg string, duration int) {
	message := status
	if errMsg != "" {
		message = errMsg
	} else if plan != "" {
		message = "membership verified: " + plan
	}
	logEntry := &db.TaskLog{
		TaskID:   r.taskID,
		Index:    index,
		Email:    email,
		Status:   status,
		Plan:     plan,
		Error:    errMsg,
		Duration: duration,
		Message:  message,
	}
	r.gormDB.Create(logEntry)

	t.DoneCount++
	if status == LogStatusSuccess {
		t.SuccessCount++
	} else {
		t.FailCount++
	}
	r.gormDB.Model(t).Updates(map[string]interface{}{
		"done_count":    t.DoneCount,
		"success_count": t.SuccessCount,
		"fail_count":    t.FailCount,
	})
}

func (r *TaskRunner) failTask(t *db.Task, errMsg string) {
	now := time.Now()
	t.Status = StatusStopped
	t.StoppedAt = &now
	r.gormDB.Save(t)
	log.Printf("[task-runner] %s: task failed: %s", r.taskID, errMsg)
}

func (r *TaskRunner) saveEmailRecord(email, status, role string) {
	rec := &db.EmailRecord{
		Email:       email,
		Status:      status,
		UsedForRole: role,
		TaskID:      r.taskID,
	}
	r.gormDB.Create(rec)
}

func (r *TaskRunner) updateEmailRecord(email, status string) {
	r.gormDB.Model(&db.EmailRecord{}).Where("email = ?", email).Update("status", status)
}

func teamAccountIDFromResult(result *chatgpt.TeamResult) string {
	if result == nil {
		return ""
	}
	return result.TeamAccountID
}

func accountStatusForResult(result *chatgpt.AccountResult) string {
	if result == nil {
		return "active"
	}
	switch result.PlanType {
	case "free", "plus", "team":
		return result.PlanType
	default:
		return "active"
	}
}

func (r *TaskRunner) saveAccountToDB(result *chatgpt.AccountResult, plan string, parentID *uint, taskID string) *uint {
	// Check if account already exists (same email)
	var existing db.Account
	if r.gormDB.Where("email = ?", result.Email).First(&existing).Error == nil {
		// Update existing record (e.g. Plus account now also has Team)
		updates := map[string]interface{}{
			"plan":              plan,
			"access_token":      result.AccessToken,
			"refresh_token":     result.RefreshToken,
			"id_token":          result.IDToken,
			"account_id":        result.ChatGPTAccountID,
			"user_id":           result.ChatGPTUserID,
			"team_workspace_id": result.TeamAccountID,
			"workspace_token":   result.WorkspaceToken,
			"status":            accountStatusForResult(result),
		}
		if parentID != nil {
			updates["parent_id"] = *parentID
		}
		r.gormDB.Model(&existing).Updates(updates)
		return &existing.ID
	}

	acct := &db.Account{
		TaskID:          taskID,
		Email:           result.Email,
		Password:        result.Password,
		Plan:            plan,
		ParentID:        parentID,
		AccessToken:     result.AccessToken,
		RefreshToken:    result.RefreshToken,
		IDToken:         result.IDToken,
		AccountID:       result.ChatGPTAccountID,
		UserID:          result.ChatGPTUserID,
		TeamWorkspaceID: result.TeamAccountID,
		WorkspaceToken:  result.WorkspaceToken,
		Status:          accountStatusForResult(result),
	}
	r.gormDB.Create(acct)
	return &acct.ID
}

func (r *TaskRunner) getB2ProxyClient(cfg *config.Config, cardCountry string, index int, solver *captcha.Solver) *httpclient.Client {
	for try := 1; try <= 5; try++ {
		proxyURL, err := proxy.FetchB2Proxy(cfg.Proxy.B2Proxy, cardCountry)
		if err != nil {
			time.Sleep(time.Second)
			continue
		}
		c, err := httpclient.NewClient(proxyURL)
		if err != nil {
			continue
		}
		if testStripeConnectivity(c) {
			if solver != nil {
				solver.SetProxy(proxyURL)
			}
			return c
		}
	}
	return nil
}

func (r *TaskRunner) getSessionClient(cfg *config.Config, index int, solver *captcha.Solver) *httpclient.Client {
	for try := 1; try <= 5; try++ {
		sessionID := randomSessionID()
		proxyURL := strings.ReplaceAll(cfg.Proxy.URL, "{SESSION}", sessionID)
		c, err := httpclient.NewClient(proxyURL)
		if err != nil {
			continue
		}
		if testStripeConnectivity(c) {
			if solver != nil {
				solver.SetProxy(proxyURL)
			}
			return c
		}
	}
	return nil
}

func testStripeConnectivity(client *httpclient.Client) bool {
	req, err := http.NewRequest("GET", "https://m.stripe.com/6", nil)
	if err != nil {
		return false
	}
	req.Header.Set("User-Agent", "Mozilla/5.0")
	resp, err := client.Do(req)
	if err != nil {
		return false
	}
	resp.Body.Close()

	req2, _ := http.NewRequest("GET", "https://api.stripe.com", nil)
	req2.Header.Set("User-Agent", "Mozilla/5.0")
	resp2, err := client.Do(req2)
	if err != nil {
		return false
	}
	resp2.Body.Close()
	return true
}

func generatePassword(length int) string {
	const charset = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@#$%"
	b := make([]byte, length)
	for i := range b {
		n, err := rand.Int(rand.Reader, big.NewInt(int64(len(charset))))
		if err != nil {
			b[i] = charset[i%len(charset)]
			continue
		}
		b[i] = charset[n.Int64()]
	}
	return string(b)
}

func randomSessionID() string {
	const chars = "abcdefghijklmnopqrstuvwxyz0123456789"
	b := make([]byte, 16)
	for i := range b {
		n, _ := rand.Int(rand.Reader, big.NewInt(int64(len(chars))))
		b[i] = chars[n.Int64()]
	}
	return string(b)
}