3x-ui/internal/web/service/client_bulk.go

package service

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"github.com/google/uuid"

	"github.com/mhsanaei/3x-ui/v3/internal/database"
	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
	"github.com/mhsanaei/3x-ui/v3/internal/logger"
	"github.com/mhsanaei/3x-ui/v3/internal/util/common"
	"github.com/mhsanaei/3x-ui/v3/internal/xray"

	"gorm.io/gorm"
)

// BulkAttachResult reports the outcome of a bulk attach across target inbounds.
type BulkAttachResult struct {
	Attached []string `json:"attached"`
	Skipped  []string `json:"skipped"`
	Errors   []string `json:"errors"`
}

// BulkAttach attaches the given existing clients (by email) to each target inbound,
// reusing their identity (email/UUID/password/subId) and a shared traffic row. It adds
// all clients to a target in a single AddInboundClient call, and reports clients already
// present on a target as skipped.
func (s *ClientService) BulkAttach(inboundSvc *InboundService, emails []string, inboundIds []int) (*BulkAttachResult, bool, error) {
	result := &BulkAttachResult{}
	if len(emails) == 0 || len(inboundIds) == 0 {
		return result, false, nil
	}

	recordErr := func(format string, args ...any) {
		msg := fmt.Sprintf(format, args...)
		result.Errors = append(result.Errors, msg)
		logger.Warningf("[BulkAttach] %s", msg)
	}

	records := make([]*model.ClientRecord, 0, len(emails))
	seenEmail := make(map[string]struct{}, len(emails))
	for _, email := range emails {
		if email == "" {
			continue
		}
		key := strings.ToLower(email)
		if _, ok := seenEmail[key]; ok {
			continue
		}
		seenEmail[key] = struct{}{}
		rec, err := s.GetRecordByEmail(nil, email)
		if err != nil {
			recordErr("%s: %v", email, err)
			continue
		}
		records = append(records, rec)
	}

	needRestart := false
	// Prepared in order first, as in Create: fillProtocolDefaults mints the
	// shared credentials, so only the node pushes below may overlap.
	attachIds := make([]int, 0, len(inboundIds))
	attachPayloads := make([]string, 0, len(inboundIds))
	attachClients := make([][]model.Client, 0, len(inboundIds))
	attachAnyTunnel := false
	// A repeated id used to be caught by the second pass seeing the client
	// already attached; the applies no longer run before the next prep.
	seenInbound := make(map[int]struct{}, len(inboundIds))
	for _, ibId := range inboundIds {
		if _, dup := seenInbound[ibId]; dup {
			continue
		}
		seenInbound[ibId] = struct{}{}
		inbound, err := inboundSvc.GetInbound(ibId)
		if err != nil {
			recordErr("inbound %d: %v", ibId, err)
			continue
		}
		if inbound.Protocol == model.WireGuard || inbound.Protocol == model.AmneziaWG {
			attachAnyTunnel = true
		}
		existingClients, err := inboundSvc.GetClients(inbound)
		if err != nil {
			recordErr("inbound %d: %v", ibId, err)
			continue
		}
		have := make(map[string]struct{}, len(existingClients))
		for _, c := range existingClients {
			have[strings.ToLower(c.Email)] = struct{}{}
		}

		clientsToAdd := make([]model.Client, 0, len(records))
		for _, rec := range records {
			if _, attached := have[strings.ToLower(rec.Email)]; attached {
				result.Skipped = append(result.Skipped, rec.Email)
				continue
			}
			client := *rec.ToClient()
			client.UpdatedAt = time.Now().UnixMilli()
			if err := s.fillProtocolDefaults(&client, inbound); err != nil {
				recordErr("%s -> inbound %d: %v", rec.Email, ibId, err)
				continue
			}
			clientsToAdd = append(clientsToAdd, clientWithInboundFlow(client, inbound))
		}

		if len(clientsToAdd) == 0 {
			continue
		}

		payload, err := json.Marshal(map[string][]model.Client{"clients": clientsToAdd})
		if err != nil {
			recordErr("inbound %d: %v", ibId, err)
			continue
		}
		attachIds = append(attachIds, ibId)
		attachPayloads = append(attachPayloads, string(payload))
		attachClients = append(attachClients, clientsToAdd)
	}

	attachResults, attachPanics := fanoutInboundResults(attachIds, addFanoutLimit(attachAnyTunnel), func(i int) inboundApplyOutcome {
		nr, err := s.AddInboundClient(inboundSvc, &model.Inbound{Id: attachIds[i], Settings: attachPayloads[i]})
		return inboundApplyOutcome{needRestart: nr, err: err}
	})
	for i, out := range attachResults {
		err := out.err
		if attachPanics[i] != nil {
			// The apply may already have committed, so ask for the restart the
			// lost return value can no longer report.
			needRestart = true
			err = attachPanics[i]
		}
		if err != nil {
			recordErr("inbound %d: %v", attachIds[i], err)
			continue
		}
		if out.needRestart {
			needRestart = true
		}
		for _, c := range attachClients[i] {
			result.Attached = append(result.Attached, c.Email)
		}
	}

	return result, needRestart, nil
}

// BulkDetachResult reports the outcome of a bulk detach across target inbounds.
type BulkDetachResult struct {
	Detached []string `json:"detached"`
	Skipped  []string `json:"skipped"`
	Errors   []string `json:"errors"`
}

