fix p2p topic canonicalization and stale subscriptions

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)
 	}
 

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) {

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

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.

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 {

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)}}

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)

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)

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.