client_locks.go 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234
  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. m, ok := inboundMutationLocks[inboundId]
  25. if !ok {
  26. m = &sync.Mutex{}
  27. inboundMutationLocks[inboundId] = m
  28. }
  29. inboundMutationLocksMu.Unlock()
  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 withdrawClientTombstones(emails ...string) {
  99. if len(emails) == 0 {
  100. return
  101. }
  102. recentlyDeletedMu.Lock()
  103. defer recentlyDeletedMu.Unlock()
  104. for _, email := range emails {
  105. if email != "" {
  106. delete(recentlyDeleted, email)
  107. }
  108. }
  109. }
  110. func tombstoneClientEmails(emails []string) {
  111. if len(emails) == 0 {
  112. return
  113. }
  114. now := time.Now()
  115. cutoff := now.Add(-deleteTombstoneTTL)
  116. recentlyDeletedMu.Lock()
  117. defer recentlyDeletedMu.Unlock()
  118. for _, email := range emails {
  119. if email != "" {
  120. recentlyDeleted[email] = now
  121. }
  122. }
  123. for e, ts := range recentlyDeleted {
  124. if ts.Before(cutoff) {
  125. delete(recentlyDeleted, e)
  126. }
  127. }
  128. }
  129. func isClientEmailTombstoned(email string) bool {
  130. if email == "" {
  131. return false
  132. }
  133. recentlyDeletedMu.Lock()
  134. defer recentlyDeletedMu.Unlock()
  135. ts, ok := recentlyDeleted[email]
  136. if !ok {
  137. return false
  138. }
  139. if time.Since(ts) > deleteTombstoneTTL {
  140. delete(recentlyDeleted, email)
  141. return false
  142. }
  143. return true
  144. }
  145. // dedupeSettingsClients collapses duplicate same-email client entries inside a
  146. // settings JSON blob, keeping the first occurrence. Node snapshots produced by
  147. // builds without the addInboundClient duplicate guard can carry duplicates
  148. // (#5770); adopting them verbatim would copy the duplication into the central
  149. // inbound. Returns the filtered JSON and whether anything was removed.
  150. func dedupeSettingsClients(settings string) (string, bool) {
  151. if settings == "" {
  152. return settings, false
  153. }
  154. var parsed map[string]any
  155. if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
  156. return settings, false
  157. }
  158. clients, _ := parsed["clients"].([]any)
  159. if len(clients) < 2 {
  160. return settings, false
  161. }
  162. seen := make(map[string]struct{}, len(clients))
  163. kept := make([]any, 0, len(clients))
  164. for _, c := range clients {
  165. if cm, ok := c.(map[string]any); ok {
  166. if email, _ := cm["email"].(string); email != "" {
  167. key := strings.ToLower(email)
  168. if _, dup := seen[key]; dup {
  169. continue
  170. }
  171. seen[key] = struct{}{}
  172. }
  173. }
  174. kept = append(kept, c)
  175. }
  176. if len(kept) == len(clients) {
  177. return settings, false
  178. }
  179. parsed["clients"] = kept
  180. b, err := json.MarshalIndent(parsed, "", " ")
  181. if err != nil {
  182. return settings, false
  183. }
  184. return string(b), true
  185. }
  186. // stripTombstonedClients drops just-deleted client entries from a node
  187. // snapshot's settings JSON so adopting a stale snapshot can't re-add them to
  188. // the central inbound while the delete tombstone is live. Returns the filtered
  189. // JSON and whether anything was removed.
  190. func stripTombstonedClients(settings string) (string, bool) {
  191. if settings == "" {
  192. return settings, false
  193. }
  194. var parsed map[string]any
  195. if err := json.Unmarshal([]byte(settings), &parsed); err != nil {
  196. return settings, false
  197. }
  198. clients, _ := parsed["clients"].([]any)
  199. if len(clients) == 0 {
  200. return settings, false
  201. }
  202. kept := make([]any, 0, len(clients))
  203. for _, c := range clients {
  204. if cm, ok := c.(map[string]any); ok {
  205. if email, _ := cm["email"].(string); email != "" && isClientEmailTombstoned(email) {
  206. continue
  207. }
  208. }
  209. kept = append(kept, c)
  210. }
  211. if len(kept) == len(clients) {
  212. return settings, false
  213. }
  214. parsed["clients"] = kept
  215. b, err := json.MarshalIndent(parsed, "", " ")
  216. if err != nil {
  217. return settings, false
  218. }
  219. return string(b), true
  220. }