// BulkDetach detaches the given existing clients (by email) from each target inbound.
// (email, inbound) pairs where the client is not currently attached are silently skipped
// at the inbound level; emails that aren't attached to any of the requested inbounds
// are reported under skipped. ClientRecord rows are kept even when they become orphaned
// (matches single-client detach semantics); callers should use bulkDelete for full removal.
func (s *ClientService) BulkDetach(inboundSvc *InboundService, emails []string, inboundIds []int) (*BulkDetachResult, bool, error) {
	result := &BulkDetachResult{}
	if len(emails) == 0 || len(inboundIds) == 0 {
		return result, false, nil
	}

	recordErr := func(format string, args ...any) {
		msg := fmt.Sprintf(format, args...)
		result.Errors = append(result.Errors, msg)
		logger.Warningf("[BulkDetach] %s", msg)
	}

	requested := make(map[int]struct{}, len(inboundIds))
	for _, id := range inboundIds {
		requested[id] = struct{}{}
	}

	recsByInbound := make(map[int][]*model.ClientRecord)
	emailOrder := make([]string, 0, len(emails))
	emailRepr := make(map[string]string, len(emails))
	emailFailed := make(map[string]bool, len(emails))
	seenEmail := make(map[string]struct{}, len(emails))
	for _, email := range emails {
		if email == "" {
			continue
		}
		key := strings.ToLower(email)
		if _, ok := seenEmail[key]; ok {
			continue
		}
		seenEmail[key] = struct{}{}

		rec, err := s.GetRecordByEmail(nil, email)
		if err != nil {
			recordErr("%s: %v", email, err)
			continue
		}
		currentIds, err := s.GetInboundIdsForRecord(rec.Id)
		if err != nil {
			recordErr("%s: %v", email, err)
			continue
		}
		matched := false
		for _, id := range currentIds {
			if _, ok := requested[id]; ok {
				recsByInbound[id] = append(recsByInbound[id], rec)
				matched = true
			}
		}
		if !matched {
			result.Skipped = append(result.Skipped, rec.Email)
			continue
		}
		emailOrder = append(emailOrder, key)
		emailRepr[key] = rec.Email
	}

	needRestart := false
	// Ordered and de-duplicated up front: the sequential loop dropped each map
	// entry as it went, which the concurrent applies can no longer do.
	detachIds := make([]int, 0, len(recsByInbound))
	detachRecs := make([][]*model.ClientRecord, 0, len(recsByInbound))
	for _, ibId := range inboundIds {
		recs, ok := recsByInbound[ibId]
		if !ok {
			continue
		}
		delete(recsByInbound, ibId)
		detachIds = append(detachIds, ibId)
		detachRecs = append(detachRecs, recs)
	}
	detachResults, detachPanics := fanoutInboundResults(detachIds, inboundFanoutConcurrency, func(i int) inboundApplyOutcome {
		nr, err := s.delInboundClients(inboundSvc, detachIds[i], detachRecs[i], true)
		return inboundApplyOutcome{needRestart: nr, err: err}
	})
	for i, out := range detachResults {
		err := out.err
		if detachPanics[i] != nil {
			// See BulkAttach: a panicking apply may already have committed.
			needRestart = true
			err = detachPanics[i]
		}
		if err != nil {
			recordErr("inbound %d: %v", detachIds[i], err)
			for _, rec := range detachRecs[i] {
				emailFailed[strings.ToLower(rec.Email)] = true
			}
			continue
		}
		if out.needRestart {
			needRestart = true
		}
	}

	for _, key := range emailOrder {
		if emailFailed[key] {
			continue
		}
		result.Detached = append(result.Detached, emailRepr[key])
	}

	return result, needRestart, nil
}

// inboundApplyOutcome carries one inbound's apply result out of the fanout.
type inboundApplyOutcome struct {
	needRestart bool
	err         error
}

// BulkAdjustResult is returned by BulkAdjust to report how many clients were
// successfully updated and which were skipped (typically because the field
// being adjusted was unlimited for that client) or failed.
type BulkAdjustResult struct {
	Adjusted int                `json:"adjusted"`
	Skipped  []BulkAdjustReport `json:"skipped,omitempty"`
}

type BulkAdjustReport struct {
	Email  string `json:"email"`
	Reason string `json:"reason"`
}

type bulkAdjustEntry struct {
	record      *model.ClientRecord
	applyExpiry bool
	newExpiry   int64
	applyTotal  bool
	newTotal    int64
}

// bulkFlowClear is the directive that strips the XTLS flow from every selected
// client. The vision values are the only positive flows xray accepts.
const bulkFlowClear = "none"

// bulkFlowAllowed whitelists the flow directives BulkAdjust accepts. Anything
// outside this set is treated as "" (leave flow untouched) so a malformed or
// hostile value can never be injected into a client's settings. The dropdown in
// ClientBulkAdjustModal.tsx offers the same set ("" / "none" / TLS_FLOW_CONTROL);
// keep the two in sync.
var bulkFlowAllowed = map[string]struct{}{
	"":                        {},
	bulkFlowClear:             {},
	"xtls-rprx-vision":        {},
	"xtls-rprx-vision-udp443": {},
}

