feat(clients): support setting HWID limit and MTProto ad-tag in bulk adjust (#6399)

* feat(clients): support setting HWID limit and MTProto ad-tag in bulk adjust Add HWID device limit and Telegram MTProto sponsor channel (ad-tag) support to the bulk client adjustment flow in both the panel API and frontend ClientBulkAdjustModal. Co-Authored-By: Claude Code <noreply@anthropic.com> * fix(clients): gate adTag to MTProto inbounds and avoid inbound rewrite for limitHwid Co-Authored-By: Claude Code <noreply@anthropic.com> * fix(clients): stamp updated_at only on the clients a bulk adjust changed The updated_at write was gated on hasInboundChanges, which accumulates over the whole inbound instead of describing the client in hand. Once any client in the settings array changed, every client after it was re-stamped as well, so whether an untouched client kept its own updated_at depended on its position in the array. That field feeds node-snapshot conflict resolution, where a spurious bump lets a stale snapshot value win over the stored record. Track the change per client and fold it into the inbound-level flag where the stamp is written, so the early return still skips a save whose settings JSON would be unchanged. Also condenses the BulkAdjust doc comment back to the two-line maximum. * docs(api): regenerate the bulkAdjust reference for limitHwid and adTag frontend/public/openapi.json was copied to docs/public/, but pnpm gen:api was never re-run, so the API reference page's heading, anchor id and search index still described bulkAdjust without limitHwid or adTag. docs-ci.yml fires only on docs/**, and that path had been touched, so nothing flagged the stale MDX. The externalLinks hunks are the generator rewrapping lines main had left stale, not a content change. * fix(i18n): stop enumerating fields in the bulk-adjust empty-form message bulkAdjustNothing listed the fields the form accepts, so it went stale every time one was added: only en-US ever gained "flow", leaving the other twelve locales describing days and traffic alone, and limitHwid and adTag would have repeated that. Say that one field is required instead of naming which, so the message cannot drift again.
2026-09-16 15:17:14 +00:00 · 2026-09-10 17:54:48 +03:30
parent 1456658028
commit ed5465d0f2
29 changed files with 626 additions and 121 deletions
@@ -312,11 +312,13 @@ var bulkFlowAllowed = map[string]struct{}{
 // 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.
+// flow sets the XTLS flow, limitHwid the max registered devices (0 = unlimited)
+// and adTag the MTProto sponsor channel; "none" clears flow or adTag.
 //
 // 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) {
+func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string, addDays int, addBytes int64, flow string, limitHwid *int, adTag string) (BulkAdjustResult, bool, error) {
 	result := BulkAdjustResult{}
 	if len(emails) == 0 {
 		return result, false, nil
@@ -325,8 +327,18 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 	if _, ok := bulkFlowAllowed[flow]; !ok {
 		flow = "" // ignore unknown directives — "" means "leave flow untouched"
 	}
+	adTag = strings.TrimSpace(adTag)
+	if adTag != "" && adTag != bulkFlowClear && !model.ValidMtprotoAdTag(adTag) {
+		return result, false, common.NewError("mtproto client ad tag must be 32 hex characters")
+	}
+	if limitHwid != nil && *limitHwid < 0 {
+		zero := 0
+		limitHwid = &zero
+	}
 	adjustFlow := flow != ""
-	if addDays == 0 && addBytes == 0 && !adjustFlow {
+	adjustHwid := limitHwid != nil
+	adjustAdTag := adTag != ""
+	if addDays == 0 && addBytes == 0 && !adjustFlow && !adjustHwid && !adjustAdTag {
 		return result, false, common.NewError("no adjustment specified")
 	}

@@ -419,7 +431,7 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 				}
 			}
 		}
-		if entry.applyExpiry || entry.applyTotal || adjustFlow {
+		if entry.applyExpiry || entry.applyTotal || adjustFlow || adjustHwid || adjustAdTag {
 			plan[email] = entry
 		}
 	}
@@ -434,8 +446,10 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 	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
+		if entry.applyExpiry || entry.applyTotal || adjustFlow || adjustAdTag {
+			plannedIds = append(plannedIds, entry.record.Id)
+			recordIdToEmail[entry.record.Id] = email
+		}
 	}

 	var mappings []model.ClientInbound
@@ -458,10 +472,12 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 	needRestart := false
 	flowHonored := map[string]bool{}
 	flowIneligible := map[string]bool{}
