client_bulk.go 47 KB

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