// BulkAdjust shifts ExpiryTime by addDays (days) and TotalGB by addBytes
// for every email in the list. Clients whose corresponding field is
// unlimited (0) are skipped — bulk extend should not accidentally
// limit an unlimited client. addDays and addBytes may be negative.
//
// Like BulkDelete, the work is grouped by inbound so each inbound's
// settings JSON is parsed and written exactly once regardless of how
// many target emails it contains.
func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string, addDays int, addBytes int64, flow string) (BulkAdjustResult, bool, error) {
	result := BulkAdjustResult{}
	if len(emails) == 0 {
		return result, false, nil
	}
	flow = strings.TrimSpace(flow)
	if _, ok := bulkFlowAllowed[flow]; !ok {
		flow = "" // ignore unknown directives — "" means "leave flow untouched"
	}
	adjustFlow := flow != ""
	if addDays == 0 && addBytes == 0 && !adjustFlow {
		return result, false, common.NewError("no adjustment specified")
	}

	addExpiryMs := int64(addDays) * 24 * 60 * 60 * 1000

	seen := map[string]struct{}{}
	cleanEmails := make([]string, 0, len(emails))
	for _, e := range emails {
		e = strings.TrimSpace(e)
		if e == "" {
			continue
		}
		if _, ok := seen[e]; ok {
			continue
		}
		seen[e] = struct{}{}
		cleanEmails = append(cleanEmails, e)
	}
	if len(cleanEmails) == 0 {
		return result, false, nil
	}

	db := database.GetDB()

	var records []model.ClientRecord
	for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
		var rows []model.ClientRecord
		if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
			return result, false, err
		}
		records = append(records, rows...)
	}
	recordsByEmail := make(map[string]*model.ClientRecord, len(records))
	for i := range records {
		recordsByEmail[records[i].Email] = &records[i]
	}

	skippedReasons := map[string]string{}
	for _, email := range cleanEmails {
		if _, ok := recordsByEmail[email]; !ok {
			skippedReasons[email] = "client not found"
		}
	}

	plan := map[string]*bulkAdjustEntry{}
	for email, rec := range recordsByEmail {
		entry := &bulkAdjustEntry{record: rec}
		if addDays != 0 {
			switch {
			case rec.ExpiryTime == 0:
				if _, exists := skippedReasons[email]; !exists {
					skippedReasons[email] = "unlimited expiry"
				}
			case rec.ExpiryTime > 0:
				next := rec.ExpiryTime + addExpiryMs
				if next <= 0 {
					if _, exists := skippedReasons[email]; !exists {
						skippedReasons[email] = "reduction exceeds remaining time"
					}
				} else {
					entry.applyExpiry = true
					entry.newExpiry = next
				}
			default:
				next := rec.ExpiryTime - addExpiryMs
				if next >= 0 {
					if _, exists := skippedReasons[email]; !exists {
						skippedReasons[email] = "reduction exceeds delay window"
					}
				} else {
					entry.applyExpiry = true
					entry.newExpiry = next
				}
			}
		}
		if addBytes != 0 {
			if rec.TotalGB == 0 {
				if _, exists := skippedReasons[email]; !exists {
					skippedReasons[email] = "unlimited traffic"
				}
			} else {
				next := rec.TotalGB + addBytes
				if next <= 0 {
					if _, exists := skippedReasons[email]; !exists {
						skippedReasons[email] = "reduction exceeds remaining quota"
					}
				} else {
					entry.applyTotal = true
					entry.newTotal = next
				}
			}
		}
		if entry.applyExpiry || entry.applyTotal || adjustFlow {
			plan[email] = entry
		}
	}

	if len(plan) == 0 {
		for email, reason := range skippedReasons {
			result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: reason})
		}
		return result, false, nil
	}

	plannedIds := make([]int, 0, len(plan))
	recordIdToEmail := make(map[int]string, len(plan))
	for email, entry := range plan {
		plannedIds = append(plannedIds, entry.record.Id)
		recordIdToEmail[entry.record.Id] = email
	}

	var mappings []model.ClientInbound
	for _, batch := range chunkInts(plannedIds, sqlInChunk) {
		var rows []model.ClientInbound
		if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
			return result, false, err
		}
		mappings = append(mappings, rows...)
	}
	emailsByInbound := map[int][]string{}
	for _, m := range mappings {
		email, ok := recordIdToEmail[m.ClientId]
		if !ok {
			continue
		}
		emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
	}

	needRestart := false
	flowHonored := map[string]bool{}
	flowIneligible := map[string]bool{}
	execFailed := map[string]bool{}
	adjustIds := sortedInboundIds(emailsByInbound)
	adjustResults, adjustPanics := fanoutInboundResults(adjustIds, inboundFanoutConcurrency, func(i int) bulkInboundAdjustResult {
		return s.bulkAdjustInboundClients(inboundSvc, adjustIds[i], emailsByInbound[adjustIds[i]], plan, flow)
	})
	for i, ibRes := range adjustResults {
		if adjustPanics[i] != nil {
			needRestart = true
			for _, email := range emailsByInbound[adjustIds[i]] {
				execFailed[email] = true
				if _, already := skippedReasons[email]; !already {
					skippedReasons[email] = adjustPanics[i].Error()
				}
			}
			continue
		}
		if ibRes.needRestart {
			needRestart = true
		}
		for email := range ibRes.flowHonored {
			flowHonored[email] = true
		}
		for email := range ibRes.flowIneligible {
			flowIneligible[email] = true
		}
		for email, reason := range ibRes.perEmailSkipped {
			execFailed[email] = true
			if _, already := skippedReasons[email]; !already {
				skippedReasons[email] = reason
			}
		}
	}

	cond, condArgs := depletedCond(db)
	candidateEmails := make([]string, 0, len(plan))
	for email, entry := range plan {
		if entry.applyExpiry || entry.applyTotal {
			candidateEmails = append(candidateEmails, email)
		}
	}
	wasDisabledDepleted := map[string]struct{}{}
	for _, batch := range chunkStrings(candidateEmails, sqlInChunk) {
		var rows []string
		if err := db.Model(xray.ClientTraffic{}).
			Where(cond+" AND enable = ? AND email IN ?", append(append([]any{}, condArgs...), false, batch)...).
			Pluck("email", &rows).Error; err != nil {
			return result, needRestart, err
		}
		for _, e := range rows {
			wasDisabledDepleted[e] = struct{}{}
		}
	}

	adjusted := map[string]struct{}{}
	for email, entry := range plan {
		if execFailed[email] {
			continue
		}
		updates := map[string]any{}
		if entry.applyExpiry {
			updates["expiry_time"] = entry.newExpiry
		}
		if entry.applyTotal {
			updates["total"] = entry.newTotal
		}
		if len(updates) > 0 {
			if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Updates(updates).Error; err != nil {
				if _, already := skippedReasons[email]; !already {
					skippedReasons[email] = err.Error()
				}
				continue
			}
		}
		// Counted when expiry/total changed, or a flow directive was honored
		// for this client (flow lives in the inbound JSON, not ClientTraffic).
		if len(updates) > 0 || flowHonored[email] {
			adjusted[email] = struct{}{}
		}
	}
	result.Adjusted = len(adjusted)

	for email, reason := range skippedReasons {
		result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: reason})
	}
	// Report a flow directive that no inbound could carry — only when it was not
	// honored anywhere and the client has no other (expiry/total) skip reason.
	// The expiry/total part, if any, has already been applied and counted above.
	for email := range flowIneligible {
		if flowHonored[email] {
			continue
		}
		if _, already := skippedReasons[email]; already {
			continue
		}
		result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: "flow not supported on inbound"})
	}

	if len(wasDisabledDepleted) > 0 {
		stillDepleted := map[string]struct{}{}
		wasList := make([]string, 0, len(wasDisabledDepleted))
		for e := range wasDisabledDepleted {
			wasList = append(wasList, e)
		}
		for _, batch := range chunkStrings(wasList, sqlInChunk) {
			var rows []string
			if err := db.Model(xray.ClientTraffic{}).
				Where(cond+" AND email IN ?", append(append([]any{}, condArgs...), batch)...).
				Pluck("email", &rows).Error; err != nil {
				return result, needRestart, err
			}
			for _, e := range rows {
				stillDepleted[e] = struct{}{}
			}
		}
		reEnable := make([]string, 0, len(wasDisabledDepleted))
		for e := range wasDisabledDepleted {
			if _, still := stillDepleted[e]; !still {
				reEnable = append(reEnable, e)
			}
		}
		if len(reEnable) > 0 {
			_, nr, err := s.BulkSetEnable(inboundSvc, reEnable, true)
			if err != nil {
				return result, needRestart, err
			}
			if nr {
				needRestart = true
			}
		}
	}

	return result, needRestart, nil
}

type bulkInboundAdjustResult struct {
	perEmailSkipped map[string]string
	flowHonored     map[string]bool
	// flowIneligible is tracked apart from perEmailSkipped: a flow directive
	// that an inbound cannot carry must not suppress the expiry/total write for
	// the same client (which would diverge the inbound JSON / ClientRecord from
	// ClientTraffic). It only feeds the final Skipped report.
	flowIneligible map[string]bool
	needRestart    bool
}

