spark-server/internal/module/message/model/memory.go

package model

import (
	"context"
	"sync"

	"spark-server/internal/module/message/entity"
	"spark-server/internal/pkg/errs"
)

type clientKey struct {
	SenderID       int64
	ConversationID int64
	ClientMsgID    string
}

type hideKey struct {
	UserID int64
	MsgID  int64
}

// MemoryRepo is an in-process message store for tests and empty-DSN bring-up.
type MemoryRepo struct {
	mu       sync.Mutex
	byID     map[int64]*entity.Message
	byClient map[clientKey]int64
	hidden   map[hideKey]struct{}
}

func NewMemoryRepo() *MemoryRepo {
	return &MemoryRepo{
		byID:     make(map[int64]*entity.Message),
		byClient: make(map[clientKey]int64),
		hidden:   make(map[hideKey]struct{}),
	}
}

func (m *MemoryRepo) clone(row *entity.Message) *entity.Message {
	cp := *row
	if cp.Content != nil {
		cp.Content = append([]byte(nil), row.Content...)
	}
	return &cp
}

func (m *MemoryRepo) Create(_ context.Context, message *entity.Message) error {
	return m.put(message, false)
}

func (m *MemoryRepo) CreateAndBump(_ context.Context, message *entity.Message) error {
	return m.put(message, true)
}

func (m *MemoryRepo) put(message *entity.Message, _ bool) error {
	if message == nil || message.ID <= 0 {
		return errs.ErrInvalidArgument
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	key := clientKey{SenderID: message.SenderID, ConversationID: message.ConversationID, ClientMsgID: message.ClientMsgID}
	if _, ok := m.byID[message.ID]; ok {
		return errs.ErrDuplicate
	}
	if message.ClientMsgID != "" {
		if _, ok := m.byClient[key]; ok {
			return errs.ErrDuplicate
		}
		m.byClient[key] = message.ID
	}
	m.byID[message.ID] = m.clone(message)
	return nil
}

func (m *MemoryRepo) GetByID(_ context.Context, id int64) (*entity.Message, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	row, ok := m.byID[id]
	if !ok {
		return nil, errs.ErrNotFound
	}
	return m.clone(row), nil
}

func (m *MemoryRepo) GetByClientMsgID(_ context.Context, senderID, conversationID int64, clientMsgID string) (*entity.Message, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	id, ok := m.byClient[clientKey{SenderID: senderID, ConversationID: conversationID, ClientMsgID: clientMsgID}]
	if !ok {
		return nil, errs.ErrNotFound
	}
	return m.clone(m.byID[id]), nil
}

func (m *MemoryRepo) ListByConversation(_ context.Context, conversationID, viewerID, lastMsgID int64, limit int) ([]*entity.Message, error) {
	m.mu.Lock()
	defer m.mu.Unlock()
	if limit <= 0 {
		limit = 200
	}
	out := make([]*entity.Message, 0)
	for _, row := range m.byID {
		if row.ConversationID != conversationID || row.ID <= lastMsgID || row.Status == entity.StatusHidden {
			continue
		}
		if _, ok := m.hidden[hideKey{UserID: viewerID, MsgID: row.ID}]; ok {
			continue
		}
		out = append(out, m.clone(row))
	}
	sortMessages(out)
	if len(out) > limit {
		out = out[:limit]
	}
	return out, nil
}

func (m *MemoryRepo) Recall(_ context.Context, id int64) error {
	m.mu.Lock()
	defer m.mu.Unlock()
	row, ok := m.byID[id]
	if !ok || row.Status != entity.StatusNormal {
		return errs.ErrNotFound
	}
	row.Status = entity.StatusRecalled
	return nil
}

func (m *MemoryRepo) HideForUser(_ context.Context, userID int64, msg *entity.Message) error {
	if userID <= 0 || msg == nil || msg.ID <= 0 {
		return errs.ErrInvalidArgument
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	m.hidden[hideKey{UserID: userID, MsgID: msg.ID}] = struct{}{}
	return nil
}

func (m *MemoryRepo) PurgeUser(_ context.Context, userID int64) error {
	if userID <= 0 {
		return errs.ErrInvalidArgument
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	for key := range m.hidden {
		if key.UserID == userID {
			delete(m.hidden, key)
		}
	}
	return nil
}

func (m *MemoryRepo) All() []*entity.Message {
	m.mu.Lock()
	defer m.mu.Unlock()
	out := make([]*entity.Message, 0, len(m.byID))
	for _, row := range m.byID {
		out = append(out, m.clone(row))
	}
	return out
}

func sortMessages(rows []*entity.Message) {
	for i := 1; i < len(rows); i++ {
		j := i
		for j > 0 && rows[j-1].ID > rows[j].ID {
			rows[j-1], rows[j] = rows[j], rows[j-1]
			j--
		}
	}
}