128 lines
3.2 KiB
Go
128 lines
3.2 KiB
Go
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)
|
|
}
|