From a9a4970d9f87d1f925b3c4cbd4130ee899654747 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Sat, 12 Sep 2026 00:08:30 +0800 Subject: [PATCH] fix p2p topic canonicalization and stale subscriptions --- server/db/mysql/adapter.go | 100 +++++++++++++++++++++++++++ server/db/mysql/tenant_adapter.go | 16 ++++- server/pres.go | 17 +---- server/push.go | 12 ++-- server/push/fcm/payload.go | 12 +--- server/push/jpush/push_jpush.go | 6 -- server/push/jpush/push_jpush_test.go | 5 +- server/topic.go | 21 +++--- server/topic_test.go | 25 ++++--- 9 files changed, 148 insertions(+), 66 deletions(-) diff --git a/server/db/mysql/adapter.go b/server/db/mysql/adapter.go index 89dc2f7..620861b 100644 --- a/server/db/mysql/adapter.go +++ b/server/db/mysql/adapter.go @@ -1111,6 +1111,9 @@ func (a *adapter) UpgradeDb() error { if err := a.ensureMessageTopicSeqIndex(); err != nil { return err } + if err := a.ensureTopicNameUniqueIndex(); err != nil { + return err + } if err := a.ensureTopicSeqMatchesMessages(); err != nil { return err } @@ -1139,6 +1142,97 @@ func (a *adapter) ensureMessageTopicSeqIndex() error { return err } +func (a *adapter) ensureTopicNameUniqueIndex() error { + ctx := context.Background() + conn, err := a.db.Connx(ctx) + if err != nil { + return err + } + defer conn.Close() + + tenantScoped, err := identityColumnExists(ctx, conn, "im_topics", "tenant_id") + if err != nil { + return err + } + + var count int + if tenantScoped { + if err = conn.QueryRowContext(ctx, + `SELECT COUNT(*) FROM information_schema.statistics + WHERE table_schema=DATABASE() AND table_name='im_topics' AND index_name='uk_topics_tenant_name' AND non_unique=0`, + ).Scan(&count); err != nil { + return err + } + if count > 0 { + return nil + } + if _, err = conn.ExecContext(ctx, `UPDATE im_topics AS keep_t + JOIN ( + SELECT tenant_id,name,MIN(id) AS keep_id, + MAX(seqid) AS seqid,MAX(delid) AS delid, + MAX(COALESCE(touchedat,'1000-01-01')) AS touchedat, + MAX(updatedat) AS updatedat + FROM im_topics GROUP BY tenant_id,name HAVING COUNT(*)>1 + ) AS dup ON dup.keep_id=keep_t.id + SET keep_t.seqid=GREATEST(keep_t.seqid,dup.seqid), + keep_t.delid=GREATEST(keep_t.delid,dup.delid), + keep_t.touchedat=NULLIF(GREATEST(COALESCE(keep_t.touchedat,'1000-01-01'),dup.touchedat),'1000-01-01'), + keep_t.updatedat=GREATEST(keep_t.updatedat,dup.updatedat)`); err != nil { + return err + } + if _, err = conn.ExecContext(ctx, `DELETE t FROM im_topics AS t + JOIN ( + SELECT tenant_id,name,MIN(id) AS keep_id + FROM im_topics GROUP BY tenant_id,name HAVING COUNT(*)>1 + ) AS dup ON dup.tenant_id=t.tenant_id AND dup.name=t.name AND t.id<>dup.keep_id`); err != nil { + return err + } + _, err = conn.ExecContext(ctx, "ALTER TABLE im_topics ADD UNIQUE INDEX uk_topics_tenant_name(tenant_id, name)") + if isDupe(err) { + return nil + } + return err + } + + if err = conn.QueryRowContext(ctx, + `SELECT COUNT(*) FROM information_schema.statistics + WHERE table_schema=DATABASE() AND table_name='im_topics' AND index_name='topics_name' AND non_unique=0`, + ).Scan(&count); err != nil { + return err + } + if count > 0 { + return nil + } + if _, err = conn.ExecContext(ctx, `UPDATE im_topics AS keep_t + JOIN ( + SELECT name,MIN(id) AS keep_id, + MAX(seqid) AS seqid,MAX(delid) AS delid, + MAX(COALESCE(touchedat,'1000-01-01')) AS touchedat, + MAX(updatedat) AS updatedat + FROM im_topics GROUP BY name HAVING COUNT(*)>1 + ) AS dup ON dup.keep_id=keep_t.id + SET keep_t.seqid=GREATEST(keep_t.seqid,dup.seqid), + keep_t.delid=GREATEST(keep_t.delid,dup.delid), + keep_t.touchedat=NULLIF(GREATEST(COALESCE(keep_t.touchedat,'1000-01-01'),dup.touchedat),'1000-01-01'), + keep_t.updatedat=GREATEST(keep_t.updatedat,dup.updatedat)`); err != nil { + return err + } + if _, err = conn.ExecContext(ctx, `DELETE t FROM im_topics AS t + JOIN ( + SELECT name,MIN(id) AS keep_id + FROM im_topics GROUP BY name HAVING COUNT(*)>1 + ) AS dup ON dup.name=t.name AND t.id<>dup.keep_id`); err != nil { + return err + } + if _, err = conn.ExecContext(ctx, "ALTER TABLE im_topics ADD UNIQUE INDEX topics_name(name)"); err != nil { + if isDupe(err) { + return nil + } + return err + } + return nil +} + func (a *adapter) ensureTopicSeqMatchesMessages() error { ctx := context.Background() conn, err := a.db.Connx(ctx) @@ -3224,6 +3318,7 @@ func (a *adapter) TopicsForUser(uid t.Uid, keepDeleted bool, opts *t.QueryOpt) ( // Fetch subscriptions. Two queries are needed: users table (p2p) and topics table (grp). // Prepare a list of separate subscriptions to users vs topics join := make(map[string]t.Subscription) // Keeping these to make a join with table for .private and .access + p2pResolved := make(map[string]bool) topq := make([]any, 0, 16) usrq := make([]any, 0, 16) for rows.Next() { @@ -3242,6 +3337,7 @@ func (a *adapter) TopicsForUser(uid t.Uid, keepDeleted bool, opts *t.QueryOpt) ( } else if tcat == t.TopicCatP2P { // P2P subscription, find the other user to get user.Public and user.Trusted. uid1, uid2, _ := t.ParseP2P(tname) + p2pResolved[tname] = false if uid1 == uid { usrq = append(usrq, store.DecodeUid(uid2)) sub.SetWith(uid2.UserId()) @@ -3374,6 +3470,7 @@ func (a *adapter) TopicsForUser(uid t.Uid, keepDeleted bool, opts *t.QueryOpt) ( sub.SetDefaultAccess(usr2.Access.Auth, usr2.Access.Anon) sub.SetLastSeenAndUA(usr2.lastSeen(), usr2.UserAgent) join[joinOn] = sub + p2pResolved[joinOn] = true } } if err == nil { @@ -3388,6 +3485,9 @@ func (a *adapter) TopicsForUser(uid t.Uid, keepDeleted bool, opts *t.QueryOpt) ( subs = make([]t.Subscription, 0, len(join)) for _, sub := range join { + if t.GetTopicCat(sub.Topic) == t.TopicCatP2P && !p2pResolved[sub.Topic] { + continue + } subs = append(subs, sub) } diff --git a/server/db/mysql/tenant_adapter.go b/server/db/mysql/tenant_adapter.go index 48a57a1..f95aaaa 100644 --- a/server/db/mysql/tenant_adapter.go +++ b/server/db/mysql/tenant_adapter.go @@ -942,6 +942,7 @@ func (a *adapter) TenantTopicsForUser(id t.TenantID, uid t.Uid, keepDeleted bool return nil, err } join := map[string]t.Subscription{} + p2pResolved := map[string]bool{} var topq, usrq []any for rows.Next() { var sub t.Subscription @@ -956,6 +957,7 @@ func (a *adapter) TenantTopicsForUser(id t.TenantID, uid t.Uid, keepDeleted bool continue case t.TopicCatP2P: uid1, uid2, _ := t.ParseP2P(tname) + p2pResolved[tname] = false if uid1 == uid { usrq = append(usrq, decodeUID(uid2)) sub.SetWith(uid2.UserId()) @@ -1047,6 +1049,7 @@ func (a *adapter) TenantTopicsForUser(id t.TenantID, uid t.Uid, keepDeleted bool sub.SetDefaultAccess(user.Access.Auth, user.Access.Anon) sub.SetLastSeenAndUA(user.lastSeen(), user.UserAgent) join[joinOn] = sub + p2pResolved[joinOn] = true } } err = rows.Err() @@ -1057,6 +1060,9 @@ func (a *adapter) TenantTopicsForUser(id t.TenantID, uid t.Uid, keepDeleted bool } subs := make([]t.Subscription, 0, len(join)) for _, sub := range join { + if t.GetTopicCat(sub.Topic) == t.TopicCatP2P && !p2pResolved[sub.Topic] { + continue + } subs = append(subs, sub) } return common.SelectEarliestUpdatedSubs(subs, opts, a.maxResults), nil @@ -9637,10 +9643,16 @@ func (a *adapter) TenantFriendshipDeletePair(id t.TenantID, uid1, uid2 t.Uid, no if cancel != nil { defer cancel() } - _, err := a.db.ExecContext(ctx, + res, err := a.db.ExecContext(ctx, "UPDATE im_friendships SET status=?,updated_at=? WHERE (user_id=? AND peer_id=?) OR (user_id=? AND peer_id=?)", t.RelationshipNone, now, decodeUID(uid1), decodeUID(uid2), decodeUID(uid2), decodeUID(uid1)) - return err + if err != nil { + return err + } + if affected, err := res.RowsAffected(); err == nil && affected == 0 { + return t.ErrNotFound + } + return nil } func (a *adapter) TenantUserBlockGet(id t.TenantID, blockerID, blockedID t.Uid) (*t.UserBlock, error) { diff --git a/server/pres.go b/server/pres.go index 16cf368..f5c2831 100644 --- a/server/pres.go +++ b/server/pres.go @@ -50,15 +50,8 @@ func (t *Topic) addToPerSubs(topic string, online, enabled bool) { } // TODO: maybe skip loading channel subscriptions. They can's send or receive these notifications anyway. - - if uid1, uid2, err := types.ParseP2P(topic); err == nil { - // If this is a P2P topic, index it by second user's ID - if uid1.UserId() == t.name { - topic = uid2.UserId() - } else { - topic = uid1.UserId() - } - } + // Keep the canonical P2P topic as the key. usr... is accepted only as a + // client input alias, while downlink frames should consistently use p2p.... t.perSubs[topic] = perSubsData{online: online, enabled: enabled} } @@ -544,7 +537,6 @@ func (t *Topic) infoCallSubsOffline(from string, target types.Uid, event string, func presSubsOfflineOffline(tenantID types.TenantID, topic string, cat types.TopicCat, subs []types.Subscription, what string, params *presParams, skipSid string) { - count := 0 original := topic for i := range subs { sub := &subs[i] @@ -554,11 +546,6 @@ func presSubsOfflineOffline(tenantID types.TenantID, topic string, cat types.Top continue } - if cat == types.TopicCatP2P { - original = types.ParseUid(subs[(count+1)%2].User).UserId() - count++ - } - user := types.ParseUid(sub.User).UserId() actor := params.actor target := params.target diff --git a/server/push.go b/server/push.go index 0fa404f..ca35d10 100644 --- a/server/push.go +++ b/server/push.go @@ -87,11 +87,11 @@ func (t *Topic) pushForData(fromUid types.Uid, data *MsgServerData, msgMarkedAsR } func (t *Topic) preparePushForSubReceipt(fromUid types.Uid, now time.Time) *push.Receipt { - // The `Topic` in the push receipt is `t.xoriginal` for group topics, `fromUid` for p2p topics, - // not the t.original(fromUid) because it's the topic name as seen by the recipient, not by the sender. + // P2P push payloads use the canonical p2p... topic. usr... is accepted only + // as an input alias. topic := t.xoriginal if t.cat == types.TopicCatP2P { - topic = fromUid.UserId() + topic = t.name } // Initialize the push receipt. @@ -165,11 +165,11 @@ func pushForChanDelete(tenantID types.TenantID, topicName string, now time.Time) // Prepares payload to be delivered to a mobile device as a push notification in response to receiving "read" notification. func (t *Topic) pushForReadRcpt(uid types.Uid, seq int64, now time.Time) *push.Receipt { - // The `Topic` in the push receipt is `t.xoriginal` for group topics, `fromUid` for p2p topics, - // not the t.original(fromUid) because it's the topic name as seen by the recipient, not by the sender. + // P2P push payloads use the canonical p2p... topic. usr... is accepted only + // as an input alias. topic := t.xoriginal if t.cat == types.TopicCatP2P { - topic = uid.UserId() + topic = t.name } // Initialize the push receipt. diff --git a/server/push/fcm/payload.go b/server/push/fcm/payload.go index b56fc46..41258ab 100644 --- a/server/push/fcm/payload.go +++ b/server/push/fcm/payload.go @@ -143,18 +143,10 @@ func PrepareV1Notifications(rcpt *push.Receipt, config *configType) ([]*fcmv1.Me for uid, devList := range devices { topic := rcpt.Payload.Topic userData := data - tcat := t.GetTopicCat(topic) - if rcpt.To[uid].Delivered > 0 || tcat == t.TopicCatP2P { + if rcpt.To[uid].Delivered > 0 { userData = clonePayload(data) - // Fix topic name for P2P pushes. - if tcat == t.TopicCatP2P { - topic, _ = t.P2PNameForUser(uid, topic) - userData["topic"] = topic - } // Silence the push for user who have received the data interactively. - if rcpt.To[uid].Delivered > 0 { - userData["silent"] = "true" - } + userData["silent"] = "true" } for i := range devList { diff --git a/server/push/jpush/push_jpush.go b/server/push/jpush/push_jpush.go index 012efec..b9a9bbe 100644 --- a/server/push/jpush/push_jpush.go +++ b/server/push/jpush/push_jpush.go @@ -352,12 +352,6 @@ func preparePayloads(rcpt *push.Receipt, config *configType) ([]*jpushPayload, [ } userData := cloneData(data) topic := rcpt.Payload.Topic - if t.GetTopicCat(topic) == t.TopicCatP2P { - if alias, err := t.P2PNameForUser(uid, topic); err == nil { - topic = alias - userData["topic"] = alias - } - } var audience map[string][]string if useAlias { audience = map[string][]string{"alias": {aliasForUid(uid)}} diff --git a/server/push/jpush/push_jpush_test.go b/server/push/jpush/push_jpush_test.go index 9f0e949..f8e4509 100644 --- a/server/push/jpush/push_jpush_test.go +++ b/server/push/jpush/push_jpush_test.go @@ -100,9 +100,8 @@ func TestPreparePayloadsUsesAliasAndNormalizesP2PTopic(t *testing.T) { if got := payload.Audience["alias"][0]; got != aliasForUid(callee) { t.Fatalf("unexpected alias: %s", got) } - userTopic, _ := types.P2PNameForUser(callee, topic) - if got := payload.Notification.Android.Extras["topic"]; got != userTopic { - t.Fatalf("topic was not normalized for callee: got %s want %s", got, userTopic) + if got := payload.Notification.Android.Extras["topic"]; got != topic { + t.Fatalf("topic was not kept canonical for callee: got %s want %s", got, topic) } if payload.Notification.IOS.Badge != "3" { t.Fatalf("unexpected badge: %s", payload.Notification.IOS.Badge) diff --git a/server/topic.go b/server/topic.go index 9d599bc..1039249 100644 --- a/server/topic.go +++ b/server/topic.go @@ -262,8 +262,8 @@ func (t *Topic) prepareBroadcastableMessage(msg *ServerComMessage, uid types.Uid } if (t.cat == types.TopicCatP2P && !uid.IsZero()) || (t.cat == types.TopicCatGrp && t.isChan) { - // For p2p topics topic name is dependent on receiver. - // Channel topics may be presented as grpXXX or chnXXX. + // P2P topics are exposed by their canonical p2p... name. Channel topics + // may be presented as grpXXX or chnXXX. var topicName string if isChanSub { topicName = types.GrpToChn(t.xoriginal) @@ -2867,14 +2867,15 @@ func (t *Topic) replyGetSub(sess *Session, asUid types.Uid, authLevel auth.Level banned = true } - // Reporting user's subscriptions to other topics. P2P topic name is the - // UID of the other user. + // Reporting user's subscriptions to other topics. P2P topic names are + // reported by their canonical p2p... names; usr... remains accepted as + // an input alias only. with := sub.GetWith() canViewLastSeen := true if with != "" { - mts.Topic = with + mts.Topic = sub.Topic canViewLastSeen = imCanViewLastSeen(t.tenantID, asUid, types.ParseUserId(with)) - mts.Online = t.perSubs[with].online && !deleted && presencer + mts.Online = t.perSubs[sub.Topic].online && !deleted && presencer if !deleted && presencer { mts.Online = redisUserOnline(t.tenantID, types.ParseUserId(with), mts.Online) } @@ -4130,8 +4131,8 @@ func (t *Topic) isDeleted() bool { // Get topic name suitable for the given client func (t *Topic) original(uid types.Uid) string { if t.cat == types.TopicCatP2P { - if pud, ok := t.perUser[uid]; ok { - return pud.topicName + if _, ok := t.perUser[uid]; ok { + return t.name } panic("Invalid P2P topic") } @@ -4350,7 +4351,6 @@ func genTopicName() string { } // Convert expanded (routable) topic name into name suitable for sending to the user. -// For example p2pAbCDef123 -> usrAbCDef func topicNameForUser(name string, uid types.Uid, isChan bool) string { switch topicCat(name) { case types.TopicCatMe: @@ -4358,8 +4358,7 @@ func topicNameForUser(name string, uid types.Uid, isChan bool) string { case types.TopicCatFnd: return "fnd" case types.TopicCatP2P: - topic, _ := types.P2PNameForUser(uid, name) - return topic + return name case types.TopicCatGrp: if isChan { return types.GrpToChn(name) diff --git a/server/topic_test.go b/server/topic_test.go index 3b7bf05..bdcfe8d 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -259,8 +259,8 @@ func TestHandleBroadcastDataP2P(t *testing.T) { if r.Data == nil { t.Fatalf("Response[0] must have a ctrl message") } - if r.Data.Topic != from { - t.Errorf("Response[0] topic: expected '%s', got '%s'", from, r.Data.Topic) + if r.Data.Topic != "p2p-test" { + t.Errorf("Response[0] topic: expected '%s', got '%s'", "p2p-test", r.Data.Topic) } if r.Data.Content.(string) != "test" { t.Errorf("Response[0] content: expected 'test', got '%s'", r.Data.Content.(string)) @@ -274,7 +274,7 @@ func TestHandleBroadcastDataP2P(t *testing.T) { if len(helper.hubMessages) != 2 { t.Fatal("Huhelper.route expected exactly two recipients routed via huhelper.") } - for i, uid := range helper.uids { + for _, uid := range helper.uids { if mm, ok := helper.hubMessages[uid.UserId()]; ok { if len(mm) == 1 { s := mm[0] @@ -286,7 +286,7 @@ func TestHandleBroadcastDataP2P(t *testing.T) { if p.SkipTopic != "p2p-test" { t.Errorf("Uid %s: pres skip topic is expected to be 'p2p-test', got %s", uid.UserId(), p.SkipTopic) } - expectedSrc := helper.uids[i^1].UserId() + expectedSrc := "p2p-test" if p.Src != expectedSrc { t.Errorf("Uid %s: pres.src expected: %s, found: %s", uid.UserId(), expectedSrc, p.Src) } @@ -425,8 +425,8 @@ func TestHandleBroadcastCall(t *testing.T) { if r.Data == nil { t.Fatalf("Response[0] must have a ctrl message") } - if r.Data.Topic != from { - t.Errorf("Response[0] topic: expected '%s', got '%s'", from, r.Data.Topic) + if r.Data.Topic != "p2p-test" { + t.Errorf("Response[0] topic: expected '%s', got '%s'", "p2p-test", r.Data.Topic) } content := r.Data.Content.(map[string]any) if content["callId"] != "call-1" || content["callType"] != "audio" || content["video"] != false { @@ -447,7 +447,7 @@ func TestHandleBroadcastCall(t *testing.T) { if len(helper.hubMessages) != 2 { t.Fatal("Huhelper.route expected exactly two recipients routed via huhelper.") } - for i, uid := range helper.uids { + for _, uid := range helper.uids { if mm, ok := helper.hubMessages[uid.UserId()]; ok { if len(mm) == 1 { s := mm[0] @@ -459,7 +459,7 @@ func TestHandleBroadcastCall(t *testing.T) { if p.SkipTopic != "p2p-test" { t.Errorf("Uid %s: pres skip topic is expected to be 'p2p-test', got %s", uid.UserId(), p.SkipTopic) } - expectedSrc := helper.uids[i^1].UserId() + expectedSrc := "p2p-test" if p.Src != expectedSrc { t.Errorf("Uid %s: pres.src expected: %s, found: %s", uid.UserId(), expectedSrc, p.Src) } @@ -1321,9 +1321,8 @@ func TestHandleBroadcastInfoP2P(t *testing.T) { res := helper.results[1].messages[0].(*ServerComMessage) if res.Info != nil { info := res.Info - // Topic name will be fixed (to -> from). - if info.Topic != from.UserId() { - t.Errorf("Info.Topic: expected '%s', found '%s'", to.UserId(), info.Topic) + if info.Topic != topicName { + t.Errorf("Info.Topic: expected '%s', found '%s'", topicName, info.Topic) } if info.From != from.UserId() { t.Errorf("Info.From: expected '%s', found '%s'", from.UserId(), info.From) @@ -1341,9 +1340,9 @@ func TestHandleBroadcastInfoP2P(t *testing.T) { if len(helper.hubMessages) != 2 { t.Fatalf("Hubhelper.route expected exactly two recipients routed via hubhelper. Found %d", len(helper.hubMessages)) } - for i, uid := range helper.uids { + for _, uid := range helper.uids { if routedMsgs, ok := helper.hubMessages[uid.UserId()]; ok { - expectedSrc := helper.uids[i^1].UserId() + expectedSrc := topicName for _, s := range routedMsgs { if s.Info != nil { // Info messages for offline sessions.