package model import ( "context" "errors" "time" mysqldriver "github.com/go-sql-driver/mysql" "gorm.io/gorm" "gorm.io/gorm/clause" "spark-server/internal/module/sync/entity" "spark-server/internal/pkg/errs" ) // Store persists mailbox_events and device_sync_cursors. type Store struct { gdb *gorm.DB ids IDGen } type IDGen interface { NextID() (int64, error) } func NewStore(gdb *gorm.DB, ids IDGen) *Store { return &Store{gdb: gdb, ids: ids} } 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("sync-mysql", err) } func (s *Store) session(ctx context.Context) (*gorm.DB, error) { if s == nil || s.gdb == nil { return nil, errs.ErrAvailable } return s.gdb.WithContext(ctx), nil } func (s *Store) Put(ctx context.Context, userID int64, eventType entity.EventType, conversationID int64, payload []byte) error { if s == nil || s.ids == nil || userID <= 0 || eventType == 0 { return nil } id, err := s.ids.NextID() if err != nil { return err } db, err := s.session(ctx) if err != nil { return err } var conv *int64 if conversationID > 0 { cid := conversationID conv = &cid } now := time.Now() return db.Transaction(func(tx *gorm.DB) error { var maxSeq int64 if err := tx.Model(&MailboxEvent{}).Where("user_id = ? AND device_id = 0", userID). Select("COALESCE(MAX(seq), 0)").Scan(&maxSeq).Error; err != nil { return mapDBErr(err) } row := MailboxEvent{ ID: id, UserID: userID, DeviceID: 0, Seq: maxSeq + 1, EventType: int16(eventType), ConversationID: conv, Payload: payload, CreatedAt: now, } return mapDBErr(tx.Create(&row).Error) }) } func (s *Store) ListSince(ctx context.Context, userID int64, deviceID uint64, seq int64, limit int) ([]*entity.MailboxItem, error) { if s == nil || userID <= 0 { return nil, nil } if limit <= 0 { limit = 200 } db, err := s.session(ctx) if err != nil { return nil, err } var rows []MailboxEvent err = db.Where("user_id = ? AND (device_id = 0 OR device_id = ?) AND seq > ?", userID, int64(deviceID), seq). Order("seq ASC").Limit(limit).Find(&rows).Error if err != nil { return nil, mapDBErr(err) } out := make([]*entity.MailboxItem, 0, len(rows)) for i := range rows { item := &entity.MailboxItem{Seq: rows[i].Seq, EventType: entity.EventType(rows[i].EventType), Payload: rows[i].Payload} if rows[i].ConversationID != nil { item.ConversationID = *rows[i].ConversationID } out = append(out, item) } return out, nil } func (s *Store) SaveCursor(ctx context.Context, userID int64, deviceID uint64, mailboxSeq, dialogScore int64) error { if s == nil || userID <= 0 || deviceID == 0 { return nil } db, err := s.session(ctx) if err != nil { return err } row := DeviceSyncCursor{ UserID: userID, DeviceID: int64(deviceID), MailboxSeq: mailboxSeq, DialogScore: dialogScore, UpdatedAt: time.Now(), } return mapDBErr(db.Clauses(clause.OnConflict{ Columns: []clause.Column{{Name: "user_id"}, {Name: "device_id"}}, DoUpdates: clause.AssignmentColumns([]string{"mailbox_seq", "dialog_score", "updated_at"}), }).Create(&row).Error) } func (s *Store) PurgeUser(ctx context.Context, userID int64) error { if s == nil || userID <= 0 { return nil } db, err := s.session(ctx) if err != nil { return err } if err := db.Where("user_id = ?", userID).Delete(&MailboxEvent{}).Error; err != nil { return mapDBErr(err) } return mapDBErr(db.Where("user_id = ?", userID).Delete(&DeviceSyncCursor{}).Error) }