client_bulk.go 47 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679
  1. package service
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. "strings"
  7. "time"
  8. "github.com/google/uuid"
  9. "github.com/mhsanaei/3x-ui/v3/internal/database"
  10. "github.com/mhsanaei/3x-ui/v3/internal/database/model"
  11. "github.com/mhsanaei/3x-ui/v3/internal/logger"
  12. "github.com/mhsanaei/3x-ui/v3/internal/util/common"
  13. "github.com/mhsanaei/3x-ui/v3/internal/xray"
  14. "gorm.io/gorm"
  15. )
  16. // BulkAttachResult reports the outcome of a bulk attach across target inbounds.
  17. type BulkAttachResult struct {
  18. Attached []string `json:"attached"`
  19. Skipped []string `json:"skipped"`
  20. Errors []string `json:"errors"`
  21. }
  22. // BulkAttach attaches the given existing clients (by email) to each target inbound,
  23. // reusing their identity (email/UUID/password/subId) and a shared traffic row. It adds
  24. // all clients to a target in a single AddInboundClient call, and reports clients already
  25. // present on a target as skipped.
  26. func (s *ClientService) BulkAttach(inboundSvc *InboundService, emails []string, inboundIds []int) (*BulkAttachResult, bool, error) {
  27. result := &BulkAttachResult{}
  28. if len(emails) == 0 || len(inboundIds) == 0 {
  29. return result, false, nil
  30. }
  31. recordErr := func(format string, args ...any) {
  32. msg := fmt.Sprintf(format, args...)
  33. result.Errors = append(result.Errors, msg)
  34. logger.Warningf("[BulkAttach] %s", msg)
  35. }
  36. records := make([]*model.ClientRecord, 0, len(emails))
  37. seenEmail := make(map[string]struct{}, len(emails))
  38. for _, email := range emails {
  39. if email == "" {
  40. continue
  41. }
  42. key := strings.ToLower(email)
  43. if _, ok := seenEmail[key]; ok {
  44. continue
  45. }
  46. seenEmail[key] = struct{}{}
  47. rec, err := s.GetRecordByEmail(nil, email)
  48. if err != nil {
  49. recordErr("%s: %v", email, err)
  50. continue
  51. }
  52. records = append(records, rec)
  53. }
  54. emailSubIDs, sidErr := inboundSvc.getAllEmailSubIDs()
  55. if sidErr != nil {
  56. emailSubIDs = nil
  57. logger.Warningf("[BulkAttach] getAllEmailSubIDs: %v", sidErr)
  58. }
  59. needRestart := false
  60. for _, ibId := range inboundIds {
  61. inbound, err := inboundSvc.GetInbound(ibId)
  62. if err != nil {
  63. recordErr("inbound %d: %v", ibId, err)
  64. continue
  65. }
  66. existingClients, err := inboundSvc.GetClients(inbound)
  67. if err != nil {
  68. recordErr("inbound %d: %v", ibId, err)
  69. continue
  70. }
  71. have := make(map[string]struct{}, len(existingClients))
  72. for _, c := range existingClients {
  73. have[strings.ToLower(c.Email)] = struct{}{}
  74. }
  75. clientsToAdd := make([]model.Client, 0, len(records))
  76. for _, rec := range records {
  77. if _, attached := have[strings.ToLower(rec.Email)]; attached {
  78. result.Skipped = append(result.Skipped, rec.Email)
  79. continue
  80. }
  81. client := *rec.ToClient()
  82. client.UpdatedAt = time.Now().UnixMilli()
  83. if err := s.fillProtocolDefaults(&client, inbound); err != nil {
  84. recordErr("%s -> inbound %d: %v", rec.Email, ibId, err)
  85. continue
  86. }
  87. clientsToAdd = append(clientsToAdd, clientWithInboundFlow(client, inbound))
  88. }
  89. if len(clientsToAdd) == 0 {
  90. continue
  91. }
  92. payload, err := json.Marshal(map[string][]model.Client{"clients": clientsToAdd})
  93. if err != nil {
  94. recordErr("inbound %d: %v", ibId, err)
  95. continue
  96. }
  97. nr, err := s.addInboundClient(inboundSvc, &model.Inbound{Id: ibId, Settings: string(payload)}, emailSubIDs)
  98. if err != nil {
  99. recordErr("inbound %d: %v", ibId, err)
  100. continue
  101. }
  102. if nr {
  103. needRestart = true
  104. }
  105. for _, c := range clientsToAdd {
  106. result.Attached = append(result.Attached, c.Email)
  107. }
  108. }
  109. return result, needRestart, nil
  110. }
  111. // BulkDetachResult reports the outcome of a bulk detach across target inbounds.
  112. type BulkDetachResult struct {
  113. Detached []string `json:"detached"`
  114. Skipped []string `json:"skipped"`
  115. Errors []string `json:"errors"`
  116. }
  117. // BulkDetach detaches the given existing clients (by email) from each target inbound.
  118. // (email, inbound) pairs where the client is not currently attached are silently skipped
  119. // at the inbound level; emails that aren't attached to any of the requested inbounds
  120. // are reported under skipped. ClientRecord rows are kept even when they become orphaned
  121. // (matches single-client detach semantics); callers should use bulkDelete for full removal.
  122. func (s *ClientService) BulkDetach(inboundSvc *InboundService, emails []string, inboundIds []int) (*BulkDetachResult, bool, error) {
  123. result := &BulkDetachResult{}
  124. if len(emails) == 0 || len(inboundIds) == 0 {
  125. return result, false, nil
  126. }
  127. recordErr := func(format string, args ...any) {
  128. msg := fmt.Sprintf(format, args...)
  129. result.Errors = append(result.Errors, msg)
  130. logger.Warningf("[BulkDetach] %s", msg)
  131. }
  132. requested := make(map[int]struct{}, len(inboundIds))
  133. for _, id := range inboundIds {
  134. requested[id] = struct{}{}
  135. }
  136. recsByInbound := make(map[int][]*model.ClientRecord)
  137. emailOrder := make([]string, 0, len(emails))
  138. emailRepr := make(map[string]string, len(emails))
  139. emailFailed := make(map[string]bool, len(emails))
  140. seenEmail := make(map[string]struct{}, len(emails))
  141. for _, email := range emails {
  142. if email == "" {
  143. continue
  144. }
  145. key := strings.ToLower(email)
  146. if _, ok := seenEmail[key]; ok {
  147. continue
  148. }
  149. seenEmail[key] = struct{}{}
  150. rec, err := s.GetRecordByEmail(nil, email)
  151. if err != nil {
  152. recordErr("%s: %v", email, err)
  153. continue
  154. }
  155. currentIds, err := s.GetInboundIdsForRecord(rec.Id)
  156. if err != nil {
  157. recordErr("%s: %v", email, err)
  158. continue
  159. }
  160. matched := false
  161. for _, id := range currentIds {
  162. if _, ok := requested[id]; ok {
  163. recsByInbound[id] = append(recsByInbound[id], rec)
  164. matched = true
  165. }
  166. }
  167. if !matched {
  168. result.Skipped = append(result.Skipped, rec.Email)
  169. continue
  170. }
  171. emailOrder = append(emailOrder, key)
  172. emailRepr[key] = rec.Email
  173. }
  174. needRestart := false
  175. for _, ibId := range inboundIds {
  176. recs, ok := recsByInbound[ibId]
  177. if !ok {
  178. continue
  179. }
  180. delete(recsByInbound, ibId)
  181. nr, err := s.delInboundClients(inboundSvc, ibId, recs, true)
  182. if err != nil {
  183. recordErr("inbound %d: %v", ibId, err)
  184. for _, rec := range recs {
  185. emailFailed[strings.ToLower(rec.Email)] = true
  186. }
  187. continue
  188. }
  189. if nr {
  190. needRestart = true
  191. }
  192. }
  193. for _, key := range emailOrder {
  194. if emailFailed[key] {
  195. continue
  196. }
  197. result.Detached = append(result.Detached, emailRepr[key])
  198. }
  199. return result, needRestart, nil
  200. }
  201. // BulkAdjustResult is returned by BulkAdjust to report how many clients were
  202. // successfully updated and which were skipped (typically because the field
  203. // being adjusted was unlimited for that client) or failed.
  204. type BulkAdjustResult struct {
  205. Adjusted int `json:"adjusted"`
  206. Skipped []BulkAdjustReport `json:"skipped,omitempty"`
  207. }
  208. type BulkAdjustReport struct {
  209. Email string `json:"email"`
  210. Reason string `json:"reason"`
  211. }
  212. type bulkAdjustEntry struct {
  213. record *model.ClientRecord
  214. applyExpiry bool
  215. newExpiry int64
  216. applyTotal bool
  217. newTotal int64
  218. }
  219. // bulkFlowClear is the directive that strips the XTLS flow from every selected
  220. // client. The vision values are the only positive flows xray accepts.
  221. const bulkFlowClear = "none"
  222. // bulkFlowAllowed whitelists the flow directives BulkAdjust accepts. Anything
  223. // outside this set is treated as "" (leave flow untouched) so a malformed or
  224. // hostile value can never be injected into a client's settings. The dropdown in
  225. // ClientBulkAdjustModal.tsx offers the same set ("" / "none" / TLS_FLOW_CONTROL);
  226. // keep the two in sync.
  227. var bulkFlowAllowed = map[string]struct{}{
  228. "": {},
  229. bulkFlowClear: {},
  230. "xtls-rprx-vision": {},
  231. "xtls-rprx-vision-udp443": {},
  232. }
  233. // BulkAdjust shifts ExpiryTime by addDays (days) and TotalGB by addBytes
  234. // for every email in the list. Clients whose corresponding field is
  235. // unlimited (0) are skipped — bulk extend should not accidentally
  236. // limit an unlimited client. addDays and addBytes may be negative.
  237. //
  238. // Like BulkDelete, the work is grouped by inbound so each inbound's
  239. // settings JSON is parsed and written exactly once regardless of how
  240. // many target emails it contains.
  241. func (s *ClientService) BulkAdjust(inboundSvc *InboundService, emails []string, addDays int, addBytes int64, flow string) (BulkAdjustResult, bool, error) {
  242. result := BulkAdjustResult{}
  243. if len(emails) == 0 {
  244. return result, false, nil
  245. }
  246. flow = strings.TrimSpace(flow)
  247. if _, ok := bulkFlowAllowed[flow]; !ok {
  248. flow = "" // ignore unknown directives — "" means "leave flow untouched"
  249. }
  250. adjustFlow := flow != ""
  251. if addDays == 0 && addBytes == 0 && !adjustFlow {
  252. return result, false, common.NewError("no adjustment specified")
  253. }
  254. addExpiryMs := int64(addDays) * 24 * 60 * 60 * 1000
  255. seen := map[string]struct{}{}
  256. cleanEmails := make([]string, 0, len(emails))
  257. for _, e := range emails {
  258. e = strings.TrimSpace(e)
  259. if e == "" {
  260. continue
  261. }
  262. if _, ok := seen[e]; ok {
  263. continue
  264. }
  265. seen[e] = struct{}{}
  266. cleanEmails = append(cleanEmails, e)
  267. }
  268. if len(cleanEmails) == 0 {
  269. return result, false, nil
  270. }
  271. db := database.GetDB()
  272. var records []model.ClientRecord
  273. for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
  274. var rows []model.ClientRecord
  275. if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
  276. return result, false, err
  277. }
  278. records = append(records, rows...)
  279. }
  280. recordsByEmail := make(map[string]*model.ClientRecord, len(records))
  281. for i := range records {
  282. recordsByEmail[records[i].Email] = &records[i]
  283. }
  284. skippedReasons := map[string]string{}
  285. for _, email := range cleanEmails {
  286. if _, ok := recordsByEmail[email]; !ok {
  287. skippedReasons[email] = "client not found"
  288. }
  289. }
  290. plan := map[string]*bulkAdjustEntry{}
  291. for email, rec := range recordsByEmail {
  292. entry := &bulkAdjustEntry{record: rec}
  293. if addDays != 0 {
  294. switch {
  295. case rec.ExpiryTime == 0:
  296. if _, exists := skippedReasons[email]; !exists {
  297. skippedReasons[email] = "unlimited expiry"
  298. }
  299. case rec.ExpiryTime > 0:
  300. next := rec.ExpiryTime + addExpiryMs
  301. if next <= 0 {
  302. if _, exists := skippedReasons[email]; !exists {
  303. skippedReasons[email] = "reduction exceeds remaining time"
  304. }
  305. } else {
  306. entry.applyExpiry = true
  307. entry.newExpiry = next
  308. }
  309. default:
  310. next := rec.ExpiryTime - addExpiryMs
  311. if next >= 0 {
  312. if _, exists := skippedReasons[email]; !exists {
  313. skippedReasons[email] = "reduction exceeds delay window"
  314. }
  315. } else {
  316. entry.applyExpiry = true
  317. entry.newExpiry = next
  318. }
  319. }
  320. }
  321. if addBytes != 0 {
  322. if rec.TotalGB == 0 {
  323. if _, exists := skippedReasons[email]; !exists {
  324. skippedReasons[email] = "unlimited traffic"
  325. }
  326. } else {
  327. next := rec.TotalGB + addBytes
  328. if next <= 0 {
  329. if _, exists := skippedReasons[email]; !exists {
  330. skippedReasons[email] = "reduction exceeds remaining quota"
  331. }
  332. } else {
  333. entry.applyTotal = true
  334. entry.newTotal = next
  335. }
  336. }
  337. }
  338. if entry.applyExpiry || entry.applyTotal || adjustFlow {
  339. plan[email] = entry
  340. }
  341. }
  342. if len(plan) == 0 {
  343. for email, reason := range skippedReasons {
  344. result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: reason})
  345. }
  346. return result, false, nil
  347. }
  348. plannedIds := make([]int, 0, len(plan))
  349. recordIdToEmail := make(map[int]string, len(plan))
  350. for email, entry := range plan {
  351. plannedIds = append(plannedIds, entry.record.Id)
  352. recordIdToEmail[entry.record.Id] = email
  353. }
  354. var mappings []model.ClientInbound
  355. for _, batch := range chunkInts(plannedIds, sqlInChunk) {
  356. var rows []model.ClientInbound
  357. if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
  358. return result, false, err
  359. }
  360. mappings = append(mappings, rows...)
  361. }
  362. emailsByInbound := map[int][]string{}
  363. for _, m := range mappings {
  364. email, ok := recordIdToEmail[m.ClientId]
  365. if !ok {
  366. continue
  367. }
  368. emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
  369. }
  370. needRestart := false
  371. flowHonored := map[string]bool{}
  372. flowIneligible := map[string]bool{}
  373. for inboundId, ibEmails := range emailsByInbound {
  374. ibRes := s.bulkAdjustInboundClients(inboundSvc, inboundId, ibEmails, plan, flow)
  375. if ibRes.needRestart {
  376. needRestart = true
  377. }
  378. for email := range ibRes.flowHonored {
  379. flowHonored[email] = true
  380. }
  381. for email := range ibRes.flowIneligible {
  382. flowIneligible[email] = true
  383. }
  384. for email, reason := range ibRes.perEmailSkipped {
  385. if _, already := skippedReasons[email]; !already {
  386. skippedReasons[email] = reason
  387. }
  388. }
  389. }
  390. now := time.Now().Unix() * 1000
  391. cond := depletedCond(db)
  392. candidateEmails := make([]string, 0, len(plan))
  393. for email, entry := range plan {
  394. if entry.applyExpiry || entry.applyTotal {
  395. candidateEmails = append(candidateEmails, email)
  396. }
  397. }
  398. wasDisabledDepleted := map[string]struct{}{}
  399. for _, batch := range chunkStrings(candidateEmails, sqlInChunk) {
  400. var rows []string
  401. if err := db.Model(xray.ClientTraffic{}).
  402. Where(cond+" AND enable = ? AND email IN ?", now, false, batch).
  403. Pluck("email", &rows).Error; err != nil {
  404. return result, needRestart, err
  405. }
  406. for _, e := range rows {
  407. wasDisabledDepleted[e] = struct{}{}
  408. }
  409. }
  410. adjusted := map[string]struct{}{}
  411. for email, entry := range plan {
  412. if _, skipped := skippedReasons[email]; skipped {
  413. continue
  414. }
  415. updates := map[string]any{}
  416. if entry.applyExpiry {
  417. updates["expiry_time"] = entry.newExpiry
  418. }
  419. if entry.applyTotal {
  420. updates["total"] = entry.newTotal
  421. }
  422. if len(updates) > 0 {
  423. if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Updates(updates).Error; err != nil {
  424. if _, already := skippedReasons[email]; !already {
  425. skippedReasons[email] = err.Error()
  426. }
  427. continue
  428. }
  429. }
  430. // Counted when expiry/total changed, or a flow directive was honored
  431. // for this client (flow lives in the inbound JSON, not ClientTraffic).
  432. if len(updates) > 0 || flowHonored[email] {
  433. adjusted[email] = struct{}{}
  434. }
  435. }
  436. result.Adjusted = len(adjusted)
  437. for email, reason := range skippedReasons {
  438. result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: reason})
  439. }
  440. // Report a flow directive that no inbound could carry — only when it was not
  441. // honored anywhere and the client has no other (expiry/total) skip reason.
  442. // The expiry/total part, if any, has already been applied and counted above.
  443. for email := range flowIneligible {
  444. if flowHonored[email] {
  445. continue
  446. }
  447. if _, already := skippedReasons[email]; already {
  448. continue
  449. }
  450. result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: "flow not supported on inbound"})
  451. }
  452. if len(wasDisabledDepleted) > 0 {
  453. stillDepleted := map[string]struct{}{}
  454. wasList := make([]string, 0, len(wasDisabledDepleted))
  455. for e := range wasDisabledDepleted {
  456. wasList = append(wasList, e)
  457. }
  458. for _, batch := range chunkStrings(wasList, sqlInChunk) {
  459. var rows []string
  460. if err := db.Model(xray.ClientTraffic{}).
  461. Where(cond+" AND email IN ?", now, batch).
  462. Pluck("email", &rows).Error; err != nil {
  463. return result, needRestart, err
  464. }
  465. for _, e := range rows {
  466. stillDepleted[e] = struct{}{}
  467. }
  468. }
  469. reEnable := make([]string, 0, len(wasDisabledDepleted))
  470. for e := range wasDisabledDepleted {
  471. if _, still := stillDepleted[e]; !still {
  472. reEnable = append(reEnable, e)
  473. }
  474. }
  475. if len(reEnable) > 0 {
  476. _, nr, err := s.BulkSetEnable(inboundSvc, reEnable, true)
  477. if err != nil {
  478. return result, needRestart, err
  479. }
  480. if nr {
  481. needRestart = true
  482. }
  483. }
  484. }
  485. return result, needRestart, nil
  486. }
  487. type bulkInboundAdjustResult struct {
  488. perEmailSkipped map[string]string
  489. flowHonored map[string]bool
  490. // flowIneligible is tracked apart from perEmailSkipped: a flow directive
  491. // that an inbound cannot carry must not suppress the expiry/total write for
  492. // the same client (which would diverge the inbound JSON / ClientRecord from
  493. // ClientTraffic). It only feeds the final Skipped report.
  494. flowIneligible map[string]bool
  495. needRestart bool
  496. }
  497. // bulkAdjustInboundClients applies expiry/total deltas to multiple clients
  498. // inside a single inbound's settings JSON. The xray runtime is updated
  499. // only for remote-node inbounds; local nodes do not need a notification
  500. // because the AddUser payload does not include totalGB/expiryTime —
  501. // changing those fields is identity-preserving and the panel's traffic
  502. // enforcement loop picks up the new limits from ClientTraffic directly.
  503. func (s *ClientService) bulkAdjustInboundClients(
  504. inboundSvc *InboundService,
  505. inboundId int,
  506. emails []string,
  507. plan map[string]*bulkAdjustEntry,
  508. flow string,
  509. ) bulkInboundAdjustResult {
  510. res := bulkInboundAdjustResult{perEmailSkipped: map[string]string{}, flowHonored: map[string]bool{}, flowIneligible: map[string]bool{}}
  511. defer lockInbound(inboundId).Unlock()
  512. oldInbound, err := inboundSvc.GetInbound(inboundId)
  513. if err != nil {
  514. logger.Error("Load Old Data Error")
  515. for _, e := range emails {
  516. res.perEmailSkipped[e] = err.Error()
  517. }
  518. return res
  519. }
  520. var settings map[string]any
  521. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  522. for _, e := range emails {
  523. res.perEmailSkipped[e] = err.Error()
  524. }
  525. return res
  526. }
  527. // Match by email — the client's stable identity (see Delete). Credentials
  528. // can drift from the inbound JSON, so they are never used for matching.
  529. wantedEmails := make(map[string]struct{}, len(emails))
  530. for _, email := range emails {
  531. if plan[email] == nil {
  532. res.perEmailSkipped[email] = "client not found"
  533. continue
  534. }
  535. wantedEmails[email] = struct{}{}
  536. }
  537. // Flow eligibility is a property of the inbound (protocol + transport), so
  538. // resolve it once. Clearing flow is always allowed; setting a vision flow
  539. // is only honored on an inbound that can carry it.
  540. flowEligible := flow == bulkFlowClear ||
  541. inboundCanEnableTlsFlow(string(oldInbound.Protocol), oldInbound.StreamSettings, oldInbound.Settings)
  542. interfaceClients, _ := settings["clients"].([]any)
  543. foundEmails := map[string]bool{}
  544. flowChanged := false
  545. nowMs := time.Now().Unix() * 1000
  546. for i, client := range interfaceClients {
  547. c, ok := client.(map[string]any)
  548. if !ok {
  549. continue
  550. }
  551. targetEmail, _ := c["email"].(string)
  552. if _, want := wantedEmails[targetEmail]; !want || targetEmail == "" {
  553. continue
  554. }
  555. entry := plan[targetEmail]
  556. if entry.applyExpiry {
  557. c["expiryTime"] = entry.newExpiry
  558. }
  559. if entry.applyTotal {
  560. c["totalGB"] = entry.newTotal
  561. }
  562. if flow != "" {
  563. if flowEligible {
  564. want := ""
  565. if flow != bulkFlowClear {
  566. want = flow
  567. }
  568. if cur, _ := c["flow"].(string); cur != want {
  569. c["flow"] = want
  570. flowChanged = true
  571. }
  572. res.flowHonored[targetEmail] = true
  573. } else {
  574. // Record separately so this never suppresses the expiry/total
  575. // write for the same client (see flowIneligible doc).
  576. res.flowIneligible[targetEmail] = true
  577. }
  578. }
  579. c["updated_at"] = nowMs
  580. interfaceClients[i] = c
  581. foundEmails[targetEmail] = true
  582. }
  583. for email := range wantedEmails {
  584. if !foundEmails[email] {
  585. res.perEmailSkipped[email] = "Client Not Found In Inbound"
  586. }
  587. }
  588. if len(foundEmails) == 0 {
  589. return res
  590. }
  591. settings["clients"] = interfaceClients
  592. newSettings, err := json.MarshalIndent(settings, "", " ")
  593. if err != nil {
  594. for email := range foundEmails {
  595. res.perEmailSkipped[email] = err.Error()
  596. }
  597. return res
  598. }
  599. prevSettings := oldInbound.Settings
  600. oldInbound.Settings = string(newSettings)
  601. // A flow change rewrites the user's xray config, which the lightweight
  602. // UpdateUser push below does not carry. Local nodes reload via restart;
  603. // remote nodes get a full reconcile (MarkNodeDirty) instead of a per-user push.
  604. if flowChanged && oldInbound.NodeID == nil {
  605. res.needRestart = true
  606. }
  607. if oldInbound.NodeID != nil {
  608. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  609. if perr != nil {
  610. for email := range foundEmails {
  611. res.perEmailSkipped[email] = perr.Error()
  612. delete(foundEmails, email)
  613. }
  614. } else {
  615. if flowChanged {
  616. push = false
  617. }
  618. // Large batches collapse into one reconcile push rather than M updates.
  619. if push && len(foundEmails) > nodeBulkPushThreshold {
  620. push = false
  621. }
  622. if push {
  623. pushFailed := false
  624. for email := range foundEmails {
  625. entry := plan[email]
  626. updated := *entry.record.ToClient()
  627. if entry.applyExpiry {
  628. updated.ExpiryTime = entry.newExpiry
  629. }
  630. if entry.applyTotal {
  631. updated.TotalGB = entry.newTotal
  632. }
  633. updated.UpdatedAt = nowMs
  634. if err1 := rt.UpdateUser(context.Background(), oldInbound, email, updated); err1 != nil {
  635. logger.Warning("Error in updating client on", rt.Name(), ":", err1)
  636. pushFailed = true
  637. }
  638. }
  639. if !pushFailed {
  640. advancePushedInbound(rt, prevSettings, oldInbound)
  641. }
  642. }
  643. }
  644. }
  645. // Serialize against the traffic poll to avoid the cross-transaction
  646. // lock-order deadlock on inbounds/client_records (runSerializedTx).
  647. txErr := runSerializedTx(func(tx *gorm.DB) error {
  648. if err := tx.Save(oldInbound).Error; err != nil {
  649. return err
  650. }
  651. finalClients, gcErr := inboundSvc.GetClients(oldInbound)
  652. if gcErr != nil {
  653. return gcErr
  654. }
  655. if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
  656. return err
  657. }
  658. if oldInbound.NodeID != nil {
  659. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  660. }
  661. return nil
  662. })
  663. if txErr != nil {
  664. for email := range foundEmails {
  665. if _, skip := res.perEmailSkipped[email]; !skip {
  666. res.perEmailSkipped[email] = txErr.Error()
  667. }
  668. }
  669. }
  670. return res
  671. }
  672. // BulkDeleteResult mirrors BulkAdjustResult: total deleted plus per-email
  673. // skip reasons when an email could not be processed.
  674. type BulkDeleteResult struct {
  675. Deleted int `json:"deleted"`
  676. Skipped []BulkDeleteReport `json:"skipped,omitempty"`
  677. }
  678. type BulkDeleteReport struct {
  679. Email string `json:"email"`
  680. Reason string `json:"reason"`
  681. }
  682. // BulkDelete removes every client in the list in one optimized pass.
  683. // Instead of running the full single-delete pipeline N times (which would
  684. // re-read, re-parse, and re-write each inbound's settings JSON for every
  685. // email), it groups emails by inbound and performs a single
  686. // read-modify-write per inbound. Per-row DB cleanups are also batched with
  687. // IN-clause queries at the end. Errors on a particular email are recorded
  688. // in the Skipped list and processing continues for the rest.
  689. func (s *ClientService) BulkDelete(inboundSvc *InboundService, emails []string, keepTraffic bool) (BulkDeleteResult, bool, error) {
  690. result := BulkDeleteResult{}
  691. seen := map[string]struct{}{}
  692. cleanEmails := make([]string, 0, len(emails))
  693. for _, e := range emails {
  694. e = strings.TrimSpace(e)
  695. if e == "" {
  696. continue
  697. }
  698. if _, ok := seen[e]; ok {
  699. continue
  700. }
  701. seen[e] = struct{}{}
  702. cleanEmails = append(cleanEmails, e)
  703. }
  704. if len(cleanEmails) == 0 {
  705. return result, false, nil
  706. }
  707. db := database.GetDB()
  708. var records []model.ClientRecord
  709. for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
  710. var rows []model.ClientRecord
  711. if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
  712. return result, false, err
  713. }
  714. records = append(records, rows...)
  715. }
  716. recordsByEmail := make(map[string]*model.ClientRecord, len(records))
  717. tombstoneEmails := make([]string, 0, len(records))
  718. for i := range records {
  719. recordsByEmail[records[i].Email] = &records[i]
  720. tombstoneEmails = append(tombstoneEmails, records[i].Email)
  721. }
  722. tombstoneClientEmails(tombstoneEmails)
  723. skippedReasons := map[string]string{}
  724. for _, email := range cleanEmails {
  725. if _, ok := recordsByEmail[email]; !ok {
  726. skippedReasons[email] = "client not found"
  727. }
  728. }
  729. clientIds := make([]int, 0, len(recordsByEmail))
  730. recordIdToEmail := make(map[int]string, len(recordsByEmail))
  731. for _, r := range recordsByEmail {
  732. clientIds = append(clientIds, r.Id)
  733. recordIdToEmail[r.Id] = r.Email
  734. }
  735. emailsByInbound := map[int][]string{}
  736. if len(clientIds) > 0 {
  737. var mappings []model.ClientInbound
  738. for _, batch := range chunkInts(clientIds, sqlInChunk) {
  739. var rows []model.ClientInbound
  740. if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
  741. return result, false, err
  742. }
  743. mappings = append(mappings, rows...)
  744. }
  745. for _, m := range mappings {
  746. email, ok := recordIdToEmail[m.ClientId]
  747. if !ok {
  748. continue
  749. }
  750. emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
  751. }
  752. }
  753. needRestart := false
  754. for inboundId, ibEmails := range emailsByInbound {
  755. ibResult := s.bulkDelInboundClients(inboundSvc, inboundId, ibEmails, recordsByEmail, false)
  756. if ibResult.needRestart {
  757. needRestart = true
  758. }
  759. for email, reason := range ibResult.perEmailSkipped {
  760. if _, already := skippedReasons[email]; !already {
  761. skippedReasons[email] = reason
  762. }
  763. }
  764. }
  765. successEmails := make([]string, 0, len(recordsByEmail))
  766. successIds := make([]int, 0, len(recordsByEmail))
  767. for email, rec := range recordsByEmail {
  768. if _, skipped := skippedReasons[email]; skipped {
  769. continue
  770. }
  771. successEmails = append(successEmails, email)
  772. successIds = append(successIds, rec.Id)
  773. }
  774. if len(successIds) > 0 {
  775. // Serialize the row cleanup against the traffic poll to avoid the
  776. // cross-transaction lock-order deadlock on client_traffics/inbounds.
  777. if err := runSerializedTx(func(tx *gorm.DB) error {
  778. if e := adjustGroupBaselinesForRemovedTraffic(tx, successEmails); e != nil {
  779. return e
  780. }
  781. for _, batch := range chunkInts(successIds, sqlInChunk) {
  782. if e := tx.Where("client_id IN ?", batch).Delete(&model.ClientInbound{}).Error; e != nil {
  783. return e
  784. }
  785. }
  786. if !keepTraffic && len(successEmails) > 0 {
  787. for _, batch := range chunkStrings(successEmails, sqlInChunk) {
  788. if e := tx.Where("email IN ?", batch).Delete(&xray.ClientTraffic{}).Error; e != nil {
  789. return e
  790. }
  791. if e := tx.Where("client_email IN ?", batch).Delete(&model.InboundClientIps{}).Error; e != nil {
  792. return e
  793. }
  794. }
  795. }
  796. for _, batch := range chunkInts(successIds, sqlInChunk) {
  797. if e := tx.Where("id IN ?", batch).Delete(&model.ClientRecord{}).Error; e != nil {
  798. return e
  799. }
  800. }
  801. return nil
  802. }); err != nil {
  803. return result, needRestart, err
  804. }
  805. }
  806. result.Deleted = len(successEmails)
  807. for email, reason := range skippedReasons {
  808. result.Skipped = append(result.Skipped, BulkDeleteReport{Email: email, Reason: reason})
  809. }
  810. return result, needRestart, nil
  811. }
  812. type bulkInboundDeleteResult struct {
  813. perEmailSkipped map[string]string
  814. needRestart bool
  815. }
  816. // bulkDelInboundClients removes multiple clients from a single inbound's
  817. // settings JSON in one read-modify-write cycle, runs the xray runtime
  818. // RemoveUser/DeleteUser calls, and persists the inbound. The returned map
  819. // holds per-email failure reasons; emails not present in the map are
  820. // considered successful for this inbound.
  821. func (s *ClientService) bulkDelInboundClients(
  822. inboundSvc *InboundService,
  823. inboundId int,
  824. emails []string,
  825. records map[string]*model.ClientRecord,
  826. keepTraffic bool,
  827. ) bulkInboundDeleteResult {
  828. res := bulkInboundDeleteResult{perEmailSkipped: map[string]string{}}
  829. defer lockInbound(inboundId).Unlock()
  830. oldInbound, err := inboundSvc.GetInbound(inboundId)
  831. if err != nil {
  832. logger.Error("Load Old Data Error")
  833. for _, e := range emails {
  834. res.perEmailSkipped[e] = err.Error()
  835. }
  836. return res
  837. }
  838. var settings map[string]any
  839. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  840. for _, e := range emails {
  841. res.perEmailSkipped[e] = err.Error()
  842. }
  843. return res
  844. }
  845. // Match by email — the client's stable identity (see Delete). Removes every
  846. // entry carrying a wanted email, independent of credential drift.
  847. wantedEmails := make(map[string]struct{}, len(emails))
  848. for _, email := range emails {
  849. if records[email] == nil {
  850. res.perEmailSkipped[email] = "client not found"
  851. continue
  852. }
  853. wantedEmails[email] = struct{}{}
  854. }
  855. interfaceClients, _ := settings["clients"].([]any)
  856. newClients := make([]any, 0, len(interfaceClients))
  857. foundEmails := map[string]bool{}
  858. enableByEmail := map[string]bool{}
  859. for _, client := range interfaceClients {
  860. c, ok := client.(map[string]any)
  861. if !ok {
  862. newClients = append(newClients, client)
  863. continue
  864. }
  865. em, _ := c["email"].(string)
  866. if _, found := wantedEmails[em]; found && em != "" {
  867. foundEmails[em] = true
  868. en, _ := c["enable"].(bool)
  869. enableByEmail[em] = en
  870. continue
  871. }
  872. newClients = append(newClients, client)
  873. }
  874. for email := range wantedEmails {
  875. if !foundEmails[email] {
  876. res.perEmailSkipped[email] = "Client Not Found In Inbound"
  877. }
  878. }
  879. db := database.GetDB()
  880. newClients = compactOrphans(db, newClients)
  881. if newClients == nil {
  882. newClients = []any{}
  883. }
  884. settings["clients"] = newClients
  885. newSettings, err := json.MarshalIndent(settings, "", " ")
  886. if err != nil {
  887. for email := range foundEmails {
  888. if _, skip := res.perEmailSkipped[email]; !skip {
  889. res.perEmailSkipped[email] = err.Error()
  890. }
  891. }
  892. return res
  893. }
  894. prevSettings := oldInbound.Settings
  895. oldInbound.Settings = string(newSettings)
  896. foundList := make([]string, 0, len(foundEmails))
  897. for email := range foundEmails {
  898. foundList = append(foundList, email)
  899. }
  900. notDepletedByEmail := map[string]bool{}
  901. if len(foundList) > 0 {
  902. type trafficRow struct {
  903. Email string
  904. Enable bool
  905. }
  906. for _, batch := range chunkStrings(foundList, sqlInChunk) {
  907. var rows []trafficRow
  908. if err := db.Model(xray.ClientTraffic{}).
  909. Where("email IN ?", batch).
  910. Select("email, enable").
  911. Scan(&rows).Error; err == nil {
  912. for _, r := range rows {
  913. notDepletedByEmail[r.Email] = r.Enable
  914. }
  915. }
  916. }
  917. }
  918. var sharedSet map[string]bool
  919. if !keepTraffic {
  920. var sharedErr error
  921. sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(foundList, inboundId)
  922. if sharedErr != nil {
  923. for email := range foundEmails {
  924. res.perEmailSkipped[email] = sharedErr.Error()
  925. delete(foundEmails, email)
  926. }
  927. return res
  928. }
  929. }
  930. if !keepTraffic {
  931. purge := make([]string, 0, len(foundEmails))
  932. for email := range foundEmails {
  933. if !sharedSet[strings.ToLower(strings.TrimSpace(email))] {
  934. purge = append(purge, email)
  935. }
  936. }
  937. if len(purge) > 0 {
  938. // Serialize the IP/stat purge against the traffic poll to avoid the
  939. // cross-transaction lock-order deadlock on client_traffics.
  940. if delErr := runSerializedTx(func(tx *gorm.DB) error {
  941. if e := inboundSvc.delClientIPsByEmails(tx, purge); e != nil {
  942. logger.Error("Error in delete client IPs")
  943. return e
  944. }
  945. if e := inboundSvc.delClientStatsByEmails(tx, purge); e != nil {
  946. logger.Error("Delete stats Data Error")
  947. return e
  948. }
  949. return nil
  950. }); delErr != nil {
  951. for _, email := range purge {
  952. res.perEmailSkipped[email] = delErr.Error()
  953. delete(foundEmails, email)
  954. }
  955. }
  956. }
  957. }
  958. if oldInbound.NodeID == nil {
  959. rt, rterr := inboundSvc.runtimeFor(oldInbound)
  960. if rterr != nil {
  961. res.needRestart = true
  962. } else {
  963. for email := range foundEmails {
  964. if !enableByEmail[email] || !notDepletedByEmail[email] {
  965. continue
  966. }
  967. err1 := rt.RemoveUser(context.Background(), oldInbound, email)
  968. if err1 == nil {
  969. logger.Debug("Client deleted on", rt.Name(), ":", email)
  970. } else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", email)) {
  971. logger.Debug("User is already deleted. Nothing to do more...")
  972. } else {
  973. logger.Debug("Error in deleting client on", rt.Name(), ":", err1)
  974. res.needRestart = true
  975. }
  976. }
  977. }
  978. } else {
  979. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  980. if perr != nil {
  981. for email := range foundEmails {
  982. res.perEmailSkipped[email] = perr.Error()
  983. delete(foundEmails, email)
  984. }
  985. } else {
  986. // Large batches collapse into one reconcile push rather than M deletes.
  987. if push && len(foundEmails) > nodeBulkPushThreshold {
  988. push = false
  989. }
  990. if push {
  991. // bulkDelInboundClients only runs for full client deletion
  992. // (BulkDelete), so the node must drop its client record too,
  993. // not just detach from this inbound (#5797).
  994. pushFailed := false
  995. for email := range foundEmails {
  996. if err1 := rt.DeleteClient(context.Background(), email); err1 != nil {
  997. logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
  998. pushFailed = true
  999. }
  1000. }
  1001. if !pushFailed {
  1002. advancePushedInbound(rt, prevSettings, oldInbound)
  1003. }
  1004. }
  1005. }
  1006. }
  1007. // Serialize against the traffic poll to avoid the cross-transaction
  1008. // lock-order deadlock on inbounds/client_records (runSerializedTx).
  1009. txErr := runSerializedTx(func(tx *gorm.DB) error {
  1010. if err := tx.Save(oldInbound).Error; err != nil {
  1011. return err
  1012. }
  1013. finalClients, err := inboundSvc.GetClients(oldInbound)
  1014. if err != nil {
  1015. return err
  1016. }
  1017. if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
  1018. return err
  1019. }
  1020. if oldInbound.NodeID != nil {
  1021. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  1022. }
  1023. return nil
  1024. })
  1025. if txErr != nil {
  1026. for email := range foundEmails {
  1027. if _, skip := res.perEmailSkipped[email]; !skip {
  1028. res.perEmailSkipped[email] = txErr.Error()
  1029. }
  1030. }
  1031. }
  1032. return res
  1033. }
  1034. // BulkCreateResult mirrors BulkAdjustResult for the create flow.
  1035. type BulkCreateResult struct {
  1036. Created int `json:"created"`
  1037. Skipped []BulkCreateReport `json:"skipped,omitempty"`
  1038. }
  1039. type BulkCreateReport struct {
  1040. Email string `json:"email"`
  1041. Reason string `json:"reason"`
  1042. }
  1043. func (s *ClientService) BulkCreate(inboundSvc *InboundService, payloads []ClientCreatePayload) (BulkCreateResult, bool, error) {
  1044. result := BulkCreateResult{}
  1045. if len(payloads) == 0 {
  1046. return result, false, nil
  1047. }
  1048. skip := func(email, reason string) {
  1049. if strings.TrimSpace(email) == "" {
  1050. email = "(missing email)"
  1051. }
  1052. result.Skipped = append(result.Skipped, BulkCreateReport{Email: email, Reason: reason})
  1053. }
  1054. emailSubIDs, err := inboundSvc.getAllEmailSubIDs()
  1055. if err != nil {
  1056. emailSubIDs = nil
  1057. }
  1058. type prepared struct {
  1059. client model.Client
  1060. inboundIds []int
  1061. }
  1062. prep := make([]prepared, 0, len(payloads))
  1063. emails := make([]string, 0, len(payloads))
  1064. subIDs := make([]string, 0, len(payloads))
  1065. seenEmail := make(map[string]struct{}, len(payloads))
  1066. seenSubID := make(map[string]string, len(payloads))
  1067. for i := range payloads {
  1068. client := payloads[i].Client
  1069. email := strings.TrimSpace(client.Email)
  1070. if email == "" {
  1071. skip("", "client email is required")
  1072. continue
  1073. }
  1074. if verr := validateClientEmail(email); verr != nil {
  1075. skip(email, verr.Error())
  1076. continue
  1077. }
  1078. if verr := validateClientSubID(client.SubID); verr != nil {
  1079. skip(email, verr.Error())
  1080. continue
  1081. }
  1082. if len(payloads[i].InboundIds) == 0 {
  1083. skip(email, "at least one inbound is required")
  1084. continue
  1085. }
  1086. client.Email = email
  1087. if client.SubID == "" {
  1088. client.SubID = uuid.NewString()
  1089. }
  1090. if !client.Enable {
  1091. client.Enable = true
  1092. }
  1093. now := time.Now().UnixMilli()
  1094. if client.CreatedAt == 0 {
  1095. client.CreatedAt = now
  1096. }
  1097. client.UpdatedAt = now
  1098. le := strings.ToLower(email)
  1099. if _, dup := seenEmail[le]; dup {
  1100. skip(email, "email already in use: "+email)
  1101. continue
  1102. }
  1103. if owner, ok := seenSubID[client.SubID]; ok && owner != le {
  1104. skip(email, "subId already in use: "+client.SubID)
  1105. continue
  1106. }
  1107. seenEmail[le] = struct{}{}
  1108. seenSubID[client.SubID] = le
  1109. prep = append(prep, prepared{client: client, inboundIds: payloads[i].InboundIds})
  1110. emails = append(emails, email)
  1111. subIDs = append(subIDs, client.SubID)
  1112. }
  1113. if len(prep) == 0 {
  1114. return result, false, nil
  1115. }
  1116. db := database.GetDB()
  1117. const lookupChunk = 400
  1118. existingByEmail := make(map[string]model.ClientRecord, len(emails))
  1119. for start := 0; start < len(emails); start += lookupChunk {
  1120. end := min(start+lookupChunk, len(emails))
  1121. var rows []model.ClientRecord
  1122. if e := db.Where("email IN ?", emails[start:end]).Find(&rows).Error; e != nil {
  1123. return result, false, e
  1124. }
  1125. for i := range rows {
  1126. existingByEmail[strings.ToLower(rows[i].Email)] = rows[i]
  1127. }
  1128. }
  1129. existingSubOwner := make(map[string]string, len(subIDs))
  1130. for start := 0; start < len(subIDs); start += lookupChunk {
  1131. end := min(start+lookupChunk, len(subIDs))
  1132. var rows []model.ClientRecord
  1133. if e := db.Where("sub_id IN ?", subIDs[start:end]).Find(&rows).Error; e != nil {
  1134. return result, false, e
  1135. }
  1136. for i := range rows {
  1137. existingSubOwner[rows[i].SubID] = strings.ToLower(rows[i].Email)
  1138. }
  1139. }
  1140. inboundCache := make(map[int]*model.Inbound)
  1141. getIb := func(id int) (*model.Inbound, error) {
  1142. if ib, ok := inboundCache[id]; ok {
  1143. return ib, nil
  1144. }
  1145. ib, e := inboundSvc.GetInbound(id)
  1146. if e != nil {
  1147. return nil, e
  1148. }
  1149. inboundCache[id] = ib
  1150. return ib, nil
  1151. }
  1152. byInbound := make(map[int][]model.Client)
  1153. idxByInbound := make(map[int][]int)
  1154. inboundOrder := make([]int, 0)
  1155. failed := make([]bool, len(prep))
  1156. reason := make([]string, len(prep))
  1157. for idx := range prep {
  1158. le := strings.ToLower(prep[idx].client.Email)
  1159. if rec, ok := existingByEmail[le]; ok {
  1160. if rec.SubID != prep[idx].client.SubID {
  1161. failed[idx] = true
  1162. reason[idx] = "email already in use: " + prep[idx].client.Email
  1163. continue
  1164. }
  1165. if prep[idx].client.ID == "" {
  1166. prep[idx].client.ID = rec.UUID
  1167. }
  1168. if prep[idx].client.Password == "" {
  1169. prep[idx].client.Password = rec.Password
  1170. }
  1171. if prep[idx].client.Auth == "" {
  1172. prep[idx].client.Auth = rec.Auth
  1173. }
  1174. if prep[idx].client.Secret == "" {
  1175. prep[idx].client.Secret = rec.Secret
  1176. }
  1177. }
  1178. if owner, ok := existingSubOwner[prep[idx].client.SubID]; ok && owner != le {
  1179. failed[idx] = true
  1180. reason[idx] = "subId already in use: " + prep[idx].client.SubID
  1181. continue
  1182. }
  1183. ok := true
  1184. for _, ibId := range prep[idx].inboundIds {
  1185. ib, e := getIb(ibId)
  1186. if e != nil {
  1187. failed[idx] = true
  1188. reason[idx] = e.Error()
  1189. ok = false
  1190. break
  1191. }
  1192. if e := s.fillProtocolDefaults(&prep[idx].client, ib); e != nil {
  1193. failed[idx] = true
  1194. reason[idx] = e.Error()
  1195. ok = false
  1196. break
  1197. }
  1198. }
  1199. if !ok {
  1200. continue
  1201. }
  1202. for _, ibId := range prep[idx].inboundIds {
  1203. ib, _ := getIb(ibId)
  1204. if _, seen := byInbound[ibId]; !seen {
  1205. inboundOrder = append(inboundOrder, ibId)
  1206. }
  1207. byInbound[ibId] = append(byInbound[ibId], clientWithInboundFlow(prep[idx].client, ib))
  1208. idxByInbound[ibId] = append(idxByInbound[ibId], idx)
  1209. }
  1210. }
  1211. needRestart := false
  1212. for _, ibId := range inboundOrder {
  1213. payload, e := json.Marshal(map[string][]model.Client{"clients": byInbound[ibId]})
  1214. if e == nil {
  1215. var nr bool
  1216. nr, e = s.addInboundClient(inboundSvc, &model.Inbound{Id: ibId, Settings: string(payload)}, emailSubIDs)
  1217. if e == nil && nr {
  1218. needRestart = true
  1219. }
  1220. }
  1221. if e != nil {
  1222. for _, idx := range idxByInbound[ibId] {
  1223. failed[idx] = true
  1224. if reason[idx] == "" {
  1225. reason[idx] = e.Error()
  1226. }
  1227. }
  1228. }
  1229. }
  1230. for idx := range prep {
  1231. if failed[idx] {
  1232. skip(prep[idx].client.Email, reason[idx])
  1233. } else {
  1234. result.Created++
  1235. }
  1236. }
  1237. return result, needRestart, nil
  1238. }
  1239. func (s *ClientService) DelDepleted(inboundSvc *InboundService) (int, bool, error) {
  1240. db := database.GetDB()
  1241. now := time.Now().UnixMilli()
  1242. depletedClause := "reset = 0 and ((total > 0 and up + down >= total) or (expiry_time > 0 and expiry_time <= ?))"
  1243. var rows []xray.ClientTraffic
  1244. if err := db.Where(depletedClause, now).Find(&rows).Error; err != nil {
  1245. return 0, false, err
  1246. }
  1247. if len(rows) == 0 {
  1248. return 0, false, nil
  1249. }
  1250. seen := make(map[string]struct{}, len(rows))
  1251. emails := make([]string, 0, len(rows))
  1252. for _, r := range rows {
  1253. if r.Email == "" {
  1254. continue
  1255. }
  1256. if _, ok := seen[r.Email]; ok {
  1257. continue
  1258. }
  1259. seen[r.Email] = struct{}{}
  1260. emails = append(emails, r.Email)
  1261. }
  1262. if len(emails) == 0 {
  1263. return 0, false, nil
  1264. }
  1265. res, needRestart, err := s.BulkDelete(inboundSvc, emails, false)
  1266. if err != nil {
  1267. return res.Deleted, needRestart, err
  1268. }
  1269. return res.Deleted, needRestart, nil
  1270. }
  1271. type BulkSetEnableResult struct {
  1272. Changed int `json:"changed"`
  1273. Skipped []BulkSetEnableReport `json:"skipped,omitempty"`
  1274. }
  1275. type BulkSetEnableReport struct {
  1276. Email string `json:"email"`
  1277. Reason string `json:"reason"`
  1278. }
  1279. func (s *ClientService) BulkSetEnable(inboundSvc *InboundService, emails []string, enable bool) (BulkSetEnableResult, bool, error) {
  1280. result := BulkSetEnableResult{}
  1281. seen := map[string]struct{}{}
  1282. cleanEmails := make([]string, 0, len(emails))
  1283. for _, e := range emails {
  1284. e = strings.TrimSpace(e)
  1285. if e == "" {
  1286. continue
  1287. }
  1288. if _, ok := seen[e]; ok {
  1289. continue
  1290. }
  1291. seen[e] = struct{}{}
  1292. cleanEmails = append(cleanEmails, e)
  1293. }
  1294. if len(cleanEmails) == 0 {
  1295. return result, false, nil
  1296. }
  1297. db := database.GetDB()
  1298. var records []model.ClientRecord
  1299. for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
  1300. var rows []model.ClientRecord
  1301. if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
  1302. return result, false, err
  1303. }
  1304. records = append(records, rows...)
  1305. }
  1306. recordsByEmail := make(map[string]*model.ClientRecord, len(records))
  1307. for i := range records {
  1308. recordsByEmail[records[i].Email] = &records[i]
  1309. }
  1310. skippedReasons := map[string]string{}
  1311. for _, email := range cleanEmails {
  1312. if _, ok := recordsByEmail[email]; !ok {
  1313. skippedReasons[email] = "client not found"
  1314. }
  1315. }
  1316. clientIds := make([]int, 0, len(recordsByEmail))
  1317. recordIdToEmail := make(map[int]string, len(recordsByEmail))
  1318. for _, r := range recordsByEmail {
  1319. clientIds = append(clientIds, r.Id)
  1320. recordIdToEmail[r.Id] = r.Email
  1321. }
  1322. emailsByInbound := map[int][]string{}
  1323. if len(clientIds) > 0 {
  1324. var mappings []model.ClientInbound
  1325. for _, batch := range chunkInts(clientIds, sqlInChunk) {
  1326. var rows []model.ClientInbound
  1327. if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
  1328. return result, false, err
  1329. }
  1330. mappings = append(mappings, rows...)
  1331. }
  1332. for _, m := range mappings {
  1333. email, ok := recordIdToEmail[m.ClientId]
  1334. if !ok {
  1335. continue
  1336. }
  1337. emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
  1338. }
  1339. }
  1340. needRestart := false
  1341. for inboundId, ibEmails := range emailsByInbound {
  1342. ibRes := s.bulkSetEnableInboundClients(inboundSvc, inboundId, ibEmails, enable)
  1343. if ibRes.needRestart {
  1344. needRestart = true
  1345. }
  1346. for email, reason := range ibRes.perEmailSkipped {
  1347. if _, already := skippedReasons[email]; !already {
  1348. skippedReasons[email] = reason
  1349. }
  1350. }
  1351. }
  1352. successEmails := make([]string, 0, len(recordsByEmail))
  1353. for email := range recordsByEmail {
  1354. if _, skipped := skippedReasons[email]; skipped {
  1355. continue
  1356. }
  1357. successEmails = append(successEmails, email)
  1358. }
  1359. if len(successEmails) > 0 {
  1360. now := time.Now().UnixMilli()
  1361. if err := runSerializedTx(func(tx *gorm.DB) error {
  1362. for _, batch := range chunkStrings(successEmails, sqlInChunk) {
  1363. if e := tx.Model(xray.ClientTraffic{}).Where("email IN ?", batch).Update("enable", enable).Error; e != nil {
  1364. return e
  1365. }
  1366. if e := tx.Model(&model.ClientRecord{}).Where("email IN ?", batch).
  1367. Updates(map[string]any{"enable": enable, "updated_at": now}).Error; e != nil {
  1368. return e
  1369. }
  1370. }
  1371. return nil
  1372. }); err != nil {
  1373. return result, needRestart, err
  1374. }
  1375. }
  1376. result.Changed = len(successEmails)
  1377. for email, reason := range skippedReasons {
  1378. result.Skipped = append(result.Skipped, BulkSetEnableReport{Email: email, Reason: reason})
  1379. }
  1380. return result, needRestart, nil
  1381. }
  1382. type bulkSetEnableInboundResult struct {
  1383. perEmailSkipped map[string]string
  1384. needRestart bool
  1385. }
  1386. func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService, inboundId int, emails []string, enable bool) bulkSetEnableInboundResult {
  1387. res := bulkSetEnableInboundResult{perEmailSkipped: map[string]string{}}
  1388. defer lockInbound(inboundId).Unlock()
  1389. oldInbound, err := inboundSvc.GetInbound(inboundId)
  1390. if err != nil {
  1391. for _, e := range emails {
  1392. res.perEmailSkipped[e] = err.Error()
  1393. }
  1394. return res
  1395. }
  1396. var settings map[string]any
  1397. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  1398. for _, e := range emails {
  1399. res.perEmailSkipped[e] = err.Error()
  1400. }
  1401. return res
  1402. }
  1403. wanted := make(map[string]struct{}, len(emails))
  1404. for _, email := range emails {
  1405. wanted[email] = struct{}{}
  1406. }
  1407. cipher := ""
  1408. if oldInbound.Protocol == model.Shadowsocks {
  1409. cipher, _ = settings["method"].(string)
  1410. }
  1411. type changedClient struct {
  1412. email string
  1413. wasEnable bool
  1414. client model.Client
  1415. }
  1416. var changed []changedClient
  1417. found := map[string]bool{}
  1418. nowMs := time.Now().UnixMilli()
  1419. interfaceClients, _ := settings["clients"].([]any)
  1420. for i, c := range interfaceClients {
  1421. entry, ok := c.(map[string]any)
  1422. if !ok {
  1423. continue
  1424. }
  1425. email, _ := entry["email"].(string)
  1426. if _, want := wanted[email]; !want || email == "" {
  1427. continue
  1428. }
  1429. found[email] = true
  1430. prev, _ := entry["enable"].(bool)
  1431. if prev == enable {
  1432. continue
  1433. }
  1434. entry["enable"] = enable
  1435. entry["updated_at"] = nowMs
  1436. interfaceClients[i] = entry
  1437. // Build the pushed client from the inbound JSON (the per-inbound source of
  1438. // truth), so a remote UpdateUser carries every field and never zeroes
  1439. // subId/totalGB/expiry from drifting ClientRecord columns (#4628/#4792).
  1440. var parsed model.Client
  1441. if b, mErr := json.Marshal(entry); mErr == nil {
  1442. _ = json.Unmarshal(b, &parsed)
  1443. }
  1444. parsed.Email = email
  1445. parsed.Enable = enable
  1446. changed = append(changed, changedClient{email: email, wasEnable: prev, client: parsed})
  1447. }
  1448. for email := range wanted {
  1449. if !found[email] {
  1450. res.perEmailSkipped[email] = "Client Not Found In Inbound"
  1451. }
  1452. }
  1453. if len(changed) == 0 {
  1454. return res
  1455. }
  1456. settings["clients"] = interfaceClients
  1457. newSettings, err := json.MarshalIndent(settings, "", " ")
  1458. if err != nil {
  1459. for _, ch := range changed {
  1460. res.perEmailSkipped[ch.email] = err.Error()
  1461. }
  1462. return res
  1463. }
  1464. prevSettings := oldInbound.Settings
  1465. oldInbound.Settings = string(newSettings)
  1466. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  1467. if perr != nil {
  1468. for _, ch := range changed {
  1469. res.perEmailSkipped[ch.email] = perr.Error()
  1470. }
  1471. return res
  1472. }
  1473. if oldInbound.NodeID != nil && push && len(changed) > nodeBulkPushThreshold {
  1474. push = false
  1475. }
  1476. txErr := runSerializedTx(func(tx *gorm.DB) error {
  1477. if e := tx.Save(oldInbound).Error; e != nil {
  1478. return e
  1479. }
  1480. finalClients, gcErr := inboundSvc.GetClients(oldInbound)
  1481. if gcErr != nil {
  1482. return gcErr
  1483. }
  1484. if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
  1485. return err
  1486. }
  1487. if oldInbound.NodeID != nil {
  1488. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  1489. }
  1490. return nil
  1491. })
  1492. if txErr != nil {
  1493. for _, ch := range changed {
  1494. res.perEmailSkipped[ch.email] = txErr.Error()
  1495. }
  1496. return res
  1497. }
  1498. if oldInbound.NodeID == nil {
  1499. if !push {
  1500. res.needRestart = true
  1501. } else {
  1502. for _, ch := range changed {
  1503. if enable {
  1504. err1 := rt.AddUser(context.Background(), oldInbound, map[string]any{
  1505. "email": ch.client.Email,
  1506. "id": ch.client.ID,
  1507. "security": ch.client.Security,
  1508. "flow": ch.client.Flow,
  1509. "auth": ch.client.Auth,
  1510. "password": ch.client.Password,
  1511. "cipher": cipher,
  1512. })
  1513. if err1 != nil {
  1514. logger.Debug("Error in adding client on", rt.Name(), ":", err1)
  1515. res.needRestart = true
  1516. }
  1517. } else if ch.wasEnable {
  1518. err1 := rt.RemoveUser(context.Background(), oldInbound, ch.email)
  1519. if err1 != nil && !strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", ch.email)) {
  1520. logger.Debug("Error in removing client on", rt.Name(), ":", err1)
  1521. res.needRestart = true
  1522. }
  1523. }
  1524. }
  1525. }
  1526. } else if push {
  1527. pushFailed := false
  1528. for _, ch := range changed {
  1529. updated := ch.client
  1530. updated.UpdatedAt = nowMs
  1531. if err1 := rt.UpdateUser(context.Background(), oldInbound, ch.email, updated); err1 != nil {
  1532. logger.Warning("Error in updating client on", rt.Name(), ":", err1)
  1533. pushFailed = true
  1534. }
  1535. }
  1536. if !pushFailed {
  1537. advancePushedInbound(rt, prevSettings, oldInbound)
  1538. }
  1539. }
  1540. return res
  1541. }