package dbsync import ( "encoding/json" "fmt" "os" "path/filepath" "sync" "time" "github.com/google/uuid" ) // FileStore 持久化通道与冲突队列(JSON),不依赖业务库类型。 type FileStore struct { mu sync.Mutex dir string chPath string cfPath string } func NewFileStore(dir string) (*FileStore, error) { if dir == "" { dir = "./data/dbsync" } if err := os.MkdirAll(dir, 0o755); err != nil { return nil, err } return &FileStore{ dir: dir, chPath: filepath.Join(dir, "channels.json"), cfPath: filepath.Join(dir, "conflicts.json"), }, nil } func (s *FileStore) ListChannels() ([]Channel, error) { s.mu.Lock() defer s.mu.Unlock() return s.readChannels() } func (s *FileStore) ListChannelsByTenant(tenantID int64) ([]Channel, error) { s.mu.Lock() defer s.mu.Unlock() list, err := s.readChannels() if err != nil { return nil, err } out := make([]Channel, 0) for _, c := range list { if c.TenantID == tenantID { out = append(out, c) } } return out, nil } func (s *FileStore) GetChannel(id string) (*Channel, error) { s.mu.Lock() defer s.mu.Unlock() list, err := s.readChannels() if err != nil { return nil, err } for i := range list { if list[i].ID == id { cp := list[i] return &cp, nil } } return nil, fmt.Errorf("channel not found") } // GetChannelForTenant 仅返回属于该租户的通道。 func (s *FileStore) GetChannelForTenant(id string, tenantID int64) (*Channel, error) { ch, err := s.GetChannel(id) if err != nil { return nil, err } if ch.TenantID != tenantID { return nil, fmt.Errorf("channel not found") } return ch, nil } func (s *FileStore) SaveChannel(ch Channel) (Channel, error) { s.mu.Lock() defer s.mu.Unlock() list, err := s.readChannels() if err != nil { return ch, err } now := time.Now().UTC() if ch.ID == "" { ch.ID = uuid.NewString() ch.CreatedAt = now } ch.UpdatedAt = now if ch.PollIntervalMS <= 0 { ch.PollIntervalMS = 500 } if ch.Direction == "" { ch.Direction = DirLocalToRemote } if ch.ConflictPolicy == "" { ch.ConflictPolicy = PolicyQueue } if ch.PKColumns == nil { ch.PKColumns = map[string]string{} } found := false for i := range list { if list[i].ID == ch.ID { // 禁止跨租户覆盖;更新时锁定原 tenant_id if list[i].TenantID != 0 && ch.TenantID != 0 && list[i].TenantID != ch.TenantID { return ch, fmt.Errorf("channel belongs to another tenant") } if ch.TenantID == 0 { ch.TenantID = list[i].TenantID } ch.CreatedAt = list[i].CreatedAt ch.Stats = list[i].Stats list[i] = ch found = true break } } if !found { if ch.TenantID <= 0 { return ch, fmt.Errorf("tenant_id required") } list = append(list, ch) } if err := s.writeChannels(list); err != nil { return ch, err } return ch, nil } func (s *FileStore) DeleteChannel(id string) error { s.mu.Lock() defer s.mu.Unlock() list, err := s.readChannels() if err != nil { return err } next := list[:0] for _, c := range list { if c.ID != id { next = append(next, c) } } return s.writeChannels(next) } func (s *FileStore) DeleteChannelForTenant(id string, tenantID int64) error { ch, err := s.GetChannelForTenant(id, tenantID) if err != nil { return err } _ = ch return s.DeleteChannel(id) } func (s *FileStore) PatchStats(id string, fn func(*Channel)) error { s.mu.Lock() defer s.mu.Unlock() list, err := s.readChannels() if err != nil { return err } for i := range list { if list[i].ID == id { fn(&list[i]) list[i].UpdatedAt = time.Now().UTC() return s.writeChannels(list) } } return fmt.Errorf("channel not found") } func (s *FileStore) AddConflict(c Conflict) error { s.mu.Lock() defer s.mu.Unlock() list, err := s.readConflicts() if err != nil { return err } if c.ID == "" { c.ID = uuid.NewString() } if c.CreatedAt.IsZero() { c.CreatedAt = time.Now().UTC() } list = append(list, c) return s.writeConflicts(list) } func (s *FileStore) ListConflicts(unresolvedOnly bool) ([]Conflict, error) { s.mu.Lock() defer s.mu.Unlock() list, err := s.readConflicts() if err != nil { return nil, err } if !unresolvedOnly { return list, nil } out := make([]Conflict, 0) for _, c := range list { if !c.Resolved { out = append(out, c) } } return out, nil } func (s *FileStore) ListConflictsByTenant(tenantID int64, unresolvedOnly bool) ([]Conflict, error) { list, err := s.ListConflicts(unresolvedOnly) if err != nil { return nil, err } out := make([]Conflict, 0) for _, c := range list { if c.TenantID == tenantID { out = append(out, c) } } return out, nil } func (s *FileStore) ResolveConflict(id, resolution string) error { s.mu.Lock() defer s.mu.Unlock() list, err := s.readConflicts() if err != nil { return err } for i := range list { if list[i].ID == id { list[i].Resolved = true list[i].Resolution = resolution return s.writeConflicts(list) } } return fmt.Errorf("conflict not found") } func (s *FileStore) ResolveConflictForTenant(id string, tenantID int64, resolution string) error { s.mu.Lock() defer s.mu.Unlock() list, err := s.readConflicts() if err != nil { return err } for i := range list { if list[i].ID == id { if list[i].TenantID != tenantID { return fmt.Errorf("conflict not found") } list[i].Resolved = true list[i].Resolution = resolution return s.writeConflicts(list) } } return fmt.Errorf("conflict not found") } func (s *FileStore) readChannels() ([]Channel, error) { b, err := os.ReadFile(s.chPath) if err != nil { if os.IsNotExist(err) { return []Channel{}, nil } return nil, err } var list []Channel if len(b) == 0 { return []Channel{}, nil } if err := json.Unmarshal(b, &list); err != nil { return nil, err } return list, nil } func (s *FileStore) writeChannels(list []Channel) error { b, err := json.MarshalIndent(list, "", " ") if err != nil { return err } tmp := s.chPath + ".tmp" if err := os.WriteFile(tmp, b, 0o600); err != nil { return err } return os.Rename(tmp, s.chPath) } func (s *FileStore) readConflicts() ([]Conflict, error) { b, err := os.ReadFile(s.cfPath) if err != nil { if os.IsNotExist(err) { return []Conflict{}, nil } return nil, err } var list []Conflict if len(b) == 0 { return []Conflict{}, nil } if err := json.Unmarshal(b, &list); err != nil { return nil, err } return list, nil } func (s *FileStore) writeConflicts(list []Conflict) error { b, err := json.MarshalIndent(list, "", " ") if err != nil { return err } tmp := s.cfPath + ".tmp" if err := os.WriteFile(tmp, b, 0o600); err != nil { return err } return os.Rename(tmp, s.cfPath) }