// bulkAdjustInboundClients applies expiry/total deltas to multiple clients
// inside a single inbound's settings JSON. The xray runtime is updated
// only for remote-node inbounds; local nodes do not need a notification
// because the AddUser payload does not include totalGB/expiryTime —
// changing those fields is identity-preserving and the panel's traffic
// enforcement loop picks up the new limits from ClientTraffic directly.
func (s *ClientService) bulkAdjustInboundClients(
	inboundSvc *InboundService,
	inboundId int,
	emails []string,
	plan map[string]*bulkAdjustEntry,
	flow string,
) bulkInboundAdjustResult {
	res := bulkInboundAdjustResult{perEmailSkipped: map[string]string{}, flowHonored: map[string]bool{}, flowIneligible: map[string]bool{}}

	defer lockInbound(inboundId).Unlock()

	oldInbound, err := inboundSvc.GetInbound(inboundId)
	if err != nil {
		logger.Error("Load Old Data Error")
		for _, e := range emails {
			res.perEmailSkipped[e] = err.Error()
		}
		return res
	}

	var settings map[string]any
	if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
		for _, e := range emails {
			res.perEmailSkipped[e] = err.Error()
		}
		return res
	}

	// Match by email — the client's stable identity (see Delete). Credentials
	// can drift from the inbound JSON, so they are never used for matching.
	wantedEmails := make(map[string]struct{}, len(emails))
	for _, email := range emails {
		if plan[email] == nil {
			res.perEmailSkipped[email] = "client not found"
			continue
		}
		wantedEmails[email] = struct{}{}
	}

	// Flow eligibility is a property of the inbound (protocol + transport), so
	// resolve it once. Clearing flow is always allowed; setting a vision flow
	// is only honored on an inbound that can carry it.
	flowEligible := flow == bulkFlowClear ||
		(!oldInbound.DisableFlow &&
			inboundCanEnableTlsFlow(string(oldInbound.Protocol), oldInbound.StreamSettings, oldInbound.Settings))

	interfaceClients, _ := settings["clients"].([]any)
	foundEmails := map[string]bool{}
	flowChanged := false
	nowMs := time.Now().Unix() * 1000
	for i, client := range interfaceClients {
		c, ok := client.(map[string]any)
		if !ok {
			continue
		}
		targetEmail, _ := c["email"].(string)
		if _, want := wantedEmails[targetEmail]; !want || targetEmail == "" {
			continue
		}
		entry := plan[targetEmail]
		if entry.applyExpiry {
			c["expiryTime"] = entry.newExpiry
		}
		if entry.applyTotal {
			c["totalGB"] = entry.newTotal
		}
		if flow != "" {
			if flowEligible {
				want := ""
				if flow != bulkFlowClear {
					want = flow
				}
				if cur, _ := c["flow"].(string); cur != want {
					c["flow"] = want
					flowChanged = true
				}
				res.flowHonored[targetEmail] = true
			} else {
				// Record separately so this never suppresses the expiry/total
				// write for the same client (see flowIneligible doc).
				res.flowIneligible[targetEmail] = true
			}
		}
		c["updated_at"] = nowMs
		interfaceClients[i] = c
		foundEmails[targetEmail] = true
	}

	for email := range wantedEmails {
		if !foundEmails[email] {
			res.perEmailSkipped[email] = "Client Not Found In Inbound"
		}
	}

	if len(foundEmails) == 0 {
		return res
	}

	settings["clients"] = interfaceClients
	newSettings, err := json.MarshalIndent(settings, "", "  ")
	if err != nil {
		for email := range foundEmails {
			res.perEmailSkipped[email] = err.Error()
		}
		return res
	}
	oldInbound.Settings = string(newSettings)

	// A flow change rewrites the user's xray config, which the lightweight
	// UpdateUser push below does not carry. Local nodes reload via restart;
	// remote nodes get a full reconcile (MarkNodeDirty) instead of a per-user push.
	if flowChanged && oldInbound.NodeID == nil {
		res.needRestart = true
	}

	// Serialize against the traffic poll to avoid the cross-transaction
	// lock-order deadlock on inbounds/client_records (runSerializedTx).
	txErr := runSerializedTx(func(tx *gorm.DB) error {
		if err := tx.Save(oldInbound).Error; err != nil {
			return err
		}
		finalClients, gcErr := inboundSvc.GetClients(oldInbound)
		if gcErr != nil {
			return gcErr
		}
		if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
			return err
		}
		if oldInbound.NodeID != nil {
			return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
		}
		return nil
	})
	if txErr != nil {
		for email := range foundEmails {
			if _, skip := res.perEmailSkipped[email]; !skip {
				res.perEmailSkipped[email] = txErr.Error()
			}
		}
	} else if oldInbound.NodeID != nil && !flowChanged && len(foundEmails) <= nodeBulkPushThreshold {
		rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
		if perr != nil {
			logger.Warning("BulkAdjust: node runtime lookup after commit failed:", perr)
		} else if push {
			for email := range foundEmails {
				entry := plan[email]
				updated := *entry.record.ToClient()
				if entry.applyExpiry {
					updated.ExpiryTime = entry.newExpiry
				}
				if entry.applyTotal {
					updated.TotalGB = entry.newTotal
				}
				updated.UpdatedAt = nowMs
				ctx, cancel := nodePushContext()
				err1 := rt.UpdateUser(ctx, oldInbound, email, updated)
				cancel()
				if err1 != nil {
					logger.Warning("Error in updating client on", rt.Name(), ":", err1)
					// First failure ends the batch push; the reconcile converges the rest.
					break
				}
			}
		}
	}

	return res
}

// BulkDeleteResult mirrors BulkAdjustResult: total deleted plus per-email
// skip reasons when an email could not be processed.
type BulkDeleteResult struct {
	Deleted int                `json:"deleted"`
	Skipped []BulkDeleteReport `json:"skipped,omitempty"`
}

type BulkDeleteReport struct {
	Email  string `json:"email"`
	Reason string `json:"reason"`
}

