package model import ( "context" "sync" "time" "spark-server/internal/module/conversation/entity" "spark-server/internal/pkg/errs" ) type p2pKey struct { Small int64 Big int64 } // MemoryP2P is an in-process conversation store for tests and empty-DSN bring-up. type MemoryP2P struct { mu sync.Mutex byID map[int64]*entity.Conversation byP2P map[p2pKey]int64 members map[int64]map[int64]*entity.Member } func NewMemoryP2P() *MemoryP2P { return &MemoryP2P{ byID: make(map[int64]*entity.Conversation), byP2P: make(map[p2pKey]int64), members: make(map[int64]map[int64]*entity.Member), } } func (m *MemoryP2P) GetP2P(_ context.Context, userA, userB int64) (*entity.Conversation, error) { m.mu.Lock() defer m.mu.Unlock() small, big := entity.OrderUsers(userA, userB) id, ok := m.byP2P[p2pKey{Small: small, Big: big}] if !ok { return nil, errs.ErrNotFound } cp := *m.byID[id] return &cp, nil } func (m *MemoryP2P) CreateP2P(_ context.Context, c *entity.Conversation, p2p *entity.P2P, members []*entity.Member) error { if c == nil || p2p == nil { return errs.ErrInvalidArgument } m.mu.Lock() defer m.mu.Unlock() key := p2pKey{Small: p2p.UserSmall, Big: p2p.UserBig} if _, ok := m.byP2P[key]; ok { return errs.ErrDuplicate } if _, ok := m.byID[c.ID]; ok { return errs.ErrDuplicate } cp := *c m.byID[c.ID] = &cp m.byP2P[key] = c.ID byUser := make(map[int64]*entity.Member, len(members)) for _, mem := range members { if mem == nil { continue } row := *mem byUser[mem.UserID] = &row } m.members[c.ID] = byUser return nil } func (m *MemoryP2P) Get(_ context.Context, id int64) (*entity.Conversation, error) { m.mu.Lock() defer m.mu.Unlock() row, ok := m.byID[id] if !ok { return nil, errs.ErrNotFound } cp := *row return &cp, nil } func (m *MemoryP2P) ListMemberIDs(_ context.Context, conversationID int64) ([]int64, error) { m.mu.Lock() defer m.mu.Unlock() byUser, ok := m.members[conversationID] if !ok { return nil, errs.ErrNotFound } ids := make([]int64, 0, len(byUser)) for uid := range byUser { ids = append(ids, uid) } return ids, nil } func (m *MemoryP2P) GetMember(_ context.Context, conversationID, userID int64) (*entity.Member, error) { m.mu.Lock() defer m.mu.Unlock() byUser, ok := m.members[conversationID] if !ok { return nil, errs.ErrNotFound } mem, ok := byUser[userID] if !ok { return nil, errs.ErrNotFound } cp := *mem return &cp, nil } func (m *MemoryP2P) UpdateLastMsg(_ context.Context, id, msgID, atMS int64) error { m.mu.Lock() defer m.mu.Unlock() c, ok := m.byID[id] if !ok { return errs.ErrNotFound } c.LastMsgID = msgID if atMS > 0 { c.LastMsgAt = atMS } else { c.LastMsgAt = time.Now().UnixMilli() } return nil } func (m *MemoryP2P) PurgeUser(_ context.Context, userID int64) error { if userID <= 0 { return errs.ErrInvalidArgument } m.mu.Lock() defer m.mu.Unlock() for _, byUser := range m.members { delete(byUser, userID) } return nil }