lan
6bf87fec46
Introduce several performance and synchronization enhancements: - Implement `SeqBufferManager` to allow sequence number pre-allocation via Redis Lua scripts, reducing atomic increment overhead. - Add `PushWorker` to handle asynchronous message pushing using Redis Streams. - Implement incremental conversation synchronization via `ConversationVersionLog` to allow clients to fetch only recent changes. - Add support for Gzip compression in WebSocket communications to reduce bandwidth usage. - Update dependency injection and configuration to support these new components.
58 lines
1.9 KiB
Go
58 lines
1.9 KiB
Go
package repository
|
|
|
|
import (
|
|
"context"
|
|
|
|
"with_you/internal/model"
|
|
|
|
"gorm.io/gorm"
|
|
)
|
|
|
|
// ConversationVersionLogRepository 会话版本日志仓储接口
|
|
type ConversationVersionLogRepository interface {
|
|
Create(ctx context.Context, log *model.ConversationVersionLog) error
|
|
BatchCreate(ctx context.Context, logs []*model.ConversationVersionLog) error
|
|
GetByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) ([]*model.ConversationVersionLog, error)
|
|
GetMaxVersion(ctx context.Context, userID string) (int64, error)
|
|
}
|
|
|
|
// conversationVersionLogRepository GORM 实现
|
|
type conversationVersionLogRepository struct {
|
|
db *gorm.DB
|
|
}
|
|
|
|
// NewConversationVersionLogRepository 创建会话版本日志仓储
|
|
func NewConversationVersionLogRepository(db *gorm.DB) ConversationVersionLogRepository {
|
|
return &conversationVersionLogRepository{db: db}
|
|
}
|
|
|
|
func (r *conversationVersionLogRepository) Create(ctx context.Context, log *model.ConversationVersionLog) error {
|
|
return r.db.WithContext(ctx).Create(log).Error
|
|
}
|
|
|
|
func (r *conversationVersionLogRepository) BatchCreate(ctx context.Context, logs []*model.ConversationVersionLog) error {
|
|
if len(logs) == 0 {
|
|
return nil
|
|
}
|
|
return r.db.WithContext(ctx).CreateInBatches(logs, 100).Error
|
|
}
|
|
|
|
func (r *conversationVersionLogRepository) GetByVersion(ctx context.Context, userID string, sinceVersion int64, limit int) ([]*model.ConversationVersionLog, error) {
|
|
var logs []*model.ConversationVersionLog
|
|
err := r.db.WithContext(ctx).
|
|
Where("user_id = ? AND version > ?", userID, sinceVersion).
|
|
Order("version ASC").
|
|
Limit(limit).
|
|
Find(&logs).Error
|
|
return logs, err
|
|
}
|
|
|
|
func (r *conversationVersionLogRepository) GetMaxVersion(ctx context.Context, userID string) (int64, error) {
|
|
var maxVersion int64
|
|
err := r.db.WithContext(ctx).
|
|
Model(&model.ConversationVersionLog{}).
|
|
Where("user_id = ?", userID).
|
|
Select("COALESCE(MAX(version), 0)").
|
|
Scan(&maxVersion).Error
|
|
return maxVersion, err
|
|
} |