// BulkDelete removes every client in the list in one optimized pass.
// Instead of running the full single-delete pipeline N times (which would
// re-read, re-parse, and re-write each inbound's settings JSON for every
// email), it groups emails by inbound and performs a single
// read-modify-write per inbound. Per-row DB cleanups are also batched with
// IN-clause queries at the end. Errors on a particular email are recorded
// in the Skipped list and processing continues for the rest.
func (s *ClientService) BulkDelete(inboundSvc *InboundService, emails []string, keepTraffic bool) (BulkDeleteResult, bool, error) {
	result := BulkDeleteResult{}

	seen := map[string]struct{}{}
	cleanEmails := make([]string, 0, len(emails))
	for _, e := range emails {
		e = strings.TrimSpace(e)
		if e == "" {
			continue
		}
		if _, ok := seen[e]; ok {
			continue
		}
		seen[e] = struct{}{}
		cleanEmails = append(cleanEmails, e)
	}
	if len(cleanEmails) == 0 {
		return result, false, nil
	}

	db := database.GetDB()

	var records []model.ClientRecord
	for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
		var rows []model.ClientRecord
		if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
			return result, false, err
		}
		records = append(records, rows...)
	}
	recordsByEmail := make(map[string]*model.ClientRecord, len(records))
	tombstoneEmails := make([]string, 0, len(records))
	for i := range records {
		recordsByEmail[records[i].Email] = &records[i]
		tombstoneEmails = append(tombstoneEmails, records[i].Email)
	}
	tombstoneClientEmails(tombstoneEmails)

	skippedReasons := map[string]string{}
	for _, email := range cleanEmails {
		if _, ok := recordsByEmail[email]; !ok {
			skippedReasons[email] = "client not found"
		}
	}

	clientIds := make([]int, 0, len(recordsByEmail))
	recordIdToEmail := make(map[int]string, len(recordsByEmail))
	for _, r := range recordsByEmail {
		clientIds = append(clientIds, r.Id)
		recordIdToEmail[r.Id] = r.Email
	}

	emailsByInbound := map[int][]string{}
	if len(clientIds) > 0 {
		var mappings []model.ClientInbound
		for _, batch := range chunkInts(clientIds, sqlInChunk) {
			var rows []model.ClientInbound
			if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
				return result, false, err
			}
			mappings = append(mappings, rows...)
		}
		for _, m := range mappings {
			email, ok := recordIdToEmail[m.ClientId]
			if !ok {
				continue
			}
			emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
		}
	}

	needRestart := false
	delIds := sortedInboundIds(emailsByInbound)
	delResults, delPanics := fanoutInboundResults(delIds, inboundFanoutConcurrency, func(i int) bulkInboundDeleteResult {
		return s.bulkDelInboundClients(inboundSvc, delIds[i], emailsByInbound[delIds[i]], recordsByEmail, keepTraffic)
	})
	for i, ibResult := range delResults {
		if delPanics[i] != nil {
			needRestart = true
			for _, email := range emailsByInbound[delIds[i]] {
				if _, already := skippedReasons[email]; !already {
					skippedReasons[email] = delPanics[i].Error()
				}
			}
			continue
		}
		if ibResult.needRestart {
			needRestart = true
		}
		for email, reason := range ibResult.perEmailSkipped {
			if _, already := skippedReasons[email]; !already {
				skippedReasons[email] = reason
			}
		}
	}

	successEmails := make([]string, 0, len(recordsByEmail))
	successIds := make([]int, 0, len(recordsByEmail))
	failedEmails := make([]string, 0, len(recordsByEmail))
	successSubIDs := make([]string, 0, len(recordsByEmail))
	for email, rec := range recordsByEmail {
		if _, skipped := skippedReasons[email]; skipped {
			failedEmails = append(failedEmails, email)
			continue
		}
		successEmails = append(successEmails, email)
		successIds = append(successIds, rec.Id)
		successSubIDs = append(successSubIDs, rec.SubID)
	}
	withdrawClientTombstones(failedEmails...)

	if len(successIds) > 0 {
		// Serialize the row cleanup against the traffic poll to avoid the
		// cross-transaction lock-order deadlock on client_traffics/inbounds.
		if err := runSerializedTx(func(tx *gorm.DB) error {
			if e := adjustGroupBaselinesForRemovedTraffic(tx, successEmails); e != nil {
				return e
			}
			if e := clearClientHwidsBySubIDTx(tx, successSubIDs...); e != nil {
				return e
			}
			for _, batch := range chunkInts(successIds, sqlInChunk) {
				if e := tx.Where("client_id IN ?", batch).Delete(&model.ClientInbound{}).Error; e != nil {
					return e
				}
				if e := tx.Where("client_id IN ?", batch).Delete(&model.ClientExternalLink{}).Error; e != nil {
					return e
				}
			}
			if !keepTraffic && len(successEmails) > 0 {
				for _, batch := range chunkStrings(successEmails, sqlInChunk) {
					if e := tx.Where("email IN ?", batch).Delete(&xray.ClientTraffic{}).Error; e != nil {
						return e
					}
					if e := tx.Where("client_email IN ?", batch).Delete(&model.InboundClientIps{}).Error; e != nil {
						return e
					}
				}
			}
			for _, batch := range chunkInts(successIds, sqlInChunk) {
				if e := tx.Where("id IN ?", batch).Delete(&model.ClientRecord{}).Error; e != nil {
					return e
				}
			}
			return nil
		}); err != nil {
			withdrawClientTombstones(successEmails...)
			return result, needRestart, err
		}
	}

	result.Deleted = len(successEmails)
	for email, reason := range skippedReasons {
		result.Skipped = append(result.Skipped, BulkDeleteReport{Email: email, Reason: reason})
	}
	return result, needRestart, nil
}

type bulkInboundDeleteResult struct {
	perEmailSkipped map[string]string
	needRestart     bool
}

// bulkDelInboundClients removes multiple clients from a single inbound's
// settings JSON in one read-modify-write cycle, runs the xray runtime
// RemoveUser/DeleteUser calls, and persists the inbound. The returned map
// holds per-email failure reasons; emails not present in the map are
// considered successful for this inbound.
func (s *ClientService) bulkDelInboundClients(
	inboundSvc *InboundService,
	inboundId int,
	emails []string,
	records map[string]*model.ClientRecord,
	keepTraffic bool,
) bulkInboundDeleteResult {
	res := bulkInboundDeleteResult{perEmailSkipped: map[string]string{}}

	defer lockInbound(inboundId).Unlock()

	oldInbound, err := inboundSvc.GetInbound(inboundId)
	if err != nil {
		logger.Error("Load Old Data Error")
		for _, e := range emails {
			res.perEmailSkipped[e] = err.Error()
		}
		return res
	}

	var settings map[string]any
	if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
		for _, e := range emails {
			res.perEmailSkipped[e] = err.Error()
		}
		return res
	}

	// Match by email — the client's stable identity (see Delete). The link-derived
	// set is deletion intent: an email already absent from settings is successful,
	// while foundEmails tracks entries that still need settings-specific cleanup.
	wantedEmails := make(map[string]struct{}, len(emails))
	for _, email := range emails {
		if records[email] == nil {
			res.perEmailSkipped[email] = "client not found"
			continue
		}
		wantedEmails[email] = struct{}{}
	}

	interfaceClients, _ := settings["clients"].([]any)
	newClients := make([]any, 0, len(interfaceClients))
	foundEmails := map[string]bool{}
	enableByEmail := map[string]bool{}
	for _, client := range interfaceClients {
		c, ok := client.(map[string]any)
		if !ok {
			newClients = append(newClients, client)
			continue
		}
		em, _ := c["email"].(string)
		if _, found := wantedEmails[em]; found && em != "" {
			foundEmails[em] = true
			en, _ := c["enable"].(bool)
			enableByEmail[em] = en
			continue
		}
		newClients = append(newClients, client)
	}

	db := database.GetDB()
	newClients = compactOrphans(db, newClients)
	if newClients == nil {
		newClients = []any{}
	}
	settings["clients"] = newClients
	newSettings, err := json.MarshalIndent(settings, "", "  ")
	if err != nil {
		for email := range wantedEmails {
			if _, skip := res.perEmailSkipped[email]; !skip {
				res.perEmailSkipped[email] = err.Error()
			}
		}
		return res
	}
	oldInbound.Settings = string(newSettings)

	foundList := make([]string, 0, len(foundEmails))
	for email := range foundEmails {
		foundList = append(foundList, email)
	}

	notDepletedByEmail := map[string]bool{}
	if len(foundList) > 0 {
		type trafficRow struct {
			Email  string
			Enable bool
		}
		for _, batch := range chunkStrings(foundList, sqlInChunk) {
			var rows []trafficRow
			if err := db.Model(xray.ClientTraffic{}).
				Where("email IN ?", batch).
				Select("email, enable").
				Scan(&rows).Error; err == nil {
				for _, r := range rows {
					notDepletedByEmail[r.Email] = r.Enable
				}
			}
		}
	}

	var sharedSet map[string]bool
	if !keepTraffic {
		var sharedErr error
		sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(foundList, inboundId)
		if sharedErr != nil {
			for email := range wantedEmails {
				res.perEmailSkipped[email] = sharedErr.Error()
			}
			return res
		}
	}
	if !keepTraffic {
		purge := make([]string, 0, len(foundEmails))
		for email := range foundEmails {
			if !sharedSet[strings.ToLower(strings.TrimSpace(email))] {
				purge = append(purge, email)
			}
		}
		if len(purge) > 0 {
			// Serialize the IP/stat purge against the traffic poll to avoid the
			// cross-transaction lock-order deadlock on client_traffics.
			if delErr := runSerializedTx(func(tx *gorm.DB) error {
				if e := inboundSvc.delClientIPsByEmails(tx, purge); e != nil {
					logger.Error("Error in delete client IPs")
					return e
				}
				if e := inboundSvc.delClientStatsByEmails(tx, purge); e != nil {
					logger.Error("Delete stats Data Error")
					return e
				}
				return nil
			}); delErr != nil {
				for _, email := range purge {
					res.perEmailSkipped[email] = delErr.Error()
					delete(foundEmails, email)
				}
			}
		}
	}

	// Serialize against the traffic poll to avoid the cross-transaction
	// lock-order deadlock on inbounds/client_records (runSerializedTx).
	txErr := runSerializedTx(func(tx *gorm.DB) error {
		if err := tx.Save(oldInbound).Error; err != nil {
			return err
		}
		finalClients, err := inboundSvc.GetClients(oldInbound)
		if err != nil {
			return err
		}
		if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
			return err
		}
		if oldInbound.NodeID != nil {
			return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
		}
		return nil
	})
	if txErr != nil {
		for email := range wantedEmails {
			if _, skip := res.perEmailSkipped[email]; !skip {
				res.perEmailSkipped[email] = txErr.Error()
			}
		}
	} else if oldInbound.NodeID == nil {
		rt, rterr := inboundSvc.runtimeFor(oldInbound)
		if rterr != nil {
			res.needRestart = true
		} else {
			for email := range foundEmails {
				if !enableByEmail[email] || !notDepletedByEmail[email] {
					continue
				}
				err1 := rt.RemoveUser(context.Background(), oldInbound, email)
				if err1 == nil {
					logger.Debug("Client deleted on", rt.Name(), ":", email)
				} else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", email)) {
					logger.Debug("User is already deleted. Nothing to do more...")
				} else {
					logger.Debug("Error in deleting client on", rt.Name(), ":", err1)
					res.needRestart = true
				}
			}
		}
	} else {
		dispatchEmails := make([]string, 0, len(wantedEmails))
		for email := range wantedEmails {
			if _, skip := res.perEmailSkipped[email]; !skip {
				dispatchEmails = append(dispatchEmails, email)
			}
		}
		if len(dispatchEmails) > nodeBulkPushThreshold {
			return res
		}
		rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
		if perr != nil {
			logger.Warning("BulkDelete: node runtime lookup after commit failed:", perr)
		} else if push {
			for _, email := range dispatchEmails {
				ctx, cancel := nodePushContext()
				err1 := rt.DeleteClient(ctx, email)
				cancel()
				if err1 != nil {
					logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
					// The node is already dirty, so one reconcile converges the rest of
					// the batch instead of paying another deadline per client.
					break
				}
			}
		}
	}

	return res
}

