Use personal topic for push notification registration

One of the issues we noticed is that the partitioned topic
in push notification is heavy in traffic, as any user using a particular
mailserver will use that partitioned topic to register for PNs.

This commit moves from the partitioned topic to the personal topic of
the PN server, so it does not clash with other users that might happen
to have the same partitioned topic as the mailserver, resulting in long
sync times.

Another issue that will need to be addressed separately is that once you
send a message to a topic, because of the way how waku/whisper works,
you will have to register to that topic, meaning that you will receive
that data. Currently waku does not support unsubscribing from a topic
without logging in and out, so that needs also to be addressed.

VERSION

@@ -1 +1 @@
-0.68.8
+0.68.9

appdatabase/migrations/bindata.go

@@ -736,7 +736,7 @@ func _0018_profile_pictures_visibilityUpSql() (*asset, error) {
 		return nil, err
 	}
 
-	info := bindataFileInfo{name: "0018_profile_pictures_visibility.up.sql", size: 84, mode: os.FileMode(0644), modTime: time.Unix(1610959528, 0)}
+	info := bindataFileInfo{name: "0018_profile_pictures_visibility.up.sql", size: 84, mode: os.FileMode(0644), modTime: time.Unix(1610966154, 0)}
 	a := &asset{bytes: bytes, info: info, digest: [32]uint8{0xc9, 0xe3, 0xc5, 0xec, 0x83, 0x55, 0x45, 0x57, 0x7a, 0xaa, 0xd2, 0xa7, 0x59, 0xa7, 0x87, 0xef, 0x63, 0x19, 0x9c, 0x46, 0x9c, 0xc5, 0x32, 0x89, 0xa4, 0x68, 0x70, 0xd8, 0x83, 0x43, 0xa4, 0x72}}
 	return a, nil
 }

protocol/common/message_processor.go

