mirror of
https://github.com/MHSanaei/3x-ui.git
synced 2026-08-09 21:00:58 +00:00
5bc81dfd1d
A client that hit its quota or expiry was disabled, then destroyed on both panels a few seconds later. Five defects fed the same hard delete. ReconcileNode pushed buildRuntimeInboundForAPI, which strips disabled clients. Every other call site targets an in-memory Xray config, where dropping a user is harmless; a node target is a peer panel's DATABASE, so the node deleted the row, stopped reporting it, and the master mirrored that deletion back. Split the builder in two: buildInboundForNodePush injects fallbacks only, buildInboundForLocalRuntime adds the strip on top. The names now say which targets they are safe for. setRemoteTrafficLocked trusted a config_dirty the caller sampled before the snapshot round-trip. A client added inside that window commits on the same serialized writer and marks the node dirty, but the merge still treated the older snapshot as authoritative and deleted it. Re-read the flag inside the writer. In "selected" sync mode, FilterNodeSnapshot strips a deselected tag, but the sweep loaded every inbound with node_id set, so deselecting a tag read as "the node deleted it" and wiped an inbound the node still serves. Skip tags outside the node's managed set. A failed SyncInbound was logged and swallowed; on SQLite the transaction still commits, and the sweep then deleted the innocent clients whose links that failure had left unbuilt. Skip the sweep for such an inbound, and close the trigger: SyncInbound now stores the trimmed email it looks up by, and email validation rejects every unicode space rather than only U+0020. ClientService.Delete tombstones up front and deliberately keeps the record when an inbound fails, so the next attempt can retry the leftovers. The tombstone did not lift with it, so the next merge dropped the client from the synced settings and finished the deletion this path had refused. Add withdrawClientTombstones on every failure path, in BulkDelete too. Finally, make the sweep itself recoverable. "Ended the merge unattached" is true for a real remote deletion and equally true for a bad merge, so it now stamps sync_orphaned_at instead of deleting; any later merge that sees the client attached clears the mark, and a reaper removes only what stayed orphaned past the grace period. The traffic row survives that window too, or a reclaimed client would come back with its usage, quota and expiry reset. The mark is written by this sweep alone, so orphans from any other cause keep their existing manual-cleanup semantics.
253 lines
6.8 KiB
Go
253 lines
6.8 KiB
Go
package service
|
|
|
|
import (
|
|
"strings"
|
|
|
|
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
|
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
|
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// applyClientRecordMerge merges incoming client-record fields onto row using the
|
|
// same rules everywhere a client record is persisted: scalar quota / lifecycle /
|
|
// subscription fields are applied unconditionally (so clearing them takes
|
|
// effect), while credentials and identifiers are only overwritten when the
|
|
// incoming value is non-empty (so a partial update preserves the stored UUID /
|
|
// password / keys). CreatedAt keeps the earliest known value. Email, UpdatedAt,
|
|
// and the Id primary key are intentionally not touched here — callers handle
|
|
// those separately. Shared by SyncInbound (per-inbound persistence) and Update
|
|
// (the no-attached-inbound fallback) so the two paths cannot diverge.
|
|
func applyClientRecordMerge(row *model.ClientRecord, incoming *model.ClientRecord) {
|
|
if incoming.UUID != "" {
|
|
row.UUID = incoming.UUID
|
|
}
|
|
if incoming.Password != "" {
|
|
row.Password = incoming.Password
|
|
}
|
|
if incoming.Auth != "" {
|
|
row.Auth = incoming.Auth
|
|
}
|
|
if incoming.Secret != "" {
|
|
row.Secret = incoming.Secret
|
|
}
|
|
if incoming.AdTag != "" {
|
|
row.AdTag = incoming.AdTag
|
|
}
|
|
row.Flow = incoming.Flow
|
|
if incoming.Security != "" {
|
|
row.Security = incoming.Security
|
|
}
|
|
if incoming.Reverse != "" {
|
|
row.Reverse = incoming.Reverse
|
|
}
|
|
if incoming.PrivateKey != "" {
|
|
row.PrivateKey = incoming.PrivateKey
|
|
}
|
|
if incoming.PublicKey != "" {
|
|
row.PublicKey = incoming.PublicKey
|
|
}
|
|
if incoming.AllowedIPs != "" {
|
|
row.AllowedIPs = incoming.AllowedIPs
|
|
}
|
|
row.PreSharedKey = incoming.PreSharedKey
|
|
row.KeepAlive = incoming.KeepAlive
|
|
row.SubID = incoming.SubID
|
|
row.LimitIP = incoming.LimitIP
|
|
row.TotalGB = incoming.TotalGB
|
|
row.ExpiryTime = incoming.ExpiryTime
|
|
row.Enable = incoming.Enable
|
|
row.TgID = incoming.TgID
|
|
if incoming.Group != "" {
|
|
row.Group = incoming.Group
|
|
}
|
|
row.Comment = incoming.Comment
|
|
row.Reset = incoming.Reset
|
|
if incoming.CreatedAt > 0 && (row.CreatedAt == 0 || incoming.CreatedAt < row.CreatedAt) {
|
|
row.CreatedAt = incoming.CreatedAt
|
|
}
|
|
}
|
|
|
|
func (s *ClientService) SyncInbound(tx *gorm.DB, inboundId int, clients []model.Client) error {
|
|
if tx == nil {
|
|
tx = database.GetDB()
|
|
}
|
|
|
|
if err := tx.Where("inbound_id = ?", inboundId).Delete(&model.ClientInbound{}).Error; err != nil {
|
|
return err
|
|
}
|
|
|
|
emails := make([]string, 0, len(clients))
|
|
seen := make(map[string]struct{}, len(clients))
|
|
for i := range clients {
|
|
email := strings.TrimSpace(clients[i].Email)
|
|
if email == "" {
|
|
continue
|
|
}
|
|
if _, ok := seen[email]; ok {
|
|
continue
|
|
}
|
|
seen[email] = struct{}{}
|
|
emails = append(emails, email)
|
|
}
|
|
|
|
existing := make(map[string]*model.ClientRecord, len(emails))
|
|
const selectChunk = 400
|
|
for start := 0; start < len(emails); start += selectChunk {
|
|
end := min(start+selectChunk, len(emails))
|
|
var rows []model.ClientRecord
|
|
if err := tx.Where("email IN ?", emails[start:end]).Find(&rows).Error; err != nil {
|
|
return err
|
|
}
|
|
for i := range rows {
|
|
r := rows[i]
|
|
existing[r.Email] = &r
|
|
}
|
|
}
|
|
|
|
idByEmail := make(map[string]int, len(emails))
|
|
pending := make(map[string]*model.ClientRecord, len(emails))
|
|
toCreate := make([]*model.ClientRecord, 0, len(emails))
|
|
for i := range clients {
|
|
email := strings.TrimSpace(clients[i].Email)
|
|
if email == "" {
|
|
continue
|
|
}
|
|
|
|
incoming := clients[i].ToRecord()
|
|
// ToRecord copies the raw email; store the trimmed key this function
|
|
// looks up by, or a padded email is inserted and never found again.
|
|
incoming.Email = email
|
|
row, ok := existing[email]
|
|
if !ok {
|
|
if _, dup := pending[email]; !dup {
|
|
pending[email] = incoming
|
|
toCreate = append(toCreate, incoming)
|
|
}
|
|
continue
|
|
}
|
|
|
|
before := *row
|
|
applyClientRecordMerge(row, incoming)
|
|
preservedUpdatedAt := max(incoming.UpdatedAt, row.UpdatedAt)
|
|
row.UpdatedAt = preservedUpdatedAt
|
|
|
|
idByEmail[email] = row.Id
|
|
|
|
if *row == before {
|
|
continue
|
|
}
|
|
if err := tx.Save(row).Error; err != nil {
|
|
return err
|
|
}
|
|
if err := tx.Model(&model.ClientRecord{}).
|
|
Where("id = ?", row.Id).
|
|
UpdateColumn("updated_at", preservedUpdatedAt).Error; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
if len(toCreate) > 0 {
|
|
if err := tx.CreateInBatches(toCreate, 200).Error; err != nil {
|
|
return err
|
|
}
|
|
for _, rec := range toCreate {
|
|
idByEmail[rec.Email] = rec.Id
|
|
}
|
|
}
|
|
|
|
links := make([]model.ClientInbound, 0, len(clients))
|
|
linked := make(map[int]struct{}, len(clients))
|
|
for i := range clients {
|
|
email := strings.TrimSpace(clients[i].Email)
|
|
if email == "" {
|
|
continue
|
|
}
|
|
id, ok := idByEmail[email]
|
|
if !ok {
|
|
continue
|
|
}
|
|
if _, dup := linked[id]; dup {
|
|
continue
|
|
}
|
|
linked[id] = struct{}{}
|
|
links = append(links, model.ClientInbound{
|
|
ClientId: id,
|
|
InboundId: inboundId,
|
|
FlowOverride: clients[i].Flow,
|
|
})
|
|
}
|
|
if len(links) > 0 {
|
|
if err := tx.CreateInBatches(links, 200).Error; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *ClientService) DetachInbound(tx *gorm.DB, inboundId int) error {
|
|
if tx == nil {
|
|
tx = database.GetDB()
|
|
}
|
|
return tx.Where("inbound_id = ?", inboundId).Delete(&model.ClientInbound{}).Error
|
|
}
|
|
|
|
func (s *ClientService) ListForInbound(tx *gorm.DB, inboundId int) ([]model.Client, error) {
|
|
if tx == nil {
|
|
tx = database.GetDB()
|
|
}
|
|
type joinedRow struct {
|
|
model.ClientRecord
|
|
FlowOverride string
|
|
}
|
|
var rows []joinedRow
|
|
err := tx.Table("clients").
|
|
Select("clients.*, client_inbounds.flow_override AS flow_override").
|
|
Joins("JOIN client_inbounds ON client_inbounds.client_id = clients.id").
|
|
Where("client_inbounds.inbound_id = ?", inboundId).
|
|
Order("clients.id ASC").
|
|
Find(&rows).Error
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
out := make([]model.Client, 0, len(rows))
|
|
for i := range rows {
|
|
c := rows[i].ToClient()
|
|
c.Flow = rows[i].FlowOverride
|
|
out = append(out, *c)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ListForInboundBySubId is ListForInbound narrowed to one subscription id —
|
|
// both filter columns are indexed, so the subscription server resolves a
|
|
// subscriber's clients without touching the inbound's settings JSON.
|
|
func (s *ClientService) ListForInboundBySubId(tx *gorm.DB, inboundId int, subId string) ([]model.Client, error) {
|
|
if tx == nil {
|
|
tx = database.GetDB()
|
|
}
|
|
type joinedRow struct {
|
|
model.ClientRecord
|
|
FlowOverride string
|
|
}
|
|
var rows []joinedRow
|
|
err := tx.Table("clients").
|
|
Select("clients.*, client_inbounds.flow_override AS flow_override").
|
|
Joins("JOIN client_inbounds ON client_inbounds.client_id = clients.id").
|
|
Where("client_inbounds.inbound_id = ? AND clients.sub_id = ?", inboundId, subId).
|
|
Order("clients.id ASC").
|
|
Find(&rows).Error
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
out := make([]model.Client, 0, len(rows))
|
|
for i := range rows {
|
|
c := rows[i].ToClient()
|
|
c.Flow = rows[i].FlowOverride
|
|
out = append(out, *c)
|
|
}
|
|
return out, nil
|
|
}
|