// BulkCreateResult mirrors BulkAdjustResult for the create flow.
type BulkCreateResult struct {
	Created int                `json:"created"`
	Skipped []BulkCreateReport `json:"skipped,omitempty"`
}

type BulkCreateReport struct {
	Email  string `json:"email"`
	Reason string `json:"reason"`
}

func (s *ClientService) BulkCreate(inboundSvc *InboundService, payloads []ClientCreatePayload) (BulkCreateResult, bool, error) {
	result := BulkCreateResult{}
	if len(payloads) == 0 {
		return result, false, nil
	}

	skip := func(email, reason string) {
		if strings.TrimSpace(email) == "" {
			email = "(missing email)"
		}
		result.Skipped = append(result.Skipped, BulkCreateReport{Email: email, Reason: reason})
	}

	type prepared struct {
		client     model.Client
		inboundIds []int
		limitHwid  int
	}
	prep := make([]prepared, 0, len(payloads))
	emails := make([]string, 0, len(payloads))
	subIDs := make([]string, 0, len(payloads))
	seenEmail := make(map[string]struct{}, len(payloads))
	seenSubID := make(map[string]string, len(payloads))

	for i := range payloads {
		client := payloads[i].Client
		email := strings.TrimSpace(client.Email)
		if email == "" {
			skip("", "client email is required")
			continue
		}
		if verr := validateClientEmail(email); verr != nil {
			skip(email, verr.Error())
			continue
		}
		if verr := validateClientSubID(client.SubID); verr != nil {
			skip(email, verr.Error())
			continue
		}
		if verr := validateClientResetDay(client.ResetDay); verr != nil {
			skip(email, verr.Error())
			continue
		}
		if verr := validateClientResetMax(client.ResetMax); verr != nil {
			skip(email, verr.Error())
			continue
		}
		if verr := validateClientTrafficReset(client.TrafficReset, client.TrafficResetDay); verr != nil {
			skip(email, verr.Error())
			continue
		}
		if len(payloads[i].InboundIds) == 0 {
			skip(email, "at least one inbound is required")
			continue
		}

		client.Email = email
		if client.SubID == "" {
			client.SubID = uuid.NewString()
		}
		if !client.Enable {
			client.Enable = true
		}
		now := time.Now().UnixMilli()
		if client.CreatedAt == 0 {
			client.CreatedAt = now
		}
		client.UpdatedAt = now

		le := strings.ToLower(email)
		if _, dup := seenEmail[le]; dup {
			skip(email, "email already in use: "+email)
			continue
		}
		if owner, ok := seenSubID[client.SubID]; ok && owner != le {
			skip(email, "subId already in use: "+client.SubID)
			continue
		}
		seenEmail[le] = struct{}{}
		seenSubID[client.SubID] = le

		prep = append(prep, prepared{client: client, inboundIds: payloads[i].InboundIds, limitHwid: payloads[i].LimitHwid})
		emails = append(emails, email)
		subIDs = append(subIDs, client.SubID)
	}

	if len(prep) == 0 {
		return result, false, nil
	}

	db := database.GetDB()
	const lookupChunk = 400
	existingByEmail := make(map[string]model.ClientRecord, len(emails))
	for start := 0; start < len(emails); start += lookupChunk {
		end := min(start+lookupChunk, len(emails))
		var rows []model.ClientRecord
		if e := db.Where("email IN ?", emails[start:end]).Find(&rows).Error; e != nil {
			return result, false, e
		}
		for i := range rows {
			existingByEmail[strings.ToLower(rows[i].Email)] = rows[i]
		}
	}
	existingSubOwner := make(map[string]string, len(subIDs))
	for start := 0; start < len(subIDs); start += lookupChunk {
		end := min(start+lookupChunk, len(subIDs))
		var rows []model.ClientRecord
		if e := db.Where("sub_id IN ?", subIDs[start:end]).Find(&rows).Error; e != nil {
			return result, false, e
		}
		for i := range rows {
			existingSubOwner[rows[i].SubID] = strings.ToLower(rows[i].Email)
		}
	}

	inboundCache := make(map[int]*model.Inbound)
	getIb := func(id int) (*model.Inbound, error) {
		if ib, ok := inboundCache[id]; ok {
			return ib, nil
		}
		ib, e := inboundSvc.GetInbound(id)
		if e != nil {
			return nil, e
		}
		inboundCache[id] = ib
		return ib, nil
	}

	byInbound := make(map[int][]model.Client)
	idxByInbound := make(map[int][]int)
	inboundOrder := make([]int, 0)
	failed := make([]bool, len(prep))
	reason := make([]string, len(prep))
	createAnyTunnel := false

	for idx := range prep {
		le := strings.ToLower(prep[idx].client.Email)
		if rec, ok := existingByEmail[le]; ok {
			if rec.SubID != prep[idx].client.SubID {
				failed[idx] = true
				reason[idx] = "email already in use: " + prep[idx].client.Email
				continue
			}
			if prep[idx].client.ID == "" {
				prep[idx].client.ID = rec.UUID
			}
			if prep[idx].client.Password == "" {
				prep[idx].client.Password = rec.Password
			}
			if prep[idx].client.Auth == "" {
				prep[idx].client.Auth = rec.Auth
			}
			if prep[idx].client.Secret == "" {
				prep[idx].client.Secret = rec.Secret
			}
		}
		if owner, ok := existingSubOwner[prep[idx].client.SubID]; ok && owner != le {
			failed[idx] = true
			reason[idx] = "subId already in use: " + prep[idx].client.SubID
			continue
		}

		ok := true
		for _, ibId := range prep[idx].inboundIds {
			ib, e := getIb(ibId)
			if e != nil {
				failed[idx] = true
				reason[idx] = e.Error()
				ok = false
				break
			}
			if ib.Protocol == model.WireGuard || ib.Protocol == model.AmneziaWG {
				createAnyTunnel = true
			}
			if e := s.fillProtocolDefaults(&prep[idx].client, ib); e != nil {
				failed[idx] = true
				reason[idx] = e.Error()
				ok = false
				break
			}
		}
		if !ok {
			continue
		}
		for _, ibId := range prep[idx].inboundIds {
			ib, _ := getIb(ibId)
			if _, seen := byInbound[ibId]; !seen {
				inboundOrder = append(inboundOrder, ibId)
			}
			byInbound[ibId] = append(byInbound[ibId], clientWithInboundFlow(prep[idx].client, ib))
			idxByInbound[ibId] = append(idxByInbound[ibId], idx)
		}
	}

	needRestart := false
	createResults, createPanics := fanoutInboundResults(inboundOrder, addFanoutLimit(createAnyTunnel), func(i int) inboundApplyOutcome {
		ibId := inboundOrder[i]
		payload, e := json.Marshal(map[string][]model.Client{"clients": byInbound[ibId]})
		if e != nil {
			return inboundApplyOutcome{err: e}
		}
		nr, e := s.AddInboundClient(inboundSvc, &model.Inbound{Id: ibId, Settings: string(payload)})
		return inboundApplyOutcome{needRestart: nr, err: e}
	})
	for i, out := range createResults {
		e := out.err
		if createPanics[i] != nil {
			// See BulkAttach: a panicking apply may already have committed.
			needRestart = true
			e = createPanics[i]
		}
		if e != nil {
			for _, idx := range idxByInbound[inboundOrder[i]] {
				failed[idx] = true
				if reason[idx] == "" {
					reason[idx] = e.Error()
				}
			}
			continue
		}
		if out.needRestart {
			needRestart = true
		}
	}

	for idx := range prep {
		if failed[idx] {
			skip(prep[idx].client.Email, reason[idx])
			continue
		}
		if err := s.setClientLimitHwidByEmail(nil, prep[idx].client.Email, prep[idx].limitHwid); err != nil {
			skip(prep[idx].client.Email, err.Error())
			continue
		}
		result.Created++
	}
	return result, needRestart, nil
}

