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

package model

import (
	"context"
	"errors"
	"time"

	mysqldriver "github.com/go-sql-driver/mysql"
	"gorm.io/gorm"

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

// Repo is MySQL persistence for t_private_message and t_group_message.
type Repo struct {
	gdb *gorm.DB
}

func NewRepo(gdb *gorm.DB) *Repo {
	return &Repo{gdb: gdb}
}

func (m *Repo) session(ctx context.Context) (*gorm.DB, error) {
	if m == nil || m.gdb == nil {
		return nil, errs.ErrAvailable
	}
	return m.gdb.WithContext(ctx), nil
}

func mapDBErr(err error) error {
	if err == nil {
		return nil
	}
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return errs.ErrNotFound
	}
	var me *mysqldriver.MySQLError
	if errors.As(err, &me) && me.Number == 1062 {
		return errs.ErrDuplicate
	}
	return errs.ErrInner.WithError("message-mysql", err)
}

func (m *Repo) lookupKind(db *gorm.DB, conversationID int64) (entity.Kind, error) {
	var row struct{ Type int }
	if err := db.Table("t_conversations").Select("type").Where("id = ?", conversationID).Take(&row).Error; err != nil {
		return 0, mapDBErr(err)
	}
	if conventity.Type(row.Type) == conventity.TypeGroup {
		return entity.KindGroup, nil
	}
	return entity.KindPrivate, nil
}

func (m *Repo) takeByID(db *gorm.DB, table string, kind entity.Kind, id int64) (*entity.Message, error) {
	if kind == entity.KindGroup {
		var row GroupMessage
		if err := db.Table(table).Where("id = ?", id).Take(&row).Error; err != nil {
			return nil, mapDBErr(err)
		}
		return row.entity(), nil
	}
	var row PrivateMessage
	if err := db.Table(table).Where("id = ?", id).Take(&row).Error; err != nil {
		return nil, mapDBErr(err)
	}
	return row.entity(), nil
}

func (m *Repo) GetByID(ctx context.Context, id int64) (*entity.Message, error) {
	if id <= 0 {
		return nil, errs.ErrInvalidArgument
	}
	db, err := m.session(ctx)
	if err != nil {
		return nil, err
	}
	msg, err := m.takeByID(db, tablePrivateMessage, entity.KindPrivate, id)
	if err == nil || !errors.Is(err, errs.ErrNotFound) {
		return msg, err
	}
	return m.takeByID(db, tableGroupMessage, entity.KindGroup, id)
}

func (m *Repo) Create(ctx context.Context, message *entity.Message) error {
	if message == nil || message.ID <= 0 || message.ConversationID <= 0 {
		return errs.ErrInvalidArgument
	}
	db, err := m.session(ctx)
	if err != nil {
		return err
	}
	kind, err := m.lookupKind(db, message.ConversationID)
	if err != nil {
		return err
	}
	message.Kind = kind
	row := rowFrom(message)
	stampCreatedAt(message, &row)
	return m.insertBody(db, message, row)
}

func (m *Repo) GetByClientMsgID(ctx context.Context, senderID, conversationID int64, clientMsgID string) (*entity.Message, error) {
	if senderID <= 0 || conversationID <= 0 || clientMsgID == "" {
		return nil, errs.ErrInvalidArgument
	}
	db, err := m.session(ctx)
	if err != nil {
		return nil, err
	}
	kind, err := m.lookupKind(db, conversationID)
	if err != nil {
		return nil, err
	}
	return m.takeByClient(db, kind, senderID, conversationID, clientMsgID)
}

func (m *Repo) takeByClient(db *gorm.DB, kind entity.Kind, senderID, conversationID int64, clientMsgID string) (*entity.Message, error) {
	table := tableOf(kind)
	if kind == entity.KindGroup {
		var row GroupMessage
		err := db.Table(table).Where("sender_id = ? AND conversation_id = ? AND client_msg_id = ?", senderID, conversationID, clientMsgID).Take(&row).Error
		if err != nil {
			return nil, mapDBErr(err)
		}
		return row.entity(), nil
	}
	var row PrivateMessage
	err := db.Table(table).Where("sender_id = ? AND conversation_id = ? AND client_msg_id = ?", senderID, conversationID, clientMsgID).Take(&row).Error
	if err != nil {
		return nil, mapDBErr(err)
	}
	return row.entity(), nil
}

func (m *Repo) CreateAndBump(ctx context.Context, message *entity.Message) error {
	if message == nil || message.ID <= 0 || message.ConversationID <= 0 {
		return errs.ErrInvalidArgument
	}
	db, err := m.session(ctx)
	if err != nil {
		return err
	}
	row := rowFrom(message)
	stampCreatedAt(message, &row)
	return db.Transaction(func(tx *gorm.DB) error {
		kind, err := m.lookupKind(tx, message.ConversationID)
		if err != nil {
			return err
		}
		message.Kind = kind
		if err := m.insertBody(tx, message, row); err != nil {
			return err
		}
		res := tx.Table("t_conversations").Where("id = ?", message.ConversationID).Updates(map[string]any{
			"last_msg_id": message.ID,
			"last_msg_at": row.CreatedAt,
		})
		if res.Error != nil {
			return mapDBErr(res.Error)
		}
		if res.RowsAffected == 0 {
			return errs.ErrNotFound
		}
		return nil
	})
}

func stampCreatedAt(message *entity.Message, row *MessageRow) {
	if message.CreatedAtMS > 0 {
		row.CreatedAt = time.UnixMilli(message.CreatedAtMS)
		return
	}
	row.CreatedAt = time.Now()
	message.CreatedAtMS = row.CreatedAt.UnixMilli()
}

func (m *Repo) insertBody(db *gorm.DB, message *entity.Message, row MessageRow) error {
	if message.Kind == entity.KindGroup {
		if err := m.fillGroupID(db, message); err != nil {
			return err
		}
		rec := groupRow(message, row)
		return mapDBErr(db.Create(&rec).Error)
	}
	if err := m.fillReceiverID(db, message); err != nil {
		return err
	}
	rec := privateRow(message, row)
	return mapDBErr(db.Create(&rec).Error)
}

func (m *Repo) fillGroupID(db *gorm.DB, message *entity.Message) error {
	if message.GroupID > 0 {
		return nil
	}
	var gid int64
	if err := db.Table("`t_groups`").Select("id").Where("conversation_id = ?", message.ConversationID).Scan(&gid).Error; err != nil {
		return mapDBErr(err)
	}
	if gid <= 0 {
		return errs.ErrNotFound
	}
	message.GroupID = gid
	return nil
}

func (m *Repo) fillReceiverID(db *gorm.DB, message *entity.Message) error {
	if message.ReceiverID > 0 || message.SenderID <= 0 {
		return nil
	}
	var peer int64
	err := db.Table("t_conversation_members").Select("user_id").
		Where("conversation_id = ? AND user_id <> ?", message.ConversationID, message.SenderID).
		Limit(1).Scan(&peer).Error
	if err != nil {
		return mapDBErr(err)
	}
	message.ReceiverID = peer
	return nil
}