client_locks.go 5.2 KB

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