func (s *ClientService) DelDepleted(inboundSvc *InboundService) (int, bool, error) {
	db := database.GetDB()
	now := time.Now().UnixMilli()
	depletedClause := depletedClientsClause

	var rows []xray.ClientTraffic
	if err := db.Where(depletedClause, now).Find(&rows).Error; err != nil {
		return 0, false, err
	}
	if len(rows) == 0 {
		return 0, false, nil
	}

	seen := make(map[string]struct{}, len(rows))
	emails := make([]string, 0, len(rows))
	for _, r := range rows {
		if r.Email == "" {
			continue
		}
		if _, ok := seen[r.Email]; ok {
			continue
		}
		seen[r.Email] = struct{}{}
		emails = append(emails, r.Email)
	}
	if len(emails) == 0 {
		return 0, false, nil
	}

	res, needRestart, err := s.BulkDelete(inboundSvc, emails, false)
	if err != nil {
		return res.Deleted, needRestart, err
	}
	return res.Deleted, needRestart, nil
}

type BulkSetEnableResult struct {
	Changed int                   `json:"changed"`
	Skipped []BulkSetEnableReport `json:"skipped,omitempty"`
}

type BulkSetEnableReport struct {
	Email  string `json:"email"`
	Reason string `json:"reason"`
}

func (s *ClientService) BulkSetEnable(inboundSvc *InboundService, emails []string, enable bool) (BulkSetEnableResult, bool, error) {
	result := BulkSetEnableResult{}

	seen := map[string]struct{}{}
	cleanEmails := make([]string, 0, len(emails))
	for _, e := range emails {
		e = strings.TrimSpace(e)
		if e == "" {
			continue
		}
		if _, ok := seen[e]; ok {
			continue
		}
		seen[e] = struct{}{}
		cleanEmails = append(cleanEmails, e)
	}
	if len(cleanEmails) == 0 {
		return result, false, nil
	}

	db := database.GetDB()

	var records []model.ClientRecord
	for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
		var rows []model.ClientRecord
		if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
			return result, false, err
		}
		records = append(records, rows...)
	}
	recordsByEmail := make(map[string]*model.ClientRecord, len(records))
	for i := range records {
		recordsByEmail[records[i].Email] = &records[i]
	}

	skippedReasons := map[string]string{}
	for _, email := range cleanEmails {
		if _, ok := recordsByEmail[email]; !ok {
			skippedReasons[email] = "client not found"
		}
	}

	clientIds := make([]int, 0, len(recordsByEmail))
	recordIdToEmail := make(map[int]string, len(recordsByEmail))
	for _, r := range recordsByEmail {
		clientIds = append(clientIds, r.Id)
		recordIdToEmail[r.Id] = r.Email
	}

	emailsByInbound := map[int][]string{}
	if len(clientIds) > 0 {
		var mappings []model.ClientInbound
		for _, batch := range chunkInts(clientIds, sqlInChunk) {
			var rows []model.ClientInbound
			if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
				return result, false, err
			}
			mappings = append(mappings, rows...)
		}
		for _, m := range mappings {
			email, ok := recordIdToEmail[m.ClientId]
			if !ok {
				continue
			}
			emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
		}
	}

	needRestart := false
	enableIds := sortedInboundIds(emailsByInbound)
	enableResults, enablePanics := fanoutInboundResults(enableIds, inboundFanoutConcurrency, func(i int) bulkSetEnableInboundResult {
		return s.bulkSetEnableInboundClients(inboundSvc, enableIds[i], emailsByInbound[enableIds[i]], enable)
	})
	for i, ibRes := range enableResults {
		if enablePanics[i] != nil {
			needRestart = true
			for _, email := range emailsByInbound[enableIds[i]] {
				if _, already := skippedReasons[email]; !already {
					skippedReasons[email] = enablePanics[i].Error()
				}
			}
			continue
		}
		if ibRes.needRestart {
			needRestart = true
		}
		for email, reason := range ibRes.perEmailSkipped {
			if _, already := skippedReasons[email]; !already {
				skippedReasons[email] = reason
			}
		}
	}

	successEmails := make([]string, 0, len(recordsByEmail))
	for email := range recordsByEmail {
		if _, skipped := skippedReasons[email]; skipped {
			continue
		}
		successEmails = append(successEmails, email)
	}

	if len(successEmails) > 0 {
		now := time.Now().UnixMilli()
		if err := runSerializedTx(func(tx *gorm.DB) error {
			for _, batch := range chunkStrings(successEmails, sqlInChunk) {
				if e := tx.Model(xray.ClientTraffic{}).Where("email IN ?", batch).Update("enable", enable).Error; e != nil {
					return e
				}
				if e := tx.Model(&model.ClientRecord{}).Where("email IN ?", batch).
					Updates(map[string]any{"enable": enable, "updated_at": now}).Error; e != nil {
					return e
				}
			}
			return nil
		}); err != nil {
			return result, needRestart, err
		}
	}

	result.Changed = len(successEmails)
	for email, reason := range skippedReasons {
		result.Skipped = append(result.Skipped, BulkSetEnableReport{Email: email, Reason: reason})
	}
	return result, needRestart, nil
}

