client_locks.go 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178
  1. package service
  2. import (
  3. "encoding/json"
  4. "sync"
  5. "time"
  6. "github.com/mhsanaei/3x-ui/v3/internal/database/model"
  7. "github.com/mhsanaei/3x-ui/v3/internal/logger"
  8. "gorm.io/gorm"
  9. )
  10. // Short-lived tombstone of just-deleted client emails so that a node snapshot
  11. // arriving between delete and node-side processing doesn't resurrect them.
  12. var (
  13. recentlyDeletedMu sync.Mutex
  14. recentlyDeleted = map[string]time.Time{}
  15. )
  16. const deleteTombstoneTTL = 90 * time.Second
  17. var (
  18. inboundMutationLocksMu sync.Mutex
  19. inboundMutationLocks = map[int]*sync.Mutex{}
  20. )
  21. func lockInbound(inboundId int) *sync.Mutex {
  22. inboundMutationLocksMu.Lock()
  23. defer inboundMutationLocksMu.Unlock()
  24. m, ok := inboundMutationLocks[inboundId]
  25. if !ok {
  26. m = &sync.Mutex{}
  27. inboundMutationLocks[inboundId] = m
  28. }
  29. m.Lock()
  30. return m
  31. }
  32. func compactOrphans(db *gorm.DB, clients []any) []any {
  33. if len(clients) == 0 {
  34. return clients
  35. }
  36. emails := make([]string, 0, len(clients))
  37. for _, c := range clients {
  38. cm, ok := c.(map[string]any)
  39. if !ok {
  40. continue
  41. }
  42. if e, _ := cm["email"].(string); e != "" {
  43. emails = append(emails, e)
  44. }
  45. }
  46. if len(emails) == 0 {
  47. return clients
  48. }
  49. existing := make(map[string]struct{}, len(emails))
  50. const orphanChunk = 400
  51. for start := 0; start < len(emails); start += orphanChunk {
  52. end := min(start+orphanChunk, len(emails))
  53. var found []string
  54. if err := db.Model(&model.ClientRecord{}).Where("email IN ?", emails[start:end]).Pluck("email", &found).Error; err != nil {
  55. logger.Warning("compactOrphans pluck:", err)
  56. return clients
  57. }
  58. for _, e := range found {
  59. existing[e] = struct{}{}
  60. }
  61. }
  62. if len(existing) == len(emails) {
  63. return clients
  64. }
  65. out := make([]any, 0, len(existing))
  66. for _, c := range clients {
  67. cm, ok := c.(map[string]any)
  68. if !ok {
  69. out = append(out, c)
  70. continue
  71. }
  72. e, _ := cm["email"].(string)
  73. if e == "" {
  74. out = append(out, c)
  75. continue
  76. }
  77. if _, ok := existing[e]; ok {
  78. out = append(out, c)
  79. }
  80. }
  81. return out
  82. }
  83. func tombstoneClientEmail(email string) {
  84. if email == "" {
  85. return
  86. }
  87. recentlyDeletedMu.Lock()
  88. defer recentlyDeletedMu.Unlock()
  89. recentlyDeleted[email] = time.Now()
  90. cutoff := time.Now().Add(-deleteTombstoneTTL)
  91. for e, ts := range recentlyDeleted {
  92. if ts.Before(cutoff) {
  93. delete(recentlyDeleted, e)
  94. }
  95. }
  96. }
  97. func tombstoneClientEmails(emails []string) {
  98. if len(emails) == 0 {
  99. return
  100. }
  101. now := time.Now()
  102. cutoff := now.Add(-deleteTombstoneTTL)
  103. recentlyDeletedMu.Lock()
  104. defer recentlyDeletedMu.Unlock()
  105. for _, email := range emails {
  106. if email != "" {
  107. recentlyDeleted[email] = now
  108. }
  109. }
  110. for e, ts := range recentlyDeleted {
  111. if ts.Before(cutoff) {
  112. delete(recentlyDeleted, e)
  113. }
  114. }
  115. }
  116. func isClientEmailTombstoned(email string) bool {
  117. if email == "" {
  118. return false
  119. }
  120. recentlyDeletedMu.Lock()
  121. defer recentlyDeletedMu.Unlock()
  122. ts, ok := recentlyDeleted[email]
  123. if !ok {
  124. return false
  125. }
  126. if time.Since(ts) > deleteTombstoneTTL {
  127. delete(recentlyDeleted, email)
  128. return false
  129. }
  130. return true
  131. }
  132. // stripTombstonedClients drops just-deleted client entries from a node
  133. // snapshot's settings JSON so adopting a stale snapshot can't re-add them to
  134. // the central inbound while the delete tombstone is live. Returns the filtered
  135. // JSON and whether anything was removed.
  136. func stripTombstonedClients(settings string) (string, bool) {
  137. if settings == "" {
  138. return settings, false
  139. }
  140. var parsed map[string]any
  141. if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
  142. return settings, false
  143. }
  144. clients, _ := parsed["clients"].([]any)
  145. if len(clients) == 0 {
  146. return settings, false
  147. }
  148. kept := make([]any, 0, len(clients))
  149. for _, c := range clients {
  150. if cm, ok := c.(map[string]any); ok {
  151. if email, _ := cm["email"].(string); email != "" && isClientEmailTombstoned(email) {
  152. continue
  153. }
  154. }
  155. kept = append(kept, c)
  156. }
  157. if len(kept) == len(clients) {
  158. return settings, false
  159. }
  160. parsed["clients"] = kept
  161. b, err := json.MarshalIndent(parsed, "", " ")
  162. if err != nil {
  163. return settings, false
  164. }
  165. return string(b), true
  166. }