client_bulk.go 47 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681
  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. execFailed := map[string]bool{}
  374. for inboundId, ibEmails := range emailsByInbound {
  375. ibRes := s.bulkAdjustInboundClients(inboundSvc, inboundId, ibEmails, plan, flow)
  376. if ibRes.needRestart {
  377. needRestart = true
  378. }
  379. for email := range ibRes.flowHonored {
  380. flowHonored[email] = true
  381. }
  382. for email := range ibRes.flowIneligible {
  383. flowIneligible[email] = true
  384. }
  385. for email, reason := range ibRes.perEmailSkipped {
  386. execFailed[email] = true
  387. if _, already := skippedReasons[email]; !already {
  388. skippedReasons[email] = reason
  389. }
  390. }
  391. }
  392. now := time.Now().Unix() * 1000
  393. cond := depletedCond(db)
  394. candidateEmails := make([]string, 0, len(plan))
  395. for email, entry := range plan {
  396. if entry.applyExpiry || entry.applyTotal {
  397. candidateEmails = append(candidateEmails, email)
  398. }
  399. }
  400. wasDisabledDepleted := map[string]struct{}{}
  401. for _, batch := range chunkStrings(candidateEmails, sqlInChunk) {
  402. var rows []string
  403. if err := db.Model(xray.ClientTraffic{}).
  404. Where(cond+" AND enable = ? AND email IN ?", now, false, batch).
  405. Pluck("email", &rows).Error; err != nil {
  406. return result, needRestart, err
  407. }
  408. for _, e := range rows {
  409. wasDisabledDepleted[e] = struct{}{}
  410. }
  411. }
  412. adjusted := map[string]struct{}{}
  413. for email, entry := range plan {
  414. if execFailed[email] {
  415. continue
  416. }
  417. updates := map[string]any{}
  418. if entry.applyExpiry {
  419. updates["expiry_time"] = entry.newExpiry
  420. }
  421. if entry.applyTotal {
  422. updates["total"] = entry.newTotal
  423. }
  424. if len(updates) > 0 {
  425. if err := db.Model(xray.ClientTraffic{}).Where("email = ?", email).Updates(updates).Error; err != nil {
  426. if _, already := skippedReasons[email]; !already {
  427. skippedReasons[email] = err.Error()
  428. }
  429. continue
  430. }
  431. }
  432. // Counted when expiry/total changed, or a flow directive was honored
  433. // for this client (flow lives in the inbound JSON, not ClientTraffic).
  434. if len(updates) > 0 || flowHonored[email] {
  435. adjusted[email] = struct{}{}
  436. }
  437. }
  438. result.Adjusted = len(adjusted)
  439. for email, reason := range skippedReasons {
  440. result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: reason})
  441. }
  442. // Report a flow directive that no inbound could carry — only when it was not
  443. // honored anywhere and the client has no other (expiry/total) skip reason.
  444. // The expiry/total part, if any, has already been applied and counted above.
  445. for email := range flowIneligible {
  446. if flowHonored[email] {
  447. continue
  448. }
  449. if _, already := skippedReasons[email]; already {
  450. continue
  451. }
  452. result.Skipped = append(result.Skipped, BulkAdjustReport{Email: email, Reason: "flow not supported on inbound"})
  453. }
  454. if len(wasDisabledDepleted) > 0 {
  455. stillDepleted := map[string]struct{}{}
  456. wasList := make([]string, 0, len(wasDisabledDepleted))
  457. for e := range wasDisabledDepleted {
  458. wasList = append(wasList, e)
  459. }
  460. for _, batch := range chunkStrings(wasList, sqlInChunk) {
  461. var rows []string
  462. if err := db.Model(xray.ClientTraffic{}).
  463. Where(cond+" AND email IN ?", now, batch).
  464. Pluck("email", &rows).Error; err != nil {
  465. return result, needRestart, err
  466. }
  467. for _, e := range rows {
  468. stillDepleted[e] = struct{}{}
  469. }
  470. }
  471. reEnable := make([]string, 0, len(wasDisabledDepleted))
  472. for e := range wasDisabledDepleted {
  473. if _, still := stillDepleted[e]; !still {
  474. reEnable = append(reEnable, e)
  475. }
  476. }
  477. if len(reEnable) > 0 {
  478. _, nr, err := s.BulkSetEnable(inboundSvc, reEnable, true)
  479. if err != nil {
  480. return result, needRestart, err
  481. }
  482. if nr {
  483. needRestart = true
  484. }
  485. }
  486. }
  487. return result, needRestart, nil
  488. }
  489. type bulkInboundAdjustResult struct {
  490. perEmailSkipped map[string]string
  491. flowHonored map[string]bool
  492. // flowIneligible is tracked apart from perEmailSkipped: a flow directive
  493. // that an inbound cannot carry must not suppress the expiry/total write for
  494. // the same client (which would diverge the inbound JSON / ClientRecord from
  495. // ClientTraffic). It only feeds the final Skipped report.
  496. flowIneligible map[string]bool
  497. needRestart bool
  498. }
  499. // bulkAdjustInboundClients applies expiry/total deltas to multiple clients
  500. // inside a single inbound's settings JSON. The xray runtime is updated
  501. // only for remote-node inbounds; local nodes do not need a notification
  502. // because the AddUser payload does not include totalGB/expiryTime —
  503. // changing those fields is identity-preserving and the panel's traffic
  504. // enforcement loop picks up the new limits from ClientTraffic directly.
  505. func (s *ClientService) bulkAdjustInboundClients(
  506. inboundSvc *InboundService,
  507. inboundId int,
  508. emails []string,
  509. plan map[string]*bulkAdjustEntry,
  510. flow string,
  511. ) bulkInboundAdjustResult {
  512. res := bulkInboundAdjustResult{perEmailSkipped: map[string]string{}, flowHonored: map[string]bool{}, flowIneligible: map[string]bool{}}
  513. defer lockInbound(inboundId).Unlock()
  514. oldInbound, err := inboundSvc.GetInbound(inboundId)
  515. if err != nil {
  516. logger.Error("Load Old Data Error")
  517. for _, e := range emails {
  518. res.perEmailSkipped[e] = err.Error()
  519. }
  520. return res
  521. }
  522. var settings map[string]any
  523. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  524. for _, e := range emails {
  525. res.perEmailSkipped[e] = err.Error()
  526. }
  527. return res
  528. }
  529. // Match by email — the client's stable identity (see Delete). Credentials
  530. // can drift from the inbound JSON, so they are never used for matching.
  531. wantedEmails := make(map[string]struct{}, len(emails))
  532. for _, email := range emails {
  533. if plan[email] == nil {
  534. res.perEmailSkipped[email] = "client not found"
  535. continue
  536. }
  537. wantedEmails[email] = struct{}{}
  538. }
  539. // Flow eligibility is a property of the inbound (protocol + transport), so
  540. // resolve it once. Clearing flow is always allowed; setting a vision flow
  541. // is only honored on an inbound that can carry it.
  542. flowEligible := flow == bulkFlowClear ||
  543. inboundCanEnableTlsFlow(string(oldInbound.Protocol), oldInbound.StreamSettings, oldInbound.Settings)
  544. interfaceClients, _ := settings["clients"].([]any)
  545. foundEmails := map[string]bool{}
  546. flowChanged := false
  547. nowMs := time.Now().Unix() * 1000
  548. for i, client := range interfaceClients {
  549. c, ok := client.(map[string]any)
  550. if !ok {
  551. continue
  552. }
  553. targetEmail, _ := c["email"].(string)
  554. if _, want := wantedEmails[targetEmail]; !want || targetEmail == "" {
  555. continue
  556. }
  557. entry := plan[targetEmail]
  558. if entry.applyExpiry {
  559. c["expiryTime"] = entry.newExpiry
  560. }
  561. if entry.applyTotal {
  562. c["totalGB"] = entry.newTotal
  563. }
  564. if flow != "" {
  565. if flowEligible {
  566. want := ""
  567. if flow != bulkFlowClear {
  568. want = flow
  569. }
  570. if cur, _ := c["flow"].(string); cur != want {
  571. c["flow"] = want
  572. flowChanged = true
  573. }
  574. res.flowHonored[targetEmail] = true
  575. } else {
  576. // Record separately so this never suppresses the expiry/total
  577. // write for the same client (see flowIneligible doc).
  578. res.flowIneligible[targetEmail] = true
  579. }
  580. }
  581. c["updated_at"] = nowMs
  582. interfaceClients[i] = c
  583. foundEmails[targetEmail] = true
  584. }
  585. for email := range wantedEmails {
  586. if !foundEmails[email] {
  587. res.perEmailSkipped[email] = "Client Not Found In Inbound"
  588. }
  589. }
  590. if len(foundEmails) == 0 {
  591. return res
  592. }
  593. settings["clients"] = interfaceClients
  594. newSettings, err := json.MarshalIndent(settings, "", " ")
  595. if err != nil {
  596. for email := range foundEmails {
  597. res.perEmailSkipped[email] = err.Error()
  598. }
  599. return res
  600. }
  601. prevSettings := oldInbound.Settings
  602. oldInbound.Settings = string(newSettings)
  603. // A flow change rewrites the user's xray config, which the lightweight
  604. // UpdateUser push below does not carry. Local nodes reload via restart;
  605. // remote nodes get a full reconcile (MarkNodeDirty) instead of a per-user push.
  606. if flowChanged && oldInbound.NodeID == nil {
  607. res.needRestart = true
  608. }
  609. if oldInbound.NodeID != nil {
  610. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  611. if perr != nil {
  612. for email := range foundEmails {
  613. res.perEmailSkipped[email] = perr.Error()
  614. delete(foundEmails, email)
  615. }
  616. } else {
  617. if flowChanged {
  618. push = false
  619. }
  620. // Large batches collapse into one reconcile push rather than M updates.
  621. if push && len(foundEmails) > nodeBulkPushThreshold {
  622. push = false
  623. }
  624. if push {
  625. pushFailed := false
  626. for email := range foundEmails {
  627. entry := plan[email]
  628. updated := *entry.record.ToClient()
  629. if entry.applyExpiry {
  630. updated.ExpiryTime = entry.newExpiry
  631. }
  632. if entry.applyTotal {
  633. updated.TotalGB = entry.newTotal
  634. }
  635. updated.UpdatedAt = nowMs
  636. if err1 := rt.UpdateUser(context.Background(), oldInbound, email, updated); err1 != nil {
  637. logger.Warning("Error in updating client on", rt.Name(), ":", err1)
  638. pushFailed = true
  639. }
  640. }
  641. if !pushFailed {
  642. advancePushedInbound(rt, prevSettings, oldInbound)
  643. }
  644. }
  645. }
  646. }
  647. // Serialize against the traffic poll to avoid the cross-transaction
  648. // lock-order deadlock on inbounds/client_records (runSerializedTx).
  649. txErr := runSerializedTx(func(tx *gorm.DB) error {
  650. if err := tx.Save(oldInbound).Error; err != nil {
  651. return err
  652. }
  653. finalClients, gcErr := inboundSvc.GetClients(oldInbound)
  654. if gcErr != nil {
  655. return gcErr
  656. }
  657. if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
  658. return err
  659. }
  660. if oldInbound.NodeID != nil {
  661. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  662. }
  663. return nil
  664. })
  665. if txErr != nil {
  666. for email := range foundEmails {
  667. if _, skip := res.perEmailSkipped[email]; !skip {
  668. res.perEmailSkipped[email] = txErr.Error()
  669. }
  670. }
  671. }
  672. return res
  673. }
  674. // BulkDeleteResult mirrors BulkAdjustResult: total deleted plus per-email
  675. // skip reasons when an email could not be processed.
  676. type BulkDeleteResult struct {
  677. Deleted int `json:"deleted"`
  678. Skipped []BulkDeleteReport `json:"skipped,omitempty"`
  679. }
  680. type BulkDeleteReport struct {
  681. Email string `json:"email"`
  682. Reason string `json:"reason"`
  683. }
  684. // BulkDelete removes every client in the list in one optimized pass.
  685. // Instead of running the full single-delete pipeline N times (which would
  686. // re-read, re-parse, and re-write each inbound's settings JSON for every
  687. // email), it groups emails by inbound and performs a single
  688. // read-modify-write per inbound. Per-row DB cleanups are also batched with
  689. // IN-clause queries at the end. Errors on a particular email are recorded
  690. // in the Skipped list and processing continues for the rest.
  691. func (s *ClientService) BulkDelete(inboundSvc *InboundService, emails []string, keepTraffic bool) (BulkDeleteResult, bool, error) {
  692. result := BulkDeleteResult{}
  693. seen := map[string]struct{}{}
  694. cleanEmails := make([]string, 0, len(emails))
  695. for _, e := range emails {
  696. e = strings.TrimSpace(e)
  697. if e == "" {
  698. continue
  699. }
  700. if _, ok := seen[e]; ok {
  701. continue
  702. }
  703. seen[e] = struct{}{}
  704. cleanEmails = append(cleanEmails, e)
  705. }
  706. if len(cleanEmails) == 0 {
  707. return result, false, nil
  708. }
  709. db := database.GetDB()
  710. var records []model.ClientRecord
  711. for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
  712. var rows []model.ClientRecord
  713. if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
  714. return result, false, err
  715. }
  716. records = append(records, rows...)
  717. }
  718. recordsByEmail := make(map[string]*model.ClientRecord, len(records))
  719. tombstoneEmails := make([]string, 0, len(records))
  720. for i := range records {
  721. recordsByEmail[records[i].Email] = &records[i]
  722. tombstoneEmails = append(tombstoneEmails, records[i].Email)
  723. }
  724. tombstoneClientEmails(tombstoneEmails)
  725. skippedReasons := map[string]string{}
  726. for _, email := range cleanEmails {
  727. if _, ok := recordsByEmail[email]; !ok {
  728. skippedReasons[email] = "client not found"
  729. }
  730. }
  731. clientIds := make([]int, 0, len(recordsByEmail))
  732. recordIdToEmail := make(map[int]string, len(recordsByEmail))
  733. for _, r := range recordsByEmail {
  734. clientIds = append(clientIds, r.Id)
  735. recordIdToEmail[r.Id] = r.Email
  736. }
  737. emailsByInbound := map[int][]string{}
  738. if len(clientIds) > 0 {
  739. var mappings []model.ClientInbound
  740. for _, batch := range chunkInts(clientIds, sqlInChunk) {
  741. var rows []model.ClientInbound
  742. if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
  743. return result, false, err
  744. }
  745. mappings = append(mappings, rows...)
  746. }
  747. for _, m := range mappings {
  748. email, ok := recordIdToEmail[m.ClientId]
  749. if !ok {
  750. continue
  751. }
  752. emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
  753. }
  754. }
  755. needRestart := false
  756. for inboundId, ibEmails := range emailsByInbound {
  757. ibResult := s.bulkDelInboundClients(inboundSvc, inboundId, ibEmails, recordsByEmail, keepTraffic)
  758. if ibResult.needRestart {
  759. needRestart = true
  760. }
  761. for email, reason := range ibResult.perEmailSkipped {
  762. if _, already := skippedReasons[email]; !already {
  763. skippedReasons[email] = reason
  764. }
  765. }
  766. }
  767. successEmails := make([]string, 0, len(recordsByEmail))
  768. successIds := make([]int, 0, len(recordsByEmail))
  769. for email, rec := range recordsByEmail {
  770. if _, skipped := skippedReasons[email]; skipped {
  771. continue
  772. }
  773. successEmails = append(successEmails, email)
  774. successIds = append(successIds, rec.Id)
  775. }
  776. if len(successIds) > 0 {
  777. // Serialize the row cleanup against the traffic poll to avoid the
  778. // cross-transaction lock-order deadlock on client_traffics/inbounds.
  779. if err := runSerializedTx(func(tx *gorm.DB) error {
  780. if e := adjustGroupBaselinesForRemovedTraffic(tx, successEmails); e != nil {
  781. return e
  782. }
  783. for _, batch := range chunkInts(successIds, sqlInChunk) {
  784. if e := tx.Where("client_id IN ?", batch).Delete(&model.ClientInbound{}).Error; e != nil {
  785. return e
  786. }
  787. }
  788. if !keepTraffic && len(successEmails) > 0 {
  789. for _, batch := range chunkStrings(successEmails, sqlInChunk) {
  790. if e := tx.Where("email IN ?", batch).Delete(&xray.ClientTraffic{}).Error; e != nil {
  791. return e
  792. }
  793. if e := tx.Where("client_email IN ?", batch).Delete(&model.InboundClientIps{}).Error; e != nil {
  794. return e
  795. }
  796. }
  797. }
  798. for _, batch := range chunkInts(successIds, sqlInChunk) {
  799. if e := tx.Where("id IN ?", batch).Delete(&model.ClientRecord{}).Error; e != nil {
  800. return e
  801. }
  802. }
  803. return nil
  804. }); err != nil {
  805. return result, needRestart, err
  806. }
  807. }
  808. result.Deleted = len(successEmails)
  809. for email, reason := range skippedReasons {
  810. result.Skipped = append(result.Skipped, BulkDeleteReport{Email: email, Reason: reason})
  811. }
  812. return result, needRestart, nil
  813. }
  814. type bulkInboundDeleteResult struct {
  815. perEmailSkipped map[string]string
  816. needRestart bool
  817. }
  818. // bulkDelInboundClients removes multiple clients from a single inbound's
  819. // settings JSON in one read-modify-write cycle, runs the xray runtime
  820. // RemoveUser/DeleteUser calls, and persists the inbound. The returned map
  821. // holds per-email failure reasons; emails not present in the map are
  822. // considered successful for this inbound.
  823. func (s *ClientService) bulkDelInboundClients(
  824. inboundSvc *InboundService,
  825. inboundId int,
  826. emails []string,
  827. records map[string]*model.ClientRecord,
  828. keepTraffic bool,
  829. ) bulkInboundDeleteResult {
  830. res := bulkInboundDeleteResult{perEmailSkipped: map[string]string{}}
  831. defer lockInbound(inboundId).Unlock()
  832. oldInbound, err := inboundSvc.GetInbound(inboundId)
  833. if err != nil {
  834. logger.Error("Load Old Data Error")
  835. for _, e := range emails {
  836. res.perEmailSkipped[e] = err.Error()
  837. }
  838. return res
  839. }
  840. var settings map[string]any
  841. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  842. for _, e := range emails {
  843. res.perEmailSkipped[e] = err.Error()
  844. }
  845. return res
  846. }
  847. // Match by email — the client's stable identity (see Delete). Removes every
  848. // entry carrying a wanted email, independent of credential drift.
  849. wantedEmails := make(map[string]struct{}, len(emails))
  850. for _, email := range emails {
  851. if records[email] == nil {
  852. res.perEmailSkipped[email] = "client not found"
  853. continue
  854. }
  855. wantedEmails[email] = struct{}{}
  856. }
  857. interfaceClients, _ := settings["clients"].([]any)
  858. newClients := make([]any, 0, len(interfaceClients))
  859. foundEmails := map[string]bool{}
  860. enableByEmail := map[string]bool{}
  861. for _, client := range interfaceClients {
  862. c, ok := client.(map[string]any)
  863. if !ok {
  864. newClients = append(newClients, client)
  865. continue
  866. }
  867. em, _ := c["email"].(string)
  868. if _, found := wantedEmails[em]; found && em != "" {
  869. foundEmails[em] = true
  870. en, _ := c["enable"].(bool)
  871. enableByEmail[em] = en
  872. continue
  873. }
  874. newClients = append(newClients, client)
  875. }
  876. for email := range wantedEmails {
  877. if !foundEmails[email] {
  878. res.perEmailSkipped[email] = "Client Not Found In Inbound"
  879. }
  880. }
  881. db := database.GetDB()
  882. newClients = compactOrphans(db, newClients)
  883. if newClients == nil {
  884. newClients = []any{}
  885. }
  886. settings["clients"] = newClients
  887. newSettings, err := json.MarshalIndent(settings, "", " ")
  888. if err != nil {
  889. for email := range foundEmails {
  890. if _, skip := res.perEmailSkipped[email]; !skip {
  891. res.perEmailSkipped[email] = err.Error()
  892. }
  893. }
  894. return res
  895. }
  896. prevSettings := oldInbound.Settings
  897. oldInbound.Settings = string(newSettings)
  898. foundList := make([]string, 0, len(foundEmails))
  899. for email := range foundEmails {
  900. foundList = append(foundList, email)
  901. }
  902. notDepletedByEmail := map[string]bool{}
  903. if len(foundList) > 0 {
  904. type trafficRow struct {
  905. Email string
  906. Enable bool
  907. }
  908. for _, batch := range chunkStrings(foundList, sqlInChunk) {
  909. var rows []trafficRow
  910. if err := db.Model(xray.ClientTraffic{}).
  911. Where("email IN ?", batch).
  912. Select("email, enable").
  913. Scan(&rows).Error; err == nil {
  914. for _, r := range rows {
  915. notDepletedByEmail[r.Email] = r.Enable
  916. }
  917. }
  918. }
  919. }
  920. var sharedSet map[string]bool
  921. if !keepTraffic {
  922. var sharedErr error
  923. sharedSet, sharedErr = inboundSvc.emailsUsedByOtherInbounds(foundList, inboundId)
  924. if sharedErr != nil {
  925. for email := range foundEmails {
  926. res.perEmailSkipped[email] = sharedErr.Error()
  927. delete(foundEmails, email)
  928. }
  929. return res
  930. }
  931. }
  932. if !keepTraffic {
  933. purge := make([]string, 0, len(foundEmails))
  934. for email := range foundEmails {
  935. if !sharedSet[strings.ToLower(strings.TrimSpace(email))] {
  936. purge = append(purge, email)
  937. }
  938. }
  939. if len(purge) > 0 {
  940. // Serialize the IP/stat purge against the traffic poll to avoid the
  941. // cross-transaction lock-order deadlock on client_traffics.
  942. if delErr := runSerializedTx(func(tx *gorm.DB) error {
  943. if e := inboundSvc.delClientIPsByEmails(tx, purge); e != nil {
  944. logger.Error("Error in delete client IPs")
  945. return e
  946. }
  947. if e := inboundSvc.delClientStatsByEmails(tx, purge); e != nil {
  948. logger.Error("Delete stats Data Error")
  949. return e
  950. }
  951. return nil
  952. }); delErr != nil {
  953. for _, email := range purge {
  954. res.perEmailSkipped[email] = delErr.Error()
  955. delete(foundEmails, email)
  956. }
  957. }
  958. }
  959. }
  960. if oldInbound.NodeID == nil {
  961. rt, rterr := inboundSvc.runtimeFor(oldInbound)
  962. if rterr != nil {
  963. res.needRestart = true
  964. } else {
  965. for email := range foundEmails {
  966. if !enableByEmail[email] || !notDepletedByEmail[email] {
  967. continue
  968. }
  969. err1 := rt.RemoveUser(context.Background(), oldInbound, email)
  970. if err1 == nil {
  971. logger.Debug("Client deleted on", rt.Name(), ":", email)
  972. } else if strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", email)) {
  973. logger.Debug("User is already deleted. Nothing to do more...")
  974. } else {
  975. logger.Debug("Error in deleting client on", rt.Name(), ":", err1)
  976. res.needRestart = true
  977. }
  978. }
  979. }
  980. } else {
  981. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  982. if perr != nil {
  983. for email := range foundEmails {
  984. res.perEmailSkipped[email] = perr.Error()
  985. delete(foundEmails, email)
  986. }
  987. } else {
  988. // Large batches collapse into one reconcile push rather than M deletes.
  989. if push && len(foundEmails) > nodeBulkPushThreshold {
  990. push = false
  991. }
  992. if push {
  993. // bulkDelInboundClients only runs for full client deletion
  994. // (BulkDelete), so the node must drop its client record too,
  995. // not just detach from this inbound (#5797).
  996. pushFailed := false
  997. for email := range foundEmails {
  998. if err1 := rt.DeleteClient(context.Background(), email); err1 != nil {
  999. logger.Warning("Error in deleting client on", rt.Name(), ":", err1)
  1000. pushFailed = true
  1001. }
  1002. }
  1003. if !pushFailed {
  1004. advancePushedInbound(rt, prevSettings, oldInbound)
  1005. }
  1006. }
  1007. }
  1008. }
  1009. // Serialize against the traffic poll to avoid the cross-transaction
  1010. // lock-order deadlock on inbounds/client_records (runSerializedTx).
  1011. txErr := runSerializedTx(func(tx *gorm.DB) error {
  1012. if err := tx.Save(oldInbound).Error; err != nil {
  1013. return err
  1014. }
  1015. finalClients, err := inboundSvc.GetClients(oldInbound)
  1016. if err != nil {
  1017. return err
  1018. }
  1019. if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
  1020. return err
  1021. }
  1022. if oldInbound.NodeID != nil {
  1023. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  1024. }
  1025. return nil
  1026. })
  1027. if txErr != nil {
  1028. for email := range foundEmails {
  1029. if _, skip := res.perEmailSkipped[email]; !skip {
  1030. res.perEmailSkipped[email] = txErr.Error()
  1031. }
  1032. }
  1033. }
  1034. return res
  1035. }
  1036. // BulkCreateResult mirrors BulkAdjustResult for the create flow.
  1037. type BulkCreateResult struct {
  1038. Created int `json:"created"`
  1039. Skipped []BulkCreateReport `json:"skipped,omitempty"`
  1040. }
  1041. type BulkCreateReport struct {
  1042. Email string `json:"email"`
  1043. Reason string `json:"reason"`
  1044. }
  1045. func (s *ClientService) BulkCreate(inboundSvc *InboundService, payloads []ClientCreatePayload) (BulkCreateResult, bool, error) {
  1046. result := BulkCreateResult{}
  1047. if len(payloads) == 0 {
  1048. return result, false, nil
  1049. }
  1050. skip := func(email, reason string) {
  1051. if strings.TrimSpace(email) == "" {
  1052. email = "(missing email)"
  1053. }
  1054. result.Skipped = append(result.Skipped, BulkCreateReport{Email: email, Reason: reason})
  1055. }
  1056. emailSubIDs, err := inboundSvc.getAllEmailSubIDs()
  1057. if err != nil {
  1058. emailSubIDs = nil
  1059. }
  1060. type prepared struct {
  1061. client model.Client
  1062. inboundIds []int
  1063. }
  1064. prep := make([]prepared, 0, len(payloads))
  1065. emails := make([]string, 0, len(payloads))
  1066. subIDs := make([]string, 0, len(payloads))
  1067. seenEmail := make(map[string]struct{}, len(payloads))
  1068. seenSubID := make(map[string]string, len(payloads))
  1069. for i := range payloads {
  1070. client := payloads[i].Client
  1071. email := strings.TrimSpace(client.Email)
  1072. if email == "" {
  1073. skip("", "client email is required")
  1074. continue
  1075. }
  1076. if verr := validateClientEmail(email); verr != nil {
  1077. skip(email, verr.Error())
  1078. continue
  1079. }
  1080. if verr := validateClientSubID(client.SubID); verr != nil {
  1081. skip(email, verr.Error())
  1082. continue
  1083. }
  1084. if len(payloads[i].InboundIds) == 0 {
  1085. skip(email, "at least one inbound is required")
  1086. continue
  1087. }
  1088. client.Email = email
  1089. if client.SubID == "" {
  1090. client.SubID = uuid.NewString()
  1091. }
  1092. if !client.Enable {
  1093. client.Enable = true
  1094. }
  1095. now := time.Now().UnixMilli()
  1096. if client.CreatedAt == 0 {
  1097. client.CreatedAt = now
  1098. }
  1099. client.UpdatedAt = now
  1100. le := strings.ToLower(email)
  1101. if _, dup := seenEmail[le]; dup {
  1102. skip(email, "email already in use: "+email)
  1103. continue
  1104. }
  1105. if owner, ok := seenSubID[client.SubID]; ok && owner != le {
  1106. skip(email, "subId already in use: "+client.SubID)
  1107. continue
  1108. }
  1109. seenEmail[le] = struct{}{}
  1110. seenSubID[client.SubID] = le
  1111. prep = append(prep, prepared{client: client, inboundIds: payloads[i].InboundIds})
  1112. emails = append(emails, email)
  1113. subIDs = append(subIDs, client.SubID)
  1114. }
  1115. if len(prep) == 0 {
  1116. return result, false, nil
  1117. }
  1118. db := database.GetDB()
  1119. const lookupChunk = 400
  1120. existingByEmail := make(map[string]model.ClientRecord, len(emails))
  1121. for start := 0; start < len(emails); start += lookupChunk {
  1122. end := min(start+lookupChunk, len(emails))
  1123. var rows []model.ClientRecord
  1124. if e := db.Where("email IN ?", emails[start:end]).Find(&rows).Error; e != nil {
  1125. return result, false, e
  1126. }
  1127. for i := range rows {
  1128. existingByEmail[strings.ToLower(rows[i].Email)] = rows[i]
  1129. }
  1130. }
  1131. existingSubOwner := make(map[string]string, len(subIDs))
  1132. for start := 0; start < len(subIDs); start += lookupChunk {
  1133. end := min(start+lookupChunk, len(subIDs))
  1134. var rows []model.ClientRecord
  1135. if e := db.Where("sub_id IN ?", subIDs[start:end]).Find(&rows).Error; e != nil {
  1136. return result, false, e
  1137. }
  1138. for i := range rows {
  1139. existingSubOwner[rows[i].SubID] = strings.ToLower(rows[i].Email)
  1140. }
  1141. }
  1142. inboundCache := make(map[int]*model.Inbound)
  1143. getIb := func(id int) (*model.Inbound, error) {
  1144. if ib, ok := inboundCache[id]; ok {
  1145. return ib, nil
  1146. }
  1147. ib, e := inboundSvc.GetInbound(id)
  1148. if e != nil {
  1149. return nil, e
  1150. }
  1151. inboundCache[id] = ib
  1152. return ib, nil
  1153. }
  1154. byInbound := make(map[int][]model.Client)
  1155. idxByInbound := make(map[int][]int)
  1156. inboundOrder := make([]int, 0)
  1157. failed := make([]bool, len(prep))
  1158. reason := make([]string, len(prep))
  1159. for idx := range prep {
  1160. le := strings.ToLower(prep[idx].client.Email)
  1161. if rec, ok := existingByEmail[le]; ok {
  1162. if rec.SubID != prep[idx].client.SubID {
  1163. failed[idx] = true
  1164. reason[idx] = "email already in use: " + prep[idx].client.Email
  1165. continue
  1166. }
  1167. if prep[idx].client.ID == "" {
  1168. prep[idx].client.ID = rec.UUID
  1169. }
  1170. if prep[idx].client.Password == "" {
  1171. prep[idx].client.Password = rec.Password
  1172. }
  1173. if prep[idx].client.Auth == "" {
  1174. prep[idx].client.Auth = rec.Auth
  1175. }
  1176. if prep[idx].client.Secret == "" {
  1177. prep[idx].client.Secret = rec.Secret
  1178. }
  1179. }
  1180. if owner, ok := existingSubOwner[prep[idx].client.SubID]; ok && owner != le {
  1181. failed[idx] = true
  1182. reason[idx] = "subId already in use: " + prep[idx].client.SubID
  1183. continue
  1184. }
  1185. ok := true
  1186. for _, ibId := range prep[idx].inboundIds {
  1187. ib, e := getIb(ibId)
  1188. if e != nil {
  1189. failed[idx] = true
  1190. reason[idx] = e.Error()
  1191. ok = false
  1192. break
  1193. }
  1194. if e := s.fillProtocolDefaults(&prep[idx].client, ib); e != nil {
  1195. failed[idx] = true
  1196. reason[idx] = e.Error()
  1197. ok = false
  1198. break
  1199. }
  1200. }
  1201. if !ok {
  1202. continue
  1203. }
  1204. for _, ibId := range prep[idx].inboundIds {
  1205. ib, _ := getIb(ibId)
  1206. if _, seen := byInbound[ibId]; !seen {
  1207. inboundOrder = append(inboundOrder, ibId)
  1208. }
  1209. byInbound[ibId] = append(byInbound[ibId], clientWithInboundFlow(prep[idx].client, ib))
  1210. idxByInbound[ibId] = append(idxByInbound[ibId], idx)
  1211. }
  1212. }
  1213. needRestart := false
  1214. for _, ibId := range inboundOrder {
  1215. payload, e := json.Marshal(map[string][]model.Client{"clients": byInbound[ibId]})
  1216. if e == nil {
  1217. var nr bool
  1218. nr, e = s.addInboundClient(inboundSvc, &model.Inbound{Id: ibId, Settings: string(payload)}, emailSubIDs)
  1219. if e == nil && nr {
  1220. needRestart = true
  1221. }
  1222. }
  1223. if e != nil {
  1224. for _, idx := range idxByInbound[ibId] {
  1225. failed[idx] = true
  1226. if reason[idx] == "" {
  1227. reason[idx] = e.Error()
  1228. }
  1229. }
  1230. }
  1231. }
  1232. for idx := range prep {
  1233. if failed[idx] {
  1234. skip(prep[idx].client.Email, reason[idx])
  1235. } else {
  1236. result.Created++
  1237. }
  1238. }
  1239. return result, needRestart, nil
  1240. }
  1241. func (s *ClientService) DelDepleted(inboundSvc *InboundService) (int, bool, error) {
  1242. db := database.GetDB()
  1243. now := time.Now().UnixMilli()
  1244. depletedClause := "reset = 0 and ((total > 0 and up + down >= total) or (expiry_time > 0 and expiry_time <= ?))"
  1245. var rows []xray.ClientTraffic
  1246. if err := db.Where(depletedClause, now).Find(&rows).Error; err != nil {
  1247. return 0, false, err
  1248. }
  1249. if len(rows) == 0 {
  1250. return 0, false, nil
  1251. }
  1252. seen := make(map[string]struct{}, len(rows))
  1253. emails := make([]string, 0, len(rows))
  1254. for _, r := range rows {
  1255. if r.Email == "" {
  1256. continue
  1257. }
  1258. if _, ok := seen[r.Email]; ok {
  1259. continue
  1260. }
  1261. seen[r.Email] = struct{}{}
  1262. emails = append(emails, r.Email)
  1263. }
  1264. if len(emails) == 0 {
  1265. return 0, false, nil
  1266. }
  1267. res, needRestart, err := s.BulkDelete(inboundSvc, emails, false)
  1268. if err != nil {
  1269. return res.Deleted, needRestart, err
  1270. }
  1271. return res.Deleted, needRestart, nil
  1272. }
  1273. type BulkSetEnableResult struct {
  1274. Changed int `json:"changed"`
  1275. Skipped []BulkSetEnableReport `json:"skipped,omitempty"`
  1276. }
  1277. type BulkSetEnableReport struct {
  1278. Email string `json:"email"`
  1279. Reason string `json:"reason"`
  1280. }
  1281. func (s *ClientService) BulkSetEnable(inboundSvc *InboundService, emails []string, enable bool) (BulkSetEnableResult, bool, error) {
  1282. result := BulkSetEnableResult{}
  1283. seen := map[string]struct{}{}
  1284. cleanEmails := make([]string, 0, len(emails))
  1285. for _, e := range emails {
  1286. e = strings.TrimSpace(e)
  1287. if e == "" {
  1288. continue
  1289. }
  1290. if _, ok := seen[e]; ok {
  1291. continue
  1292. }
  1293. seen[e] = struct{}{}
  1294. cleanEmails = append(cleanEmails, e)
  1295. }
  1296. if len(cleanEmails) == 0 {
  1297. return result, false, nil
  1298. }
  1299. db := database.GetDB()
  1300. var records []model.ClientRecord
  1301. for _, batch := range chunkStrings(cleanEmails, sqlInChunk) {
  1302. var rows []model.ClientRecord
  1303. if err := db.Where("email IN ?", batch).Find(&rows).Error; err != nil {
  1304. return result, false, err
  1305. }
  1306. records = append(records, rows...)
  1307. }
  1308. recordsByEmail := make(map[string]*model.ClientRecord, len(records))
  1309. for i := range records {
  1310. recordsByEmail[records[i].Email] = &records[i]
  1311. }
  1312. skippedReasons := map[string]string{}
  1313. for _, email := range cleanEmails {
  1314. if _, ok := recordsByEmail[email]; !ok {
  1315. skippedReasons[email] = "client not found"
  1316. }
  1317. }
  1318. clientIds := make([]int, 0, len(recordsByEmail))
  1319. recordIdToEmail := make(map[int]string, len(recordsByEmail))
  1320. for _, r := range recordsByEmail {
  1321. clientIds = append(clientIds, r.Id)
  1322. recordIdToEmail[r.Id] = r.Email
  1323. }
  1324. emailsByInbound := map[int][]string{}
  1325. if len(clientIds) > 0 {
  1326. var mappings []model.ClientInbound
  1327. for _, batch := range chunkInts(clientIds, sqlInChunk) {
  1328. var rows []model.ClientInbound
  1329. if err := db.Where("client_id IN ?", batch).Find(&rows).Error; err != nil {
  1330. return result, false, err
  1331. }
  1332. mappings = append(mappings, rows...)
  1333. }
  1334. for _, m := range mappings {
  1335. email, ok := recordIdToEmail[m.ClientId]
  1336. if !ok {
  1337. continue
  1338. }
  1339. emailsByInbound[m.InboundId] = append(emailsByInbound[m.InboundId], email)
  1340. }
  1341. }
  1342. needRestart := false
  1343. for inboundId, ibEmails := range emailsByInbound {
  1344. ibRes := s.bulkSetEnableInboundClients(inboundSvc, inboundId, ibEmails, enable)
  1345. if ibRes.needRestart {
  1346. needRestart = true
  1347. }
  1348. for email, reason := range ibRes.perEmailSkipped {
  1349. if _, already := skippedReasons[email]; !already {
  1350. skippedReasons[email] = reason
  1351. }
  1352. }
  1353. }
  1354. successEmails := make([]string, 0, len(recordsByEmail))
  1355. for email := range recordsByEmail {
  1356. if _, skipped := skippedReasons[email]; skipped {
  1357. continue
  1358. }
  1359. successEmails = append(successEmails, email)
  1360. }
  1361. if len(successEmails) > 0 {
  1362. now := time.Now().UnixMilli()
  1363. if err := runSerializedTx(func(tx *gorm.DB) error {
  1364. for _, batch := range chunkStrings(successEmails, sqlInChunk) {
  1365. if e := tx.Model(xray.ClientTraffic{}).Where("email IN ?", batch).Update("enable", enable).Error; e != nil {
  1366. return e
  1367. }
  1368. if e := tx.Model(&model.ClientRecord{}).Where("email IN ?", batch).
  1369. Updates(map[string]any{"enable": enable, "updated_at": now}).Error; e != nil {
  1370. return e
  1371. }
  1372. }
  1373. return nil
  1374. }); err != nil {
  1375. return result, needRestart, err
  1376. }
  1377. }
  1378. result.Changed = len(successEmails)
  1379. for email, reason := range skippedReasons {
  1380. result.Skipped = append(result.Skipped, BulkSetEnableReport{Email: email, Reason: reason})
  1381. }
  1382. return result, needRestart, nil
  1383. }
  1384. type bulkSetEnableInboundResult struct {
  1385. perEmailSkipped map[string]string
  1386. needRestart bool
  1387. }
  1388. func (s *ClientService) bulkSetEnableInboundClients(inboundSvc *InboundService, inboundId int, emails []string, enable bool) bulkSetEnableInboundResult {
  1389. res := bulkSetEnableInboundResult{perEmailSkipped: map[string]string{}}
  1390. defer lockInbound(inboundId).Unlock()
  1391. oldInbound, err := inboundSvc.GetInbound(inboundId)
  1392. if err != nil {
  1393. for _, e := range emails {
  1394. res.perEmailSkipped[e] = err.Error()
  1395. }
  1396. return res
  1397. }
  1398. var settings map[string]any
  1399. if err := json.Unmarshal([]byte(oldInbound.Settings), &settings); err != nil {
  1400. for _, e := range emails {
  1401. res.perEmailSkipped[e] = err.Error()
  1402. }
  1403. return res
  1404. }
  1405. wanted := make(map[string]struct{}, len(emails))
  1406. for _, email := range emails {
  1407. wanted[email] = struct{}{}
  1408. }
  1409. cipher := ""
  1410. if oldInbound.Protocol == model.Shadowsocks {
  1411. cipher, _ = settings["method"].(string)
  1412. }
  1413. type changedClient struct {
  1414. email string
  1415. wasEnable bool
  1416. client model.Client
  1417. }
  1418. var changed []changedClient
  1419. found := map[string]bool{}
  1420. nowMs := time.Now().UnixMilli()
  1421. interfaceClients, _ := settings["clients"].([]any)
  1422. for i, c := range interfaceClients {
  1423. entry, ok := c.(map[string]any)
  1424. if !ok {
  1425. continue
  1426. }
  1427. email, _ := entry["email"].(string)
  1428. if _, want := wanted[email]; !want || email == "" {
  1429. continue
  1430. }
  1431. found[email] = true
  1432. prev, _ := entry["enable"].(bool)
  1433. if prev == enable {
  1434. continue
  1435. }
  1436. entry["enable"] = enable
  1437. entry["updated_at"] = nowMs
  1438. interfaceClients[i] = entry
  1439. // Build the pushed client from the inbound JSON (the per-inbound source of
  1440. // truth), so a remote UpdateUser carries every field and never zeroes
  1441. // subId/totalGB/expiry from drifting ClientRecord columns (#4628/#4792).
  1442. var parsed model.Client
  1443. if b, mErr := json.Marshal(entry); mErr == nil {
  1444. _ = json.Unmarshal(b, &parsed)
  1445. }
  1446. parsed.Email = email
  1447. parsed.Enable = enable
  1448. changed = append(changed, changedClient{email: email, wasEnable: prev, client: parsed})
  1449. }
  1450. for email := range wanted {
  1451. if !found[email] {
  1452. res.perEmailSkipped[email] = "Client Not Found In Inbound"
  1453. }
  1454. }
  1455. if len(changed) == 0 {
  1456. return res
  1457. }
  1458. settings["clients"] = interfaceClients
  1459. newSettings, err := json.MarshalIndent(settings, "", " ")
  1460. if err != nil {
  1461. for _, ch := range changed {
  1462. res.perEmailSkipped[ch.email] = err.Error()
  1463. }
  1464. return res
  1465. }
  1466. prevSettings := oldInbound.Settings
  1467. oldInbound.Settings = string(newSettings)
  1468. rt, push, _, perr := inboundSvc.nodePushPlan(oldInbound)
  1469. if perr != nil {
  1470. for _, ch := range changed {
  1471. res.perEmailSkipped[ch.email] = perr.Error()
  1472. }
  1473. return res
  1474. }
  1475. if oldInbound.NodeID != nil && push && len(changed) > nodeBulkPushThreshold {
  1476. push = false
  1477. }
  1478. txErr := runSerializedTx(func(tx *gorm.DB) error {
  1479. if e := tx.Save(oldInbound).Error; e != nil {
  1480. return e
  1481. }
  1482. finalClients, gcErr := inboundSvc.GetClients(oldInbound)
  1483. if gcErr != nil {
  1484. return gcErr
  1485. }
  1486. if err := s.SyncInbound(tx, inboundId, finalClients); err != nil {
  1487. return err
  1488. }
  1489. if oldInbound.NodeID != nil {
  1490. return (&NodeService{}).MarkNodeDirtyTx(tx, *oldInbound.NodeID)
  1491. }
  1492. return nil
  1493. })
  1494. if txErr != nil {
  1495. for _, ch := range changed {
  1496. res.perEmailSkipped[ch.email] = txErr.Error()
  1497. }
  1498. return res
  1499. }
  1500. if oldInbound.NodeID == nil {
  1501. if !push {
  1502. res.needRestart = true
  1503. } else {
  1504. for _, ch := range changed {
  1505. if enable {
  1506. err1 := rt.AddUser(context.Background(), oldInbound, map[string]any{
  1507. "email": ch.client.Email,
  1508. "id": ch.client.ID,
  1509. "security": ch.client.Security,
  1510. "flow": ch.client.Flow,
  1511. "auth": ch.client.Auth,
  1512. "password": ch.client.Password,
  1513. "cipher": cipher,
  1514. })
  1515. if err1 != nil {
  1516. logger.Debug("Error in adding client on", rt.Name(), ":", err1)
  1517. res.needRestart = true
  1518. }
  1519. } else if ch.wasEnable {
  1520. err1 := rt.RemoveUser(context.Background(), oldInbound, ch.email)
  1521. if err1 != nil && !strings.Contains(err1.Error(), fmt.Sprintf("User %s not found.", ch.email)) {
  1522. logger.Debug("Error in removing client on", rt.Name(), ":", err1)
  1523. res.needRestart = true
  1524. }
  1525. }
  1526. }
  1527. }
  1528. } else if push {
  1529. pushFailed := false
  1530. for _, ch := range changed {
  1531. updated := ch.client
  1532. updated.UpdatedAt = nowMs
  1533. if err1 := rt.UpdateUser(context.Background(), oldInbound, ch.email, updated); err1 != nil {
  1534. logger.Warning("Error in updating client on", rt.Name(), ":", err1)
  1535. pushFailed = true
  1536. }
  1537. }
  1538. if !pushFailed {
  1539. advancePushedInbound(rt, prevSettings, oldInbound)
  1540. }
  1541. }
  1542. return res
  1543. }