type bulkSetEnableInboundResult struct {
	perEmailSkipped map[string]string
	needRestart     bool
}

func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService, inboundId int, emails []string, enable bool) bulkSetEnableInboundResult {
	res := bulkSetEnableInboundResult{perEmailSkipped: map[string]string{}}

	defer lockInbound(inboundId).Unlock()

	oldInbound, err := inboundSvc.GetInbound(inboundId)
	if err != nil {
		for _, e := range emails {
			res.perEmailSkipped[e] = err.Error()
		}
		return res
	}

	var settings map[string]any
	if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
		for _, e := range emails {
			res.perEmailSkipped[e] = err.Error()
		}
		return res
	}

	wanted := make(map[string]struct{}, len(emails))
	for _, email := range emails {
		wanted[email] = struct{}{}
	}

	cipher := ""
	if oldInbound.Protocol == model.Shadowsocks {
		cipher, _ = settings["method"].(string)
	}

	type changedClient struct {
		email     string
		wasEnable bool
		client    model.Client
	}
	var changed []changedClient
	found := map[string]bool{}
	nowMs := time.Now().UnixMilli()

	interfaceClients, _ := settings["clients"].([]any)
	for i, c := range interfaceClients {
		entry, ok := c.(map[string]any)
		if !ok {
			continue
		}
		email, _ := entry["email"].(string)
		if _, want := wanted[email]; !want || email == "" {
			continue
		}
		found[email] = true
		prev, _ := entry["enable"].(bool)
		if prev == enable {
			continue
		}
		entry["enable"] = enable
		entry["updated_at"] = nowMs
		interfaceClients[i] = entry
		// Build the pushed client from the inbound JSON (the per-inbound source of
		// truth), so a remote UpdateUser carries every field and never zeroes
		// subId/totalGB/expiry from drifting ClientRecord columns (#4628/#4792).
		var parsed model.Client
		if b, mErr := json.Marshal(entry); mErr == nil {
			_ = json.Unmarshal(b, &parsed)
		}
		parsed.Email = email
		parsed.Enable = enable
		changed = append(changed, changedClient{email: email, wasEnable: prev, client: parsed})
	}

	for email := range wanted {
		if !found[email] {
			res.perEmailSkipped[email] = "Client Not Found In Inbound"
		}
	}

	if len(changed) == 0 {
		return res
	}

	settings["clients"] = interfaceClients
	newSettings, err := json.MarshalIndent(settings, "", "  ")
	if err != nil {
		for _, ch := range changed {
			res.perEmailSkipped[ch.email] = err.Error()
		}
		return res
	}
	prevSettings := oldInbound.Settings
	oldInbound.Settings = string(newSettings)

	rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
	if perr != nil {
		for _, ch := range changed {
			res.perEmailSkipped[ch.email] = perr.Error()
		}
		return res
	}
	if oldInbound.NodeID != nil && push && len(changed) > nodeBulkPushThreshold {
		push = false
	}

	txErr := runSerializedTx(func(tx *gorm.DB) error {
		if e := tx.Save(oldInbound).Error; e != nil {
			return e
		}
		finalClients, gcErr := inboundSvc.GetClients(oldInbound)
		if gcErr != nil {
			return gcErr
		}
		if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
			return err
		}
		if oldInbound.NodeID != nil {
			return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
		}
		return nil
	})
	if txErr != nil {
		for _, ch := range changed {
			res.perEmailSkipped[ch.email] = txErr.Error()
		}
		return res
	}

	if oldInbound.NodeID == nil {
		if !push {
			res.needRestart = true
		} else {
			for _, ch := range changed {
				if enable {
					err1 := rt.AddUser(context.Background(), oldInbound, map[string]any{
						"email":    ch.client.Email,
						"id":       ch.client.ID,
						"security": ch.client.Security,
						"flow":     ch.client.Flow,
						"auth":     ch.client.Auth,
						"password": ch.client.Password,
						"cipher":   cipher,
					})
					if err1 != nil {
						logger.Debug("Error in adding client on", rt.Name(), ":", err1)
						res.needRestart = true
					}
				} else if ch.wasEnable {
					err1 := rt.RemoveUser(context.Background(), oldInbound, ch.email)
					if err1 != nil && !strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", ch.email)) {
						logger.Debug("Error in removing client on", rt.Name(), ":", err1)
						res.needRestart = true
					}
				}
			}
		}
	} else if push {
		pushFailed := false
		for _, ch := range changed {
			updated := ch.client
			updated.UpdatedAt = nowMs
			ctx, cancel := nodePushContext()
			err1 := rt.UpdateUser(ctx, oldInbound, ch.email, updated)
			cancel()
			if err1 != nil {
				logger.Warning("Error in updating client on", rt.Name(), ":", err1)
				pushFailed = true
				// First failure ends the batch push; the reconcile converges the rest.
				break
			}
		}
		if !pushFailed {
			advancePushedInbound(rt, prevSettings, oldInbound)
		}
	}

	return res
}