+	adTagHonored := map[string]bool{}
+	adTagIneligible := 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)
+		return s.bulkAdjustInboundClients(inboundSvc, adjustIds[i], emailsByInbound[adjustIds[i]], plan, flow, adTag)
 	})
 	for i, ibRes := range adjustResults {
 		if adjustPanics[i] != nil {
@@ -483,6 +499,12 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 		for email := range ibRes.flowIneligible {
 			flowIneligible[email] = true
 		}
+		for email := range ibRes.adTagHonored {
+			adTagHonored[email] = true
+		}
+		for email := range ibRes.adTagIneligible {
+			adTagIneligible[email] = true
+		}
 		for email, reason := range ibRes.perEmailSkipped {
 			execFailed[email] = true
 			if _, already := skippedReasons[email]; !already {
@@ -511,6 +533,11 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 		}
 	}

+	wantAdTag := ""
+	if adjustAdTag && adTag != bulkFlowClear {
+		wantAdTag = strings.ToLower(adTag)
+	}
+
 	adjusted := map[string]struct{}{}
 	for email, entry := range plan {
 		if execFailed[email] {
@@ -531,9 +558,24 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 				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] {
+		if adjustHwid {
+			if err := s.setClientLimitHwidByEmail(db, email, *limitHwid); err != nil {
+				if _, already := skippedReasons[email]; !already {
+					skippedReasons[email] = err.Error()
+				}
+				continue
+			}
+		}
+		if adjustAdTag && adTagHonored[email] {
+			if err := db.Model(&model.ClientRecord{}).Where("email = ?", email).UpdateColumn("ad_tag", wantAdTag).Error; err != nil {
+				if _, already := skippedReasons[email]; !already {
+					skippedReasons[email] = err.Error()
+				}
+				continue
+			}
+		}
+		// Counted when expiry/total changed, flow was honored, adTag was honored, or limitHwid was adjusted.
+		if len(updates) > 0 || flowHonored[email] || adTagHonored[email] || adjustHwid {
 			adjusted[email] = struct{}{}
 		}
 	}
@@ -554,6 +596,15 @@ func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string,
 		}
 		result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: "flow not supported on inbound"})
 	}
+	for email := range adTagIneligible {
+		if adTagHonored[email] {
+			continue
+		}
+		if _, already := skippedReasons[email]; already {
+			continue
+		}
+		result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: "adTag not supported on inbound"})
+	}

 	if len(wasDisabledDepleted) > 0 {
 		stillDepleted := map[string]struct{}{}
@@ -599,8 +650,10 @@ type bulkInboundAdjustResult struct {
 	// 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
+	flowIneligible  map[string]bool
+	adTagHonored    map[string]bool
+	adTagIneligible map[string]bool
+	needRestart     bool
 }

 // bulkAdjustInboundClients applies expiry/total deltas to multiple clients
@@ -615,8 +668,15 @@ func (s *ClientService) bulkAdjustInboundClients(
 	emails []string,
 	plan map[string]*bulkAdjustEntry,
 	flow string,
+	adTag string,
 ) bulkInboundAdjustResult {
-	res := bulkInboundAdjustResult{perEmailSkipped: map[string]string{}, flowHonored: map[string]bool{}, flowIneligible: map[string]bool{}}
+	res := bulkInboundAdjustResult{
+		perEmailSkipped: map[string]string{},
+		flowHonored:     map[string]bool{},
+		flowIneligible:  map[string]bool{},
+		adTagHonored:    map[string]bool{},
+		adTagIneligible: map[string]bool{},
+	}

 	defer lockInbound(inboundId).Unlock()

@@ -655,9 +715,16 @@ func (s *ClientService) bulkAdjustInboundClients(
 		(!oldInbound.DisableFlow &&
 			inboundCanEnableTlsFlow(string(oldInbound.Protocol), oldInbound.StreamSettings, oldInbound.Settings))

+	wantAdTag := ""
+	if adTag != "" && adTag != bulkFlowClear {
+		wantAdTag = strings.ToLower(adTag)
+	}
+
 	interfaceClients, _ := settings["clients"].([]any)
 	foundEmails := map[string]bool{}
 	flowChanged := false
+	adTagChanged := false
+	hasInboundChanges := false
 	nowMs := time.Now().Unix() * 1000
 	for i, client := range interfaceClients {
 		c, ok := client.(map[string]any)
@@ -668,12 +735,15 @@ func (s *ClientService) bulkAdjustInboundClients(
 		if _, want := wantedEmails[targetEmail]; !want || targetEmail == "" {
 			continue
 		}
+		clientChanged := false
 		entry := plan[targetEmail]
 		if entry.applyExpiry {
 			c["expiryTime"] = entry.newExpiry
+			clientChanged = true
 		}
 		if entry.applyTotal {
 			c["totalGB"] = entry.newTotal
+			clientChanged = true
 		}
 		if flow != "" {
 			if flowEligible {
@@ -686,13 +756,29 @@ func (s *ClientService) bulkAdjustInboundClients(
 					flowChanged = true
 				}
 				res.flowHonored[targetEmail] = true
+				clientChanged = 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
+		if adTag != "" {
+			if oldInbound.Protocol == model.MTProto {
+				if cur, _ := c["adTag"].(string); cur != wantAdTag {
+					c["adTag"] = wantAdTag
+					adTagChanged = true
+				}
+				res.adTagHonored[targetEmail] = true
+				clientChanged = true
+			} else {
+				res.adTagIneligible[targetEmail] = true
+			}
+		}
+		if clientChanged {
+			c["updated_at"] = nowMs
+			hasInboundChanges = true
+		}
 		interfaceClients[i] = c
 		foundEmails[targetEmail] = true
 	}
@@ -703,7 +789,7 @@ func (s *ClientService) bulkAdjustInboundClients(
 		}
 	}

-	if len(foundEmails) == 0 {
+	if len(foundEmails) == 0 || !hasInboundChanges {
 		return res
 	}

@@ -748,28 +834,36 @@ func (s *ClientService) bulkAdjustInboundClients(
 				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
+	} else {
+		if adTagChanged && oldInbound.Protocol == model.MTProto && oldInbound.NodeID == nil {
+			inboundSvc.applyLocalMtproto(oldInbound.Id)
+		}
+		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
+					}
+					if adTag != "" && oldInbound.Protocol == model.MTProto {
+						updated.AdTag = wantAdTag
+					}
+					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
+					}
 				}
 			}
 		}