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 private_message and 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("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("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("`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("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 }