client_portable.go 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. package service
  2. import (
  3. "strings"
  4. "time"
  5. "github.com/google/uuid"
  6. "github.com/mhsanaei/3x-ui/v3/internal/database"
  7. "github.com/mhsanaei/3x-ui/v3/internal/database/model"
  8. "github.com/mhsanaei/3x-ui/v3/internal/xray"
  9. "gorm.io/gorm"
  10. )
  11. // ClientPortableTraffic is the client_traffics snapshot carried in export/import.
  12. // model.Client only has the limit (totalGB); usage counters live in this table.
  13. type ClientPortableTraffic struct {
  14. Up int64 `json:"up"`
  15. Down int64 `json:"down"`
  16. ResetCount int `json:"resetCount"`
  17. LastOnline int64 `json:"lastOnline,omitempty"`
  18. LastSubFetch int64 `json:"lastSubFetch,omitempty"`
  19. }
  20. // ExportAll returns every client as {client, inboundIds[, traffic]} for round-trip
  21. // import; orphan clients keep empty inboundIds, and traffic preserves usage (#5858).
  22. func (s *ClientService) ExportAll() ([]ClientCreatePayload, error) {
  23. db := database.GetDB()
  24. var rows []model.ClientRecord
  25. if err := db.Order("id ASC").Find(&rows).Error; err != nil {
  26. return nil, err
  27. }
  28. out := make([]ClientCreatePayload, 0, len(rows))
  29. if len(rows) == 0 {
  30. return out, nil
  31. }
  32. ids := make([]int, 0, len(rows))
  33. emails := make([]string, 0, len(rows))
  34. for i := range rows {
  35. ids = append(ids, rows[i].Id)
  36. if rows[i].Email != "" {
  37. emails = append(emails, rows[i].Email)
  38. }
  39. }
  40. attachments := make(map[int][]int, len(rows))
  41. for _, batch := range chunkInts(ids, sqlInChunk) {
  42. var links []model.ClientInbound
  43. if err := db.Where("client_id IN ?", batch).Order("inbound_id ASC").Find(&links).Error; err != nil {
  44. return nil, err
  45. }
  46. for _, l := range links {
  47. attachments[l.ClientId] = append(attachments[l.ClientId], l.InboundId)
  48. }
  49. }
  50. trafficByEmail := make(map[string]*ClientPortableTraffic, len(emails))
  51. for _, batch := range chunkStrings(emails, sqlInChunk) {
  52. var traffics []xray.ClientTraffic
  53. if err := db.Where("email IN ?", batch).Find(&traffics).Error; err != nil {
  54. return nil, err
  55. }
  56. for i := range traffics {
  57. t := traffics[i]
  58. trafficByEmail[t.Email] = &ClientPortableTraffic{
  59. Up: t.Up,
  60. Down: t.Down,
  61. ResetCount: t.ResetCount,
  62. LastOnline: t.LastOnline,
  63. LastSubFetch: t.LastSubFetch,
  64. }
  65. }
  66. }
  67. for i := range rows {
  68. client := rows[i].ToClient()
  69. // The per-inbound flow_override is the reliable flow for multi-inbound
  70. // clients; the canonical column can be left stale by SyncInbound (#4792).
  71. if flow, err := s.EffectiveFlow(db, rows[i].Id); err == nil && flow != "" {
  72. client.Flow = flow
  73. }
  74. out = append(out, ClientCreatePayload{
  75. Client: *client,
  76. InboundIds: attachments[rows[i].Id],
  77. LimitHwid: rows[i].LimitHwid,
  78. Traffic: trafficByEmail[rows[i].Email],
  79. })
  80. }
  81. return out, nil
  82. }
  83. // ImportClients recreates exported clients; existing emails are Skipped.
  84. // Traffic is applied only for newly created emails so live counters stay intact (#5858).
  85. func (s *ClientService) ImportClients(inboundSvc *InboundService, items []ClientCreatePayload) (BulkCreateResult, bool, error) {
  86. result := BulkCreateResult{}
  87. if len(items) == 0 {
  88. return result, false, nil
  89. }
  90. attached := make([]ClientCreatePayload, 0, len(items))
  91. attachedSrc := make([]int, 0, len(items))
  92. orphans := make([]ClientCreatePayload, 0)
  93. orphanSrc := make([]int, 0)
  94. for i := range items {
  95. if len(items[i].InboundIds) > 0 {
  96. attached = append(attached, items[i])
  97. attachedSrc = append(attachedSrc, i)
  98. } else {
  99. orphans = append(orphans, items[i])
  100. orphanSrc = append(orphanSrc, i)
  101. }
  102. }
  103. inserted := make([]int, 0, len(items))
  104. skip := func(email, reason string) {
  105. if strings.TrimSpace(email) == "" {
  106. email = "(missing email)"
  107. }
  108. result.Skipped = append(result.Skipped, BulkCreateReport{Email: email, Reason: reason})
  109. }
  110. needRestart := false
  111. if len(attached) > 0 {
  112. sub, subInserted, nr, err := s.bulkCreate(inboundSvc, attached)
  113. if err != nil {
  114. return result, needRestart, err
  115. }
  116. needRestart = needRestart || nr
  117. result.Created += sub.Created
  118. result.Skipped = append(result.Skipped, sub.Skipped...)
  119. for _, j := range subInserted {
  120. inserted = append(inserted, attachedSrc[j])
  121. }
  122. }
  123. db := database.GetDB()
  124. for i := range orphans {
  125. client := orphans[i].Client
  126. email := strings.TrimSpace(client.Email)
  127. if email == "" {
  128. skip("", "client email is required")
  129. continue
  130. }
  131. if verr := validateClientEmail(email); verr != nil {
  132. skip(email, verr.Error())
  133. continue
  134. }
  135. if verr := validateClientSubID(client.SubID); verr != nil {
  136. skip(email, verr.Error())
  137. continue
  138. }
  139. if verr := validateClientRenewal(client); verr != nil {
  140. skip(email, verr.Error())
  141. continue
  142. }
  143. if verr := validateClientResetMax(client.ResetMax); verr != nil {
  144. skip(email, verr.Error())
  145. continue
  146. }
  147. if verr := validateClientTrafficReset(client.TrafficReset, client.TrafficResetDay); verr != nil {
  148. skip(email, verr.Error())
  149. continue
  150. }
  151. // An existing record (in the DB or just created from the attached set
  152. // above) always wins — import never clobbers a live client.
  153. var taken int64
  154. if err := db.Model(&model.ClientRecord{}).Where("email = ?", email).Count(&taken).Error; err != nil {
  155. return result, needRestart, err
  156. }
  157. if taken > 0 {
  158. skip(email, "email already in use: "+email)
  159. continue
  160. }
  161. client.Email = email
  162. if client.SubID == "" {
  163. client.SubID = uuid.NewString()
  164. }
  165. if client.SubID != "" {
  166. var subTaken int64
  167. if err := db.Model(&model.ClientRecord{}).
  168. Where("sub_id = ? AND email <> ?", client.SubID, email).
  169. Count(&subTaken).Error; err != nil {
  170. return result, needRestart, err
  171. }
  172. if subTaken > 0 {
  173. skip(email, "subId already in use: "+client.SubID)
  174. continue
  175. }
  176. }
  177. // Preserve exported enable so a disabled orphan stays disabled (#6478).
  178. now := time.Now().UnixMilli()
  179. if client.CreatedAt == 0 {
  180. client.CreatedAt = now
  181. }
  182. client.UpdatedAt = now
  183. rec := client.ToRecord()
  184. rec.LimitHwid = orphans[i].LimitHwid
  185. if err := db.Create(rec).Error; err != nil {
  186. skip(email, err.Error())
  187. continue
  188. }
  189. // gorm default:true drops enable=false on Create — restate (#6478).
  190. if !client.Enable {
  191. if err := db.Model(&model.ClientRecord{}).Where("id = ?", rec.Id).
  192. UpdateColumn("enable", false).Error; err != nil {
  193. return result, needRestart, err
  194. }
  195. }
  196. result.Created++
  197. inserted = append(inserted, orphanSrc[i])
  198. }
  199. if err := applyPortableTraffics(inboundSvc, items, inserted); err != nil {
  200. return result, needRestart, err
  201. }
  202. return result, needRestart, nil
  203. }
  204. // applyPortableTraffics restores counters only for items that inserted a record,
  205. // in batched serialized transactions rather than one writer round-trip per client.
  206. func applyPortableTraffics(inboundSvc *InboundService, items []ClientCreatePayload, inserted []int) error {
  207. const batchSize = 400
  208. withTraffic := make([]int, 0, len(inserted))
  209. for _, i := range inserted {
  210. if items[i].Traffic != nil {
  211. withTraffic = append(withTraffic, i)
  212. }
  213. }
  214. for start := 0; start < len(withTraffic); start += batchSize {
  215. batch := withTraffic[start:min(start+batchSize, len(withTraffic))]
  216. if err := runSerializedTx(func(tx *gorm.DB) error {
  217. emails := make([]string, 0, len(batch))
  218. for _, i := range batch {
  219. if err := applyPortableTraffic(tx, inboundSvc, items[i]); err != nil {
  220. return err
  221. }
  222. emails = append(emails, strings.TrimSpace(items[i].Client.Email))
  223. }
  224. return adjustGroupBaselinesForRestoredTraffic(tx, emails)
  225. }); err != nil {
  226. return err
  227. }
  228. }
  229. return nil
  230. }
  231. // applyPortableTraffic writes the exported counters. Attached clients got their row
  232. // on create; an orphan's row (new, or kept by a keepTraffic delete) is upserted here.
  233. func applyPortableTraffic(tx *gorm.DB, inboundSvc *InboundService, item ClientCreatePayload) error {
  234. client := item.Client
  235. client.Email = strings.TrimSpace(client.Email)
  236. if len(item.InboundIds) == 0 {
  237. if err := inboundSvc.AddClientStat(tx, 0, &client); err != nil {
  238. return err
  239. }
  240. }
  241. return tx.Model(&xray.ClientTraffic{}).Where("email = ?", client.Email).Updates(map[string]any{
  242. "up": item.Traffic.Up,
  243. "down": item.Traffic.Down,
  244. "reset_count": item.Traffic.ResetCount,
  245. "last_online": item.Traffic.LastOnline,
  246. "last_sub_fetch": item.Traffic.LastSubFetch,
  247. }).Error
  248. }
  249. // DeleteOrphans removes every unattached client plus its traffic, IP log, and
  250. // external links in one transaction; returns how many clients were deleted.
  251. func (s *ClientService) DeleteOrphans() (int, error) {
  252. db := database.GetDB()
  253. sub := database.GetDB().Table("client_inbounds").Select("client_id")
  254. var rows []model.ClientRecord
  255. if err := db.Where("id NOT IN (?)", sub).Order("id ASC").Find(&rows).Error; err != nil {
  256. return 0, err
  257. }
  258. if len(rows) == 0 {
  259. return 0, nil
  260. }
  261. ids := make([]int, 0, len(rows))
  262. emails := make([]string, 0, len(rows))
  263. subIDs := make([]string, 0, len(rows))
  264. for i := range rows {
  265. ids = append(ids, rows[i].Id)
  266. if rows[i].Email != "" {
  267. emails = append(emails, rows[i].Email)
  268. }
  269. subIDs = append(subIDs, rows[i].SubID)
  270. }
  271. tombstoneClientEmails(emails)
  272. if err := runSerializedTx(func(tx *gorm.DB) error {
  273. if e := adjustGroupBaselinesForRemovedTraffic(tx, emails); e != nil {
  274. return e
  275. }
  276. if e := clearClientHwidsBySubIDTx(tx, subIDs...); e != nil {
  277. return e
  278. }
  279. for _, batch := range chunkInts(ids, sqlInChunk) {
  280. if e := tx.Where("client_id IN ?", batch).Delete(&model.ClientInbound{}).Error; e != nil {
  281. return e
  282. }
  283. if e := tx.Where("client_id IN ?", batch).Delete(&model.ClientExternalLink{}).Error; e != nil {
  284. return e
  285. }
  286. }
  287. if len(emails) > 0 {
  288. for _, batch := range chunkStrings(emails, sqlInChunk) {
  289. if e := tx.Where("email IN ?", batch).Delete(&xray.ClientTraffic{}).Error; e != nil {
  290. return e
  291. }
  292. if e := tx.Where("client_email IN ?", batch).Delete(&model.InboundClientIps{}).Error; e != nil {
  293. return e
  294. }
  295. }
  296. if e := clearGlobalTraffic(tx, emails...); e != nil {
  297. return e
  298. }
  299. }
  300. for _, batch := range chunkInts(ids, sqlInChunk) {
  301. if e := tx.Where("id IN ?", batch).Delete(&model.ClientRecord{}).Error; e != nil {
  302. return e
  303. }
  304. }
  305. return nil
  306. }); err != nil {
  307. return 0, err
  308. }
  309. return len(ids), nil
  310. }