@@ -140,8 +140,8 @@ func (p *MessageProcessor) SendPrivate(
 	// Currently we don't support sending through datasync and setting custom waku fields,
 	// as the datasync interface is not rich enough to propagate that information, so we
 	// would have to add some complexity to handle this.
-	if rawMessage.ResendAutomatically && (rawMessage.Sender != nil || rawMessage.SkipEncryption) {
-		return nil, errors.New("setting identity, skip-encryption and datasync not supported")
+	if rawMessage.ResendAutomatically && (rawMessage.Sender != nil || rawMessage.SkipEncryption || rawMessage.SendOnPersonalTopic) {
+		return nil, errors.New("setting identity, skip-encryption or personal topic and datasync not supported")
 	}
 
 	// Set sender identity if not specified
@@ -216,7 +216,7 @@ func (p *MessageProcessor) sendPrivate(
 	} else if rawMessage.SkipEncryption {
 		// When SkipEncryption is set we don't pass the message to the encryption layer
 		messageIDs := [][]byte{messageID}
-		hash, newMessage, err := p.sendPrivateRawMessage(ctx, recipient, wrappedMessage, messageIDs)
+		hash, newMessage, err := p.sendPrivateRawMessage(ctx, rawMessage, recipient, wrappedMessage, messageIDs)
 		if err != nil {
 			p.logger.Error("failed to send a private message", zap.Error(err))
 			return nil, errors.Wrap(err, "failed to send a message spec")
@@ -569,15 +569,21 @@ func (p *MessageProcessor) sendDataSync(ctx context.Context, publicKey *ecdsa.Pu
 }
 
 // sendPrivateRawMessage sends a message not wrapped in an encryption layer
-func (p *MessageProcessor) sendPrivateRawMessage(ctx context.Context, publicKey *ecdsa.PublicKey, payload []byte, messageIDs [][]byte) ([]byte, *types.NewMessage, error) {
+func (p *MessageProcessor) sendPrivateRawMessage(ctx context.Context, rawMessage *RawMessage, publicKey *ecdsa.PublicKey, payload []byte, messageIDs [][]byte) ([]byte, *types.NewMessage, error) {
 	newMessage := &types.NewMessage{
 		TTL:       whisperTTL,
 		Payload:   payload,
 		PowTarget: calculatePoW(payload),
 		PowTime:   whisperPoWTime,
 	}
+	var hash []byte
+	var err error
 
-	hash, err := p.transport.SendPrivateWithPartitioned(ctx, newMessage, publicKey)
+	if rawMessage.SendOnPersonalTopic {
+		hash, err = p.transport.SendPrivateOnPersonalTopic(ctx, newMessage, publicKey)
+	} else {
+		hash, err = p.transport.SendPrivateWithPartitioned(ctx, newMessage, publicKey)
+	}
 	if err != nil {
 		return nil, nil, err
 	}

protocol/common/raw_message.go

@@ -22,4 +22,5 @@ type RawMessage struct {
 	Sender               *ecdsa.PrivateKey
 	Recipients           []*ecdsa.PublicKey
 	SkipGroupMessageWrap bool
+	SendOnPersonalTopic  bool
 }

protocol/migrations/migrations.go

@@ -22,6 +22,7 @@
 // 1605075346_add_communities.up.sql (6.971kB)
 // 1610117927_add_message_cache.up.sql (142B)
 // 1610959908_add_dont_wrap_to_raw_messages.up.sql (83B)
+// 1610960912_add_send_on_personal_topic.up.sql (82B)
 // README.md (554B)
 // doc.go (850B)
 
@@ -527,11 +528,31 @@ func _1610959908_add_dont_wrap_to_raw_messagesUpSql() (*asset, error) {
 		return nil, err
 	}
 
-	info := bindataFileInfo{name: "1610959908_add_dont_wrap_to_raw_messages.up.sql", size: 83, mode: os.FileMode(0644), modTime: time.Unix(1610959977, 0)}
+	info := bindataFileInfo{name: "1610959908_add_dont_wrap_to_raw_messages.up.sql", size: 83, mode: os.FileMode(0644), modTime: time.Unix(1610991421, 0)}
 	a := &asset{bytes: bytes, info: info, digest: [32]uint8{0x71, 0x2, 0x9a, 0xca, 0xd4, 0x38, 0x44, 0x30, 0x2b, 0xa8, 0x27, 0x32, 0x63, 0x53, 0x22, 0x60, 0x59, 0x84, 0x23, 0x96, 0x77, 0xf0, 0x56, 0xd7, 0x94, 0xe0, 0x95, 0x28, 0x6, 0x1d, 0x4e, 0xb1}}
 	return a, nil
 }
 
+var __1610960912_add_send_on_personal_topicUpSql = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\x04\xc0\xc1\x0a\x82\x21\x0c\x07\xf0\x7b\x4f\xf1\x7f\x8f\x4e\x33\xe7\x69\x29\x94\x9e\x87\xd4\x88\xa0\x54\x5c\xd0\xeb\x7f\x3f\x92\xca\x37\x54\x0a\xc2\xd8\xfd\xaf\x5f\x73\xef\x2f\x73\x50\x8c\xb8\x14\x69\xd7\x0c\xb7\xf1\xd4\x39\x74\xd9\xf6\x39\xfa\x47\x7f\x73\xbd\x1f\x08\xa5\x08\x53\x46\xe4\x44\x4d\x2a\x12\xc9\x9d\xcf\xa7\x23\x00\x00\xff\xff\x14\x1b\x69\x22\x52\x00\x00\x00")
+
+func _1610960912_add_send_on_personal_topicUpSqlBytes() ([]byte, error) {
+	return bindataRead(
+		__1610960912_add_send_on_personal_topicUpSql,
+		"1610960912_add_send_on_personal_topic.up.sql",
+	)
+}
+
+func _1610960912_add_send_on_personal_topicUpSql() (*asset, error) {
+	bytes, err := _1610960912_add_send_on_personal_topicUpSqlBytes()
+	if err != nil {
+		return nil, err
+	}
+
+	info := bindataFileInfo{name: "1610960912_add_send_on_personal_topic.up.sql", size: 82, mode: os.FileMode(0644), modTime: time.Unix(1610991445, 0)}
+	a := &asset{bytes: bytes, info: info, digest: [32]uint8{0x77, 0xac, 0x2f, 0xc4, 0xd, 0xa7, 0x1b, 0x37, 0x30, 0xc2, 0x68, 0xee, 0xde, 0x54, 0x5e, 0xbf, 0x3f, 0xa0, 0xd6, 0xc6, 0x9f, 0xd4, 0x34, 0x12, 0x76, 0x1e, 0x66, 0x4a, 0xfc, 0xf, 0xee, 0xc9}}
+	return a, nil
+}
+
 var _readmeMd = []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\x54\x91\xc1\xce\xd3\x30\x10\x84\xef\x7e\x8a\x91\x7a\x01\xa9\x2a\x8f\xc0\x0d\x71\x82\x03\x48\x1c\xc9\x36\x9e\x36\x96\x1c\x6f\xf0\xae\x93\xe6\xed\x91\xa3\xc2\xdf\xff\x66\xed\xd8\x33\xdf\x78\x4f\xa7\x13\xbe\xea\x06\x57\x6c\x35\x39\x31\xa7\x7b\x15\x4f\x5a\xec\x73\x08\xbf\x08\x2d\x79\x7f\x4a\x43\x5b\x86\x17\xfd\x8c\x21\xea\x56\x5e\x47\x90\x4a\x14\x75\x48\xde\x64\x37\x2c\x6a\x96\xae\x99\x48\x05\xf6\x27\x77\x13\xad\x08\xae\x8a\x51\xe7\x25\xf3\xf1\xa9\x9f\xf9\x58\x58\x2c\xad\xbc\xe0\x8b\x56\xf0\x21\x5d\xeb\x4c\x95\xb3\xae\x84\x60\xd4\xdc\xe6\x82\x5d\x1b\x36\x6d\x39\x62\x92\xf5\xb8\x11\xdb\x92\xd3\x28\xce\xe0\x13\xe1\x72\xcd\x3c\x63\xd4\x65\x87\xae\xac\xe8\xc3\x28\x2e\x67\x44\x66\x3a\x21\x25\xa2\x72\xac\x14\x67\xbc\x84\x9f\x53\x32\x8c\x52\x70\x25\x56\xd6\xfd\x8d\x05\x37\xad\x30\x9d\x9f\xa6\x86\x0f\xcd\x58\x7f\xcf\x34\x93\x3b\xed\x90\x9f\xa4\x1f\xcf\x30\x85\x4d\x07\x58\xaf\x7f\x25\xc4\x9d\xf3\x72\x64\x84\xd0\x7f\xf9\x9b\x3a\x2d\x84\xef\x85\x48\x66\x8d\xd8\x88\x9b\x8c\x8c\x98\x5b\xf6\x74\x14\x4e\x33\x0d\xc9\xe0\x93\x38\xda\x12\xc5\x69\xbd\xe4\xf0\x2e\x7a\x78\x07\x1c\xfe\x13\x9f\x91\x29\x31\x95\x7b\x7f\x62\x59\x37\xb4\xe5\x5e\x25\xfe\x33\xee\xd5\x53\x71\xd6\xda\x3a\xd8\xcb\xde\x2e\xf8\xa1\x90\x55\x53\x0c\xc7\xaa\x0d\xe9\x76\x14\x29\x1c\x7b\x68\xdd\x2f\xe1\x6f\x00\x00\x00\xff\xff\x3c\x0a\xc2\xfe\x2a\x02\x00\x00")
 
 func readmeMdBytes() ([]byte, error) {
@@ -707,6 +728,8 @@ var _bindata = map[string]func() (*asset, error){
 
 	"1610959908_add_dont_wrap_to_raw_messages.up.sql": _1610959908_add_dont_wrap_to_raw_messagesUpSql,
 
+	"1610960912_add_send_on_personal_topic.up.sql": _1610960912_add_send_on_personal_topicUpSql,
+
 	"README.md": readmeMd,
 
 	"doc.go": docGo,
@@ -775,6 +798,7 @@ var _bintree = &bintree{nil, map[string]*bintree{
 	"1605075346_add_communities.up.sql":                           &bintree{_1605075346_add_communitiesUpSql, map[string]*bintree{}},
 	"1610117927_add_message_cache.up.sql":                         &bintree{_1610117927_add_message_cacheUpSql, map[string]*bintree{}},
 	"1610959908_add_dont_wrap_to_raw_messages.up.sql":             &bintree{_1610959908_add_dont_wrap_to_raw_messagesUpSql, map[string]*bintree{}},
+	"1610960912_add_send_on_personal_topic.up.sql":                &bintree{_1610960912_add_send_on_personal_topicUpSql, map[string]*bintree{}},
 	"README.md": &bintree{readmeMd, map[string]*bintree{}},
 	"doc.go":    &bintree{docGo, map[string]*bintree{}},
 }}

protocol/migrations/sqlite/1610960912_add_send_on_personal_topic.up.sql

@@ -0,0 +1 @@
+ALTER TABLE raw_messages ADD COLUMN send_on_personal_topic BOOLEAN DEFAULT FALSE;

protocol/persistence.go

@@ -590,11 +590,12 @@ func (db sqlitePersistence) SaveRawMessage(message *common.RawMessage) error {
 		   resend_automatically,
 		   recipients,
 		   skip_encryption,
-		   send_push_notification,
+	           send_push_notification,
 		   skip_group_message_wrap,
+		   send_on_personal_topic,
 		   payload
 		)
-		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?)`,
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		message.ID,
 		message.LocalChatID,
 		message.LastSent,
@@ -606,6 +607,7 @@ func (db sqlitePersistence) SaveRawMessage(message *common.RawMessage) error {
 		message.SkipEncryption,
 		message.SendPushNotification,
 		message.SkipGroupMessageWrap,
+		message.SendOnPersonalTopic,
 		message.Payload)
 	return err
 }
@@ -614,6 +616,7 @@ func (db sqlitePersistence) RawMessageByID(id string) (*common.RawMessage, error
 	var rawPubKeys [][]byte
 	var encodedRecipients []byte
 	var skipGroupMessageWrap sql.NullBool
+	var sendOnPersonalTopic sql.NullBool
 	message := &common.RawMessage{}
 
 	err := db.db.QueryRow(`
@@ -627,8 +630,9 @@ func (db sqlitePersistence) RawMessageByID(id string) (*common.RawMessage, error
 			  resend_automatically,
 			  recipients,
 			  skip_encryption,
-			  send_push_notification,
+		          send_push_notification,
 			  skip_group_message_wrap,
+			  send_on_personal_topic,
 			  payload
 			FROM
 				raw_messages
@@ -647,6 +651,7 @@ func (db sqlitePersistence) RawMessageByID(id string) (*common.RawMessage, error
 		&message.SkipEncryption,
 		&message.SendPushNotification,
 		&skipGroupMessageWrap,
+		&sendOnPersonalTopic,
 		&message.Payload,
 	)
 	if err != nil {
@@ -669,7 +674,10 @@ func (db sqlitePersistence) RawMessageByID(id string) (*common.RawMessage, error
 
 	if skipGroupMessageWrap.Valid {
 		message.SkipGroupMessageWrap = skipGroupMessageWrap.Bool
+	}
 
+	if sendOnPersonalTopic.Valid {
+		message.SendOnPersonalTopic = sendOnPersonalTopic.Bool
 	}
 
 	return message, nil

protocol/push_notification_test.go

@@ -124,6 +124,15 @@ func (s *MessengerPushNotificationSuite) TestReceivePushNotification() {
 		if !registered {
 			return errors.New("not registered")
 		}
+		bobServers, err := bob1.GetPushNotificationsServers()
+		if err != nil {
+			return err
+		}
+
+		if len(bobServers) == 0 {
+			return errors.New("not registered")
+		}
+
 		return nil
 	})
 	// Make sure we receive it
@@ -155,6 +164,15 @@ func (s *MessengerPushNotificationSuite) TestReceivePushNotification() {
 		if !registered {
 			return errors.New("not registered")
 		}
+		bobServers, err := bob2.GetPushNotificationsServers()
+		if err != nil {
+			return err
+		}
+
+		if len(bobServers) == 0 {
+			return errors.New("not registered")
+		}
+
 		return nil
 	})
 	// Make sure we receive it
@@ -304,6 +322,15 @@ func (s *MessengerPushNotificationSuite) TestReceivePushNotificationFromContactO
 		if !registered {
 			return errors.New("not registered")
 		}
+		bobServers, err := bob.GetPushNotificationsServers()
+		if err != nil {
+			return err
+		}
+
+		if len(bobServers) == 0 {
+			return errors.New("not registered")
+		}
+
 		return nil
 	})
 	// Make sure we receive it
@@ -449,6 +476,15 @@ func (s *MessengerPushNotificationSuite) TestReceivePushNotificationRetries() {
 		if !registered {
 			return errors.New("not registered")
 		}
+		bobServers, err := bob.GetPushNotificationsServers()
+		if err != nil {
+			return err
+		}
+
+		if len(bobServers) == 0 {
+			return errors.New("not registered")
+		}
+
 		return nil
 	})
 	// Make sure we receive it
@@ -637,6 +673,15 @@ func (s *MessengerPushNotificationSuite) TestContactCode() {
 		if !registered {
 			return errors.New("not registered")
 		}
+		bobServers, err := bob1.GetPushNotificationsServers()
+		if err != nil {
+			return err
+		}
+
+		if len(bobServers) == 0 {
+			return errors.New("not registered")
+		}
+
 		return nil
 	})
 	// Make sure we receive it
@@ -703,6 +748,16 @@ func (s *MessengerPushNotificationSuite) TestReceivePushNotificationMention() {
 		if !registered {
 			return errors.New("not registered")
 		}
+
+		bobServers, err := bob.GetPushNotificationsServers()
+		if err != nil {
+			return err
+		}
+
+		if len(bobServers) == 0 {
+			return errors.New("not registered")
+		}
+
 		return nil
 	})
 	// Make sure we receive it

protocol/pushnotificationclient/client.go

@@ -1267,6 +1267,9 @@ func (c *Client) registerWithServer(registration *protobuf.PushNotificationRegis
 	rawMessage := common.RawMessage{
 		Payload:     encryptedRegistration,
 		MessageType: protobuf.ApplicationMetadataMessage_PUSH_NOTIFICATION_REGISTRATION,
+		// We send on personal topic to avoid a lot of traffic on the partitioned topic
+		SendOnPersonalTopic: true,
+		SkipEncryption:      true,
 	}
 
 	_, err = c.messageProcessor.SendPrivate(context.Background(), server.PublicKey, rawMessage)

protocol/transport/filters_manager.go

@@ -253,6 +253,38 @@ func (s *FiltersManager) LoadPartitioned(publicKey *ecdsa.PublicKey, identity *e
 	return s.loadPartitioned(publicKey, identity, listen)
 }
 
+// LoadPersonal creates a filter for a personal topic.
+func (s *FiltersManager) LoadPersonal(publicKey *ecdsa.PublicKey, identity *ecdsa.PrivateKey, listen bool) (*Filter, error) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	chatID := PersonalDiscoveryTopic(publicKey)
+	if _, ok := s.filters[chatID]; ok {
+		return s.filters[chatID], nil
+	}
+
+	// We set up a filter so we can publish,
+	// but we discard envelopes if listen is false.
+	filter, err := s.addAsymmetric(chatID, identity, listen)
+	if err != nil {
+		return nil, err
+	}
+
+	chat := &Filter{
+		ChatID:   chatID,
+		FilterID: filter.FilterID,
+		Topic:    filter.Topic,
+		Identity: PublicKeyToStr(publicKey),
+		Listen:   listen,
+		OneToOne: true,
+	}
+
+	s.filters[chatID] = chat
+
+	return chat, nil
+
+}
+
 func (s *FiltersManager) loadMyPartitioned() (*Filter, error) {
 	return s.loadPartitioned(&s.privateKey.PublicKey, s.privateKey, true)
 }

protocol/transport/transport.go

@@ -22,6 +22,7 @@ type Transport interface {
 	SendPublic(ctx context.Context, newMessage *types.NewMessage, chatName string) ([]byte, error)
 	SendPrivateWithSharedSecret(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey, secret []byte) ([]byte, error)
 	SendPrivateWithPartitioned(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error)
+	SendPrivateOnPersonalTopic(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error)
 	SendMessagesRequest(
 		ctx context.Context,
 		peerID []byte,

protocol/transport/waku/waku_service.go

@@ -305,6 +305,22 @@ func (a *Transport) SendPrivateWithPartitioned(ctx context.Context, newMessage *
 	return a.api.Post(ctx, *newMessage)
 }
 
+func (a *Transport) SendPrivateOnPersonalTopic(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error) {
+	if err := a.addSig(newMessage); err != nil {
+		return nil, err
+	}
+
+	filter, err := a.filters.LoadPersonal(publicKey, a.keysManager.privateKey, false)
+	if err != nil {
+		return nil, err
+	}
+
+	newMessage.Topic = filter.Topic
+	newMessage.PublicKey = crypto.FromECDSAPub(publicKey)
+
+	return a.api.Post(ctx, *newMessage)
+}
+
 func (a *Transport) LoadKeyFilters(key *ecdsa.PrivateKey) (*transport.Filter, error) {
 	return a.filters.LoadPartitioned(&key.PublicKey, key, true)
 }

protocol/transport/whisper/whisper_service.go

@@ -335,6 +335,22 @@ func (a *Transport) SendPrivateWithPartitioned(ctx context.Context, newMessage *
 	return a.shhAPI.Post(ctx, *newMessage)
 }
 
+func (a *Transport) SendPrivateOnPersonalTopic(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error) {
+	if err := a.addSig(newMessage); err != nil {
+		return nil, err
+	}
+
+	filter, err := a.filters.LoadPersonal(publicKey, a.keysManager.privateKey, false)
+	if err != nil {
+		return nil, err
+	}
+
+	newMessage.Topic = filter.Topic
+	newMessage.PublicKey = crypto.FromECDSAPub(publicKey)
+
+	return a.shhAPI.Post(ctx, *newMessage)
+}
+
 func (a *Transport) SendPrivateOnDiscovery(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error) {
 	if err := a.addSig(newMessage); err != nil {
 		return nil, err