client_bulk.go 47 KB

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