package model import ( "context" "errors" "time" mysqldriver "github.com/go-sql-driver/mysql" "gorm.io/gorm" "spark-server/internal/module/conversation/entity" "spark-server/internal/pkg/errs" ) // Repo is MySQL persistence for conversations / p2p / members. 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("conversation-mysql", err) } func (m *Repo) Get(ctx context.Context, id int64) (*entity.Conversation, error) { db, err := m.session(ctx) if err != nil { return nil, err } var row Conversation if err := db.Where("id = ?", id).Take(&row).Error; err != nil { return nil, mapDBErr(err) } return row.entity(), nil } func (m *Repo) GetP2P(ctx context.Context, userA, userB int64) (*entity.Conversation, error) { db, err := m.session(ctx) if err != nil { return nil, err } small, big := entity.OrderUsers(userA, userB) var row P2PConversation if err := db.Where("user_small = ? AND user_big = ?", small, big).Take(&row).Error; err != nil { return nil, mapDBErr(err) } return m.Get(ctx, row.ConversationID) } func (m *Repo) CreateP2P(ctx context.Context, c *entity.Conversation, p2p *entity.P2P, members []*entity.Member) error { if c == nil || p2p == nil { return errs.ErrInvalidArgument } db, err := m.session(ctx) if err != nil { return err } return db.Transaction(func(tx *gorm.DB) error { conv := Conversation{ ID: c.ID, Type: int(c.Type), LastMsgID: c.LastMsgID, LastMsgAt: msTime(c.LastMsgAt), CreatedAt: msTime(c.LastMsgAt), } if err := tx.Create(&conv).Error; err != nil { return mapDBErr(err) } link := P2PConversation{UserSmall: p2p.UserSmall, UserBig: p2p.UserBig, ConversationID: p2p.ConversationID} if err := tx.Create(&link).Error; err != nil { return mapDBErr(err) } for _, mem := range members { row := ConversationMember{ ConversationID: mem.ConversationID, UserID: mem.UserID, LastReadMsgID: mem.LastReadMsgID, JoinedAt: msTime(mem.JoinedAtMS), } if mem.Muted { row.Muted = 1 } if err := tx.Create(&row).Error; err != nil { return mapDBErr(err) } } return nil }) } func (m *Repo) UpdateLastMsg(ctx context.Context, id, msgID, atMS int64) error { db, err := m.session(ctx) if err != nil { return err } res := db.Model(&Conversation{}).Where("id = ?", id).Updates(map[string]any{ "last_msg_id": msgID, "last_msg_at": msTime(atMS), }) if res.Error != nil { return mapDBErr(res.Error) } if res.RowsAffected == 0 { return errs.ErrNotFound } return nil } func (m *Repo) PurgeUser(ctx context.Context, userID int64) error { if userID <= 0 { return errs.ErrInvalidArgument } db, err := m.session(ctx) if err != nil { return err } return mapDBErr(db.Where("user_id = ?", userID).Delete(&ConversationMember{}).Error) } func (r *Conversation) entity() *entity.Conversation { if r == nil { return nil } return &entity.Conversation{ ID: r.ID, Type: entity.Type(r.Type), LastMsgID: r.LastMsgID, LastMsgAt: r.LastMsgAt.UnixMilli(), } } func msTime(ms int64) time.Time { if ms <= 0 { return time.Now() } return time.UnixMilli(ms) }