feat: add message mutation sync

2026-09-15 15:54:52 +08:00
parent 55afd976c9
commit bc27a5c574
12 changed files with 949 additions and 13 deletions
@@ -1008,6 +1008,21 @@ func (mr *MockTopicsPersistenceInterfaceMockRecorder) OwnerChange(tenantID, topi
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "OwnerChange", reflect.TypeOf((*MockTopicsPersistenceInterface)(nil).OwnerChange), tenantID, topic, newOwner)
 }

+// BumpMutationVersion mocks base method.
+func (m *MockTopicsPersistenceInterface) BumpMutationVersion(tenantID types.TenantID, topic string, at time.Time) (int64, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "BumpMutationVersion", tenantID, topic, at)
+	ret0, _ := ret[0].(int64)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// BumpMutationVersion indicates an expected call of BumpMutationVersion.
+func (mr *MockTopicsPersistenceInterfaceMockRecorder) BumpMutationVersion(tenantID, topic, at interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "BumpMutationVersion", reflect.TypeOf((*MockTopicsPersistenceInterface)(nil).BumpMutationVersion), tenantID, topic, at)
+}
+
 // Update mocks base method.
 func (m *MockTopicsPersistenceInterface) Update(tenantID types.TenantID, topic string, update map[string]any) error {
 	m.ctrl.T.Helper()
@@ -1220,6 +1235,24 @@ func (mr *MockMessagesPersistenceInterfaceMockRecorder) Search(tenantID, topic,
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Search", reflect.TypeOf((*MockMessagesPersistenceInterface)(nil).Search), tenantID, topic, forUser, query, messageType, cursor, limit)
 }

+// Mutations mocks base method.
+func (m *MockMessagesPersistenceInterface) Mutations(tenantID types.TenantID, topic string, forUser types.Uid, sinceVersion int64, limit int) ([]types.MessageMutation, string, bool, int64, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Mutations", tenantID, topic, forUser, sinceVersion, limit)
+	ret0, _ := ret[0].([]types.MessageMutation)
+	ret1, _ := ret[1].(string)
+	ret2, _ := ret[2].(bool)
+	ret3, _ := ret[3].(int64)
+	ret4, _ := ret[4].(error)
+	return ret0, ret1, ret2, ret3, ret4
+}
+
+// Mutations indicates an expected call of Mutations.
+func (mr *MockMessagesPersistenceInterfaceMockRecorder) Mutations(tenantID, topic, forUser, sinceVersion, limit interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Mutations", reflect.TypeOf((*MockMessagesPersistenceInterface)(nil).Mutations), tenantID, topic, forUser, sinceVersion, limit)
+}
+
 // GetOne mocks base method.
 func (m *MockMessagesPersistenceInterface) GetOne(tenantID types.TenantID, topic string, seq int64) (*types.Message, error) {
 	m.ctrl.T.Helper()
@@ -1079,6 +1079,7 @@ type TopicsPersistenceInterface interface {
 	GetSubs(tenantID types.TenantID, topic string, opts *types.QueryOpt) ([]types.Subscription, error)
 	GetSubsAny(tenantID types.TenantID, topic string, opts *types.QueryOpt) ([]types.Subscription, error)
 	Update(tenantID types.TenantID, topic string, update map[string]any) error
+	BumpMutationVersion(tenantID types.TenantID, topic string, at time.Time) (int64, error)
 	UpdateSubCnt(tenantID types.TenantID, topic string) error
 	OwnerChange(tenantID types.TenantID, topic string, newOwner types.Uid) error
 	Delete(tenantID types.TenantID, topic string, isChan, hard bool) error
@@ -1235,6 +1236,18 @@ func (topicsMapper) Update(tenantID types.TenantID, topic string, update map[str
 	return nil
 }

+func (topicsMapper) BumpMutationVersion(tenantID types.TenantID, topic string, at time.Time) (int64, error) {
+	tenantAdp, err := tenantBusinessAdapter(tenantID)
+	if err != nil {
+		return 0, err
+	}
+	version, err := tenantAdp.TenantTopicBumpMutationVersion(tenantID, topic, at)
+	if err == nil {
+		_ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic)
+	}
+	return version, err
+}
+
 // OwnerChange replaces the old topic owner with the new owner.
 func (topicsMapper) OwnerChange(tenantID types.TenantID, topic string, newOwner types.Uid) error {
 	tenantAdp, err := tenantBusinessAdapter(tenantID)
@@ -1386,6 +1399,7 @@ type MessagesPersistenceInterface interface {
 	GetAll(tenantID types.TenantID, topic string, forUser types.Uid, opt *types.QueryOpt) ([]types.Message, error)
 	GetDeleted(tenantID types.TenantID, topic string, forUser types.Uid, opt *types.QueryOpt) ([]types.Range, int, error)
 	Search(tenantID types.TenantID, topic string, forUser types.Uid, query, messageType, cursor string, limit int) ([]types.Message, string, error)
+	Mutations(tenantID types.TenantID, topic string, forUser types.Uid, sinceVersion int64, limit int) ([]types.MessageMutation, string, bool, int64, error)
 }

 // messagesMapper is a concrete type implementing MessagesPersistenceInterface.
@@ -1635,6 +1649,7 @@ type GroupBusinessPersistenceInterface interface {
 	PutAnnouncement(types.TenantID, *types.GroupAnnouncement, int) (*types.GroupAnnouncement, error)
 	ReadAnnouncement(types.TenantID, string, string, types.Uid, time.Time) error
 	ListMutes(types.TenantID, string, string, int) ([]types.GroupMute, string, error)
+	GetMute(types.TenantID, string, types.Uid) (*types.GroupMute, error)
 	UpsertMute(types.TenantID, *types.GroupMute) (*types.GroupMute, error)
 	DeleteMute(types.TenantID, string, types.Uid, types.Uid, time.Time) error
 	ListFiles(types.TenantID, string, string, string, string, int) ([]types.GroupFile, string, error)
@@ -2007,6 +2022,17 @@ func (groupBusinessMapper) ListMutes(tenantID types.TenantID, topic, cursor stri
 	return tenantAdp.TenantGroupMuteList(tenantID, strings.TrimSpace(topic), strings.TrimSpace(cursor), limit)
 }

+func (groupBusinessMapper) GetMute(tenantID types.TenantID, topic string, uid types.Uid) (*types.GroupMute, error) {
+	if strings.TrimSpace(topic) == "" || uid.IsZero() {
+		return nil, types.ErrMalformed
+	}
+	tenantAdp, err := tenantBusinessAdapter(tenantID)
+	if err != nil {
+		return nil, err
+	}
+	return tenantAdp.TenantGroupMuteGet(tenantID, strings.TrimSpace(topic), uid)
+}
+
 func (groupBusinessMapper) UpsertMute(tenantID types.TenantID, mute *types.GroupMute) (*types.GroupMute, error) {
 	if mute == nil || mute.Topic == "" || mute.UserID.IsZero() || mute.MutedBy.IsZero() {
 		return nil, types.ErrMalformed
@@ -4080,6 +4106,20 @@ func (messagesMapper) Search(tenantID types.TenantID, topic string, forUser type
 	return tenantAdp.TenantMessageSearch(tenantID, topic, forUser, query, messageType, cursor, limit)
 }

+func (messagesMapper) Mutations(tenantID types.TenantID, topic string, forUser types.Uid, sinceVersion int64, limit int) ([]types.MessageMutation, string, bool, int64, error) {
+	if topic == "" || forUser.IsZero() || sinceVersion < 0 {
+		return nil, "", false, 0, types.ErrMalformed
+	}
+	if limit <= 0 || limit > 100 {
+		limit = 50
+	}
+	tenantAdp, err := tenantBusinessAdapter(tenantID)
+	if err != nil {
+		return nil, "", false, 0, err
+	}
+	return tenantAdp.TenantMessageMutations(tenantID, topic, forUser, sinceVersion, limit)
+}
+
 // GetDeleted returns the ranges of deleted messages and the largest DelId reported in the list.
 func (messagesMapper) GetDeleted(tenantID types.TenantID, topic string, forUser types.Uid, opt *types.QueryOpt) ([]types.Range, int, error) {
 	tenantAdp, err := tenantBusinessAdapter(tenantID)
@@ -1151,6 +1151,9 @@ type Topic struct {
 	// Count of topic subscribers.
 	SubCnt int

+	// Conversation-level version for in-place message mutations such as recall and edit.
+	MutationVersion int64 `db:"mutation_version" json:"MutationVersion,omitempty" bson:"mutation_version,omitempty"`
+
 	Public  any
 	Trusted any

@@ -1264,6 +1267,16 @@ type MessageRevision struct {
 	Reason    string
 }

+// MessageMutation describes an in-place message change used for offline compensation.
+type MessageMutation struct {
+	SeqId           int64
+	Version         int
+	Event           string
+	Operator        string
+	MutatedAt       time.Time
+	MutationVersion int64
+}
+
 // MessageSaveResult describes the outcome of a message persistence attempt.
 type MessageSaveResult struct {
 	SeqId              int64