feat: shard group messages by month

server/store/group_message_runtime.go

@@ -54,6 +54,14 @@ type MessageUniqueClaim struct {
 	Seq    int64
 }
 
+// GroupMessageShardRoute describes the Seq range stored in one monthly table.
+type GroupMessageShardRoute struct {
+	Month  string `db:"month"`
+	MinSeq int64  `db:"min_seq"`
+	MaxSeq int64  `db:"max_seq"`
+	Count  int64  `db:"count"`
+}
+
 type groupFanoutEnvelope struct {
 	OriginNodeID string          `json:"origin_node_id"`
 	Message      json.RawMessage `json:"message"`
@@ -84,10 +92,13 @@ type LoginSession struct {
 type groupMessageRuntime interface {
 	Close() error
 	Claim(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*groupMessageUniqueRecord, bool, error)
-	NextSeq(tenantID types.TenantID, topic string, baseline int64) (int64, error)
+	NextSeq(tenantID types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, error)
 	PeekSeq(tenantID types.TenantID, topic string) (int64, error)
 	BumpSeqFloor(tenantID types.TenantID, topic string, seq int64) error
 	MarkPending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error
+	PutShardRoute(tenantID types.TenantID, topic string, route GroupMessageShardRoute) error
+	GetShardRoutes(tenantID types.TenantID, topic string) ([]GroupMessageShardRoute, bool, error)
+	DeleteShardRoutes(tenantID types.TenantID, topic string) error
 	MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error
 	MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error
 	CacheHotMessage(tenantID types.TenantID, msg *types.Message) error
@@ -132,12 +143,33 @@ func ClaimMessageUnique(tenantID types.TenantID, topic string, from types.Uid, c
 	return &MessageUniqueClaim{Status: rec.Status, Seq: rec.Seq}, claimed, true, err
 }
 
-func ReserveMessageSeq(tenantID types.TenantID, topic string, baseline int64) (int64, bool, error) {
+func ReserveMessageSeq(tenantID types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, bool, error) {
 	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
-		return 0, false, nil
+		return 0, candidateMonth, false, nil
 	}
-	seq, err := nextMessageSeqWithFloor(tenantID, topic, baseline)
-	return seq, true, err
+	seq, month, err := nextMessageSeqWithFloor(tenantID, topic, baseline, candidateMonth)
+	return seq, month, true, err
+}
+
+func PutGroupMessageShardRoute(tenantID types.TenantID, topic string, route GroupMessageShardRoute) error {
+	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || route.Month == "" {
+		return nil
+	}
+	return groupMsgRuntime.PutShardRoute(tenantID, topic, route)
+}
+
+func GetGroupMessageShardRoutes(tenantID types.TenantID, topic string) ([]GroupMessageShardRoute, bool, error) {
+	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
+		return nil, false, nil
+	}
+	return groupMsgRuntime.GetShardRoutes(tenantID, topic)
+}
+
+func DeleteGroupMessageShardRoutes(tenantID types.TenantID, topic string) error {
+	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
+		return nil
+	}
+	return groupMsgRuntime.DeleteShardRoutes(tenantID, topic)
 }
 
 // TopicLastSeqID returns the seq last persisted for the topic in the DB, or 0
@@ -158,7 +190,7 @@ func TopicLastSeqID(tenantID types.TenantID, topic string) int64 {
 // counter key is missing (Redis restart, eviction or flush) it re-seeds from
 // the DB topic seq instead of trusting the caller's in-memory baseline, which
 // may be stale and would otherwise hand out already-persisted seq values.
-func nextMessageSeqWithFloor(tenantID types.TenantID, topic string, baseline int64) (int64, error) {
+func nextMessageSeqWithFloor(tenantID types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, error) {
 	if baseline < 0 {
 		baseline = 0
 	}
@@ -167,7 +199,7 @@ func nextMessageSeqWithFloor(tenantID types.TenantID, topic string, baseline int
 			baseline = floor
 		}
 	}
-	return groupMsgRuntime.NextSeq(tenantID, topic, baseline)
+	return groupMsgRuntime.NextSeq(tenantID, topic, baseline, candidateMonth)
 }
 
 // BumpMessageSeqFloor pushes the Redis seq counter forward to seq (never
@@ -555,24 +587,129 @@ func (r *redisGroupMessageRuntime) Claim(tenantID types.TenantID, topic string,
 	return &current, false, nil
 }
 
-func (r *redisGroupMessageRuntime) NextSeq(tenantID types.TenantID, topic string, baseline int64) (int64, error) {
+func (r *redisGroupMessageRuntime) NextSeq(tenantID types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, error) {
 	if baseline < 0 {
 		baseline = 0
 	}
+	redisNow, err := r.client.Time(context.Background()).Result()
+	if err != nil {
+		return 0, "", err
+	}
+	candidateMonth = redisNow.UTC().Format("200601")
+	if !validShardMonth(candidateMonth) {
+		return 0, "", types.ErrMalformed
+	}
 	seqKey := r.seqKey(tenantID, topic)
-	seq, err := r.client.Eval(context.Background(), `
+	result, err := r.client.Eval(context.Background(), `
 local current = redis.call("GET", KEYS[1])
 if not current then
   redis.call("SET", KEYS[1], ARGV[1])
 elseif tonumber(current) < tonumber(ARGV[1]) then
   redis.call("SET", KEYS[1], ARGV[1])
 end
-return redis.call("INCR", KEYS[1])
-`, []string{seqKey}, baseline).Int64()
+
+local active_month = redis.call("GET", KEYS[2])
+if not active_month or active_month < ARGV[2] then
+  redis.call("SET", KEYS[2], ARGV[2])
+  active_month = ARGV[2]
+end
+
+local seq = redis.call("INCR", KEYS[1])
+return {seq, active_month}
+`, []string{seqKey, r.activeMonthKey(tenantID)}, baseline, candidateMonth).Slice()
 	if err != nil {
-		return 0, err
+		return 0, "", err
 	}
-	return seq, nil
+	if len(result) != 2 {
+		return 0, "", fmt.Errorf("redis group message sequence returned %d values", len(result))
+	}
+	seq, err := strconv.ParseInt(fmt.Sprint(result[0]), 10, 64)
+	if err != nil {
+		return 0, "", err
+	}
+	month := fmt.Sprint(result[1])
+	if !validShardMonth(month) {
+		return 0, "", fmt.Errorf("redis returned invalid active month %q", month)
+	}
+	return seq, month, nil
+}
+
+func (r *redisGroupMessageRuntime) PutShardRoute(tenantID types.TenantID, topic string, route GroupMessageShardRoute) error {
+	if !validShardMonth(route.Month) || route.MinSeq <= 0 || route.MaxSeq < route.MinSeq {
+		return types.ErrMalformed
+	}
+	_, err := r.client.Eval(context.Background(), `
+local min_seq = tonumber(ARGV[2])
+local max_seq = tonumber(ARGV[3])
+local count = tonumber(ARGV[4])
+local meta = redis.call("HGET", KEYS[2], ARGV[1])
+if meta then
+  local old_min, old_max, old_count = string.match(meta, "^(%d+):(%d+):(%d+)$")
+  if old_min then
+    min_seq = math.min(min_seq, tonumber(old_min))
+    max_seq = math.max(max_seq, tonumber(old_max))
+    if count == 1 then
+      count = tonumber(old_count) + 1
+    else
+      count = math.max(count, tonumber(old_count))
+    end
+  end
+end
+redis.call("ZADD", KEYS[1], min_seq, ARGV[1])
+redis.call("HSET", KEYS[2], ARGV[1], min_seq .. ":" .. max_seq .. ":" .. count)
+return 1
+`, []string{r.shardRouteKey(tenantID, topic), r.shardRouteMetaKey(tenantID, topic)},
+		route.Month, route.MinSeq, route.MaxSeq, route.Count).Result()
+	return err
+}
+
+func (r *redisGroupMessageRuntime) GetShardRoutes(tenantID types.TenantID, topic string) ([]GroupMessageShardRoute, bool, error) {
+	ctx := context.Background()
+	months, err := r.client.ZRevRange(ctx, r.shardRouteKey(tenantID, topic), 0, -1).Result()
+	if err != nil {
+		return nil, false, err
+	}
+	if len(months) == 0 {
+		return nil, false, nil
+	}
+	values, err := r.client.HMGet(ctx, r.shardRouteMetaKey(tenantID, topic), months...).Result()
+	if err != nil {
+		return nil, false, err
+	}
+	routes := make([]GroupMessageShardRoute, 0, len(months))
+	for i, month := range months {
+		if values[i] == nil {
+			continue
+		}
+		parts := strings.Split(fmt.Sprint(values[i]), ":")
+		if len(parts) != 3 {
+			continue
+		}
+		minSeq, minErr := strconv.ParseInt(parts[0], 10, 64)
+		maxSeq, maxErr := strconv.ParseInt(parts[1], 10, 64)
+		count, countErr := strconv.ParseInt(parts[2], 10, 64)
+		if minErr == nil && maxErr == nil && countErr == nil && validShardMonth(month) {
+			routes = append(routes, GroupMessageShardRoute{Month: month, MinSeq: minSeq, MaxSeq: maxSeq, Count: count})
+		}
+	}
+	return routes, len(routes) > 0, nil
+}
+
+func (r *redisGroupMessageRuntime) DeleteShardRoutes(tenantID types.TenantID, topic string) error {
+	return r.client.Del(context.Background(), r.shardRouteKey(tenantID, topic), r.shardRouteMetaKey(tenantID, topic)).Err()
+}
+
+func validShardMonth(month string) bool {
+	if len(month) != 6 {
+		return false
+	}
+	for i := range month {
+		if month[i] < '0' || month[i] > '9' {
+			return false
+		}
+	}
+	parsed, err := time.Parse("200601", month)
+	return err == nil && parsed.Format("200601") == month
 }
 
 func (r *redisGroupMessageRuntime) PeekSeq(tenantID types.TenantID, topic string) (int64, error) {
@@ -1182,7 +1319,19 @@ func expandRanges(ranges []types.Range, limit int) []int64 {
 }
 
 func (r *redisGroupMessageRuntime) seqKey(tenantID types.TenantID, topic string) string {
-	return r.key(tenantID, "topic", topic, "last_seq")
+	return r.shardKey(tenantID, "topic", topic, "last_seq")
+}
+
+func (r *redisGroupMessageRuntime) activeMonthKey(tenantID types.TenantID) string {
+	return r.shardKey(tenantID, "groupmsg", "active_month")
+}
+
+func (r *redisGroupMessageRuntime) shardRouteKey(tenantID types.TenantID, topic string) string {
+	return r.shardKey(tenantID, "groupmsg", "route", topic)
+}
+
+func (r *redisGroupMessageRuntime) shardRouteMetaKey(tenantID types.TenantID, topic string) string {
+	return r.shardKey(tenantID, "groupmsg", "route_meta", topic)
 }
 
 func (r *redisGroupMessageRuntime) uniqueKey(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) string {
@@ -1276,3 +1425,11 @@ func (r *redisGroupMessageRuntime) key(tenantID types.TenantID, parts ...string)
 	all = append(all, parts...)
 	return strings.Join(all, ":")
 }
+
+func (r *redisGroupMessageRuntime) shardKey(tenantID types.TenantID, parts ...string) string {
+	prefix := strings.TrimSuffix(r.prefix, ":")
+	slot := "{" + strconv.FormatInt(int64(tenantID), 10) + "}"
+	all := []string{prefix, slot}
+	all = append(all, parts...)
+	return strings.Join(all, ":")
+}

server/store/group_message_runtime_test.go

@@ -35,12 +35,24 @@ func (r *testGroupMessageRuntime) Claim(tenantID types.TenantID, topic string, f
 	return rec, true, nil
 }
 
-func (r *testGroupMessageRuntime) NextSeq(_ types.TenantID, topic string, baseline int64) (int64, error) {
+func (r *testGroupMessageRuntime) NextSeq(_ types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, error) {
 	if r.seqs[topic] < baseline {
 		r.seqs[topic] = baseline
 	}
 	r.seqs[topic]++
-	return r.seqs[topic], nil
+	return r.seqs[topic], candidateMonth, nil
+}
+
+func (r *testGroupMessageRuntime) PutShardRoute(types.TenantID, string, GroupMessageShardRoute) error {
+	return nil
+}
+
+func (r *testGroupMessageRuntime) GetShardRoutes(types.TenantID, string) ([]GroupMessageShardRoute, bool, error) {
+	return nil, false, nil
+}
+
+func (r *testGroupMessageRuntime) DeleteShardRoutes(types.TenantID, string) error {
+	return nil
 }
 
 func (r *testGroupMessageRuntime) PeekSeq(_ types.TenantID, topic string) (int64, error) {
@@ -234,9 +246,9 @@ func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) {
 	if err != nil || !ready || !claimed || claim == nil || claim.Status != "pending" {
 		t.Fatalf("unexpected first claim: claim=%#v claimed=%v ready=%v err=%v", claim, claimed, ready, err)
 	}
-	seq, ready, err := ReserveMessageSeq(tenantID, topic, 184)
-	if err != nil || !ready || seq != 185 {
-		t.Fatalf("unexpected first seq: seq=%d ready=%v err=%v", seq, ready, err)
+	seq, month, ready, err := ReserveMessageSeq(tenantID, topic, 184, "202609")
+	if err != nil || !ready || seq != 185 || month != "202609" {
+		t.Fatalf("unexpected first seq: seq=%d month=%s ready=%v err=%v", seq, month, ready, err)
 	}
 	if err = MarkMessageUniqueDone(tenantID, topic, from, "mid-1", seq, ""); err != nil {
 		t.Fatalf("mark done: %v", err)
@@ -251,9 +263,9 @@ func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) {
 	if err != nil || !ready || !claimed || claim == nil {
 		t.Fatalf("unexpected second claim: claim=%#v claimed=%v ready=%v err=%v", claim, claimed, ready, err)
 	}
-	seq, ready, err = ReserveMessageSeq(tenantID, topic, 184)
-	if err != nil || !ready || seq != 186 {
-		t.Fatalf("unexpected second seq: seq=%d ready=%v err=%v", seq, ready, err)
+	seq, month, ready, err = ReserveMessageSeq(tenantID, topic, 184, "202609")
+	if err != nil || !ready || seq != 186 || month != "202609" {
+		t.Fatalf("unexpected second seq: seq=%d month=%s ready=%v err=%v", seq, month, ready, err)
 	}
 }
 
@@ -266,7 +278,7 @@ func TestBumpMessageSeqFloorNeverMovesBackward(t *testing.T) {
 	tenantID := types.TenantID(7)
 	topic := types.Uid(1).P2PName(types.Uid(2))
 
-	seq, ready, err := ReserveMessageSeq(tenantID, topic, 0)
+	seq, _, ready, err := ReserveMessageSeq(tenantID, topic, 0, "")
 	if err != nil || !ready || seq != 1 {
 		t.Fatalf("unexpected first seq: seq=%d ready=%v err=%v", seq, ready, err)
 	}
@@ -283,7 +295,7 @@ func TestBumpMessageSeqFloorNeverMovesBackward(t *testing.T) {
 		t.Fatalf("floor must never move backward: current=%d err=%v", current, err)
 	}
 	// The next allocation must continue above the bumped floor.
-	if seq, _, err := ReserveMessageSeq(tenantID, topic, 0); err != nil || seq != 6 {
+	if seq, _, _, err := ReserveMessageSeq(tenantID, topic, 0, ""); err != nil || seq != 6 {
 		t.Fatalf("unexpected seq after bump: seq=%d err=%v", seq, err)
 	}
 }

server/store/store.go

@@ -1271,6 +1271,9 @@ func (topicsMapper) Delete(tenantID types.TenantID, topic string, isChan, hard b
 		return err
 	}
 	_ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic)
+	if hard && types.GetTopicCat(topic) == types.TopicCatGrp {
+		_ = DeleteGroupMessageShardRoutes(tenantID, topic)
+	}
 	return nil
 }
 
@@ -3034,9 +3037,8 @@ func refreshUserProfileProjections(tenantID types.TenantID, uid types.Uid, profi
 			if item.Friendship.Topic == "" || item.Friendship.PeerID.IsZero() {
 				continue
 			}
-			if err := Subs.Update(tenantID, item.Friendship.Topic, item.Friendship.PeerID, map[string]any{"Public": public}); err != nil {
-				logs.Warn.Printf("profile %s: refresh subscription %s/%s public failed: %v", uid, item.Friendship.Topic, item.Friendship.PeerID, err)
-			}
+			_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(item.Friendship.Topic, item.Friendship.PeerID, false))
+			_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(item.Friendship.Topic, item.Friendship.PeerID, true))
 		}
 		if nextCursor == "" {
 			break
@@ -3615,6 +3617,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
 		return nil, err
 	}
 	msg.ClientMsgID = clientMsgID
+	msg.InitTimes()
 	fromUid := types.ParseUid(msg.From)
 	isGroupMessage := types.GetTopicCat(msg.Topic) == types.TopicCatGrp
 	if isGroupMessage && groupMsgRuntime != nil {
@@ -3631,7 +3634,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
 				return nil, types.ErrDuplicate
 			}
 		}
-		seq, err := nextMessageSeqWithFloor(tenantID, msg.Topic, msg.SeqId-1)
+		seq, shardMonth, err := nextMessageSeqWithFloor(tenantID, msg.Topic, msg.SeqId-1, msg.CreatedAt.UTC().Format("200601"))
 		if err != nil {
 			if clientMsgID != "" {
 				_ = groupMsgRuntime.MarkFailed(tenantID, msg.Topic, fromUid, clientMsgID)
@@ -3639,6 +3642,10 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
 			return nil, err
 		}
 		msg.SeqId = seq
+		msg.ShardMonth = shardMonth
+	}
+	if isGroupMessage && msg.ShardMonth == "" {
+		msg.ShardMonth = msg.CreatedAt.UTC().Format("200601")
 	}
 	if clientMsgID != "" {
 		var duplicate *types.Message
@@ -3657,6 +3664,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
 			msg.DeletedFor = duplicate.DeletedFor
 			msg.SeqId = duplicate.SeqId
 			msg.Topic = duplicate.Topic
+			msg.ShardMonth = duplicate.ShardMonth
 			msg.From = duplicate.From
 			msg.ClientMsgID = duplicate.ClientMsgID
 			msg.Head = duplicate.Head
@@ -3664,7 +3672,6 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
 			return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil
 		}
 	}
-	msg.InitTimes()
 	msg.SetUid(Store.GetUid())
 	// Reserve topic's or user's SeqId before optional async persistence.
 	err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg)
@@ -3703,6 +3710,7 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag
 				}
 				if duplicate != nil {
 					msg.SeqId = duplicate.SeqId
+					msg.ShardMonth = duplicate.ShardMonth
 					if groupMsgRuntime != nil {
 						_ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, fromUid, clientMsgID, duplicate.SeqId, duplicate.Uid().String())
 					}
@@ -3742,6 +3750,11 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag
 		}
 	}
 	if isGroupMessage && groupMsgRuntime != nil {
+		if err := groupMsgRuntime.PutShardRoute(tenantID, msg.Topic, GroupMessageShardRoute{
+			Month: msg.ShardMonth, MinSeq: msg.SeqId, MaxSeq: msg.SeqId, Count: 1,
+		}); err != nil {
+			logs.Warn.Printf("topic[%s]: failed to update group message shard route: %v", msg.Topic, err)
+		}
 		if err := groupMsgRuntime.CacheHotMessage(tenantID, msg); err != nil {
 			logs.Warn.Printf("topic[%s]: failed to cache hot group message: %v", msg.Topic, err)
 		}
@@ -3804,6 +3817,7 @@ func copyPersistedMessage(dst, src *types.Message) {
 	dst.DeletedFor = src.DeletedFor
 	dst.SeqId = src.SeqId
 	dst.Topic = src.Topic
+	dst.ShardMonth = src.ShardMonth
 	dst.From = src.From
 	dst.ClientMsgID = src.ClientMsgID
 	dst.Head = src.Head
@@ -4052,6 +4066,11 @@ func (messagesMapper) DeleteList(tenantID types.TenantID, topic string, delID in
 		if cacheErr := groupMsgRuntime.InvalidateHotMessages(tenantID, topic, ranges); cacheErr != nil {
 			logs.Warn.Printf("topic[%s]: failed to invalidate deleted group message cache: %v", topic, cacheErr)
 		}
+		if delID == 0 {
+			if routeErr := groupMsgRuntime.DeleteShardRoutes(tenantID, topic); routeErr != nil {
+				logs.Warn.Printf("topic[%s]: failed to delete group message shard route: %v", topic, routeErr)
+			}
+		}
 	}
 
 	// TODO: move to adapter.

server/store/types/types.go

@@ -138,6 +138,10 @@ func (uid *Uid) MarshalJSON() ([]byte, error) {
 // UnmarshalJSON reads Uid from a double quoted string.
 func (uid *Uid) UnmarshalJSON(b []byte) error {
 	size := len(b)
+	if size == 2 && b[0] == '"' && b[1] == '"' {
+		*uid = ZeroUid
+		return nil
+	}
 	if size != (uidBase64Unpadded + 2) {
 		return errors.New("Uid.UnmarshalJSON: invalid length")
 	} else if b[0] != '"' || b[size-1] != '"' {
@@ -1244,6 +1248,7 @@ type Message struct {
 	DeletedFor []SoftDelete `json:"DeletedFor,omitempty" bson:",omitempty"`
 	SeqId      int64
 	Topic      string
+	ShardMonth string `db:"shard_month" json:"-" bson:"shard_month,omitempty"`
 	// Sender's user ID as string (without 'usr' prefix), could be empty.
 	From        string
 	ClientMsgID string     `db:"client_msg_id" json:"ClientMsgId,omitempty" bson:"client_msg_id,omitempty"`

server/store/types/uid_json_test.go

@@ -0,0 +1,23 @@
+package types
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestZeroUidJSONRoundTrip(t *testing.T) {
+	type wrapper struct {
+		UID Uid
+	}
+	raw, err := json.Marshal(&wrapper{UID: ZeroUid})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var decoded wrapper
+	if err = json.Unmarshal(raw, &decoded); err != nil {
+		t.Fatal(err)
+	}
+	if decoded.UID != ZeroUid {
+		t.Fatalf("decoded uid=%v want zero", decoded.UID)
+	}
+}