Send abridged version of history with messages in group chats
When sending a message in a private group chat we send the whole history for redundancy and allow out-of-order processing. This can be very expensive in some chats, resulting in long delay when sending a message and calculating the POW. This commit improves the performance by only forwarding the events necessary for the user to be able to construct a group chat and process correctly the message.
This commit is contained in:
Andrea Maria Piana
parent
d7772d3fc6
commit
9d2117e227
|
|
@ -284,16 +284,10 @@ func (p *MessageProcessor) SendPairInstallation(
|
|||
return messageID, nil
|
||||
}
|
||||
|
||||
// EncodeMembershipUpdate takes a group and an optional chat message and returns the protobuf representation to be sent on the wire.
|
||||
// All the events in a group are encoded and added to the payload
|
||||
func (p *MessageProcessor) EncodeMembershipUpdate(
|
||||
group *v1protocol.Group,
|
||||
func (p *MessageProcessor) encodeMembershipUpdate(
|
||||
message v1protocol.MembershipUpdateMessage,
|
||||
chatEntity ChatEntity,
|
||||
) ([]byte, error) {
|
||||
message := v1protocol.MembershipUpdateMessage{
|
||||
ChatID: group.ChatID(),
|
||||
Events: group.Events(),
|
||||
}
|
||||
|
||||
if chatEntity != nil {
|
||||
chatEntityProtobuf := chatEntity.GetProtobuf()
|
||||
|
|
@ -314,6 +308,33 @@ func (p *MessageProcessor) EncodeMembershipUpdate(
|
|||
return encodedMessage, nil
|
||||
}
|
||||
|
||||
// EncodeMembershipUpdate takes a group and an optional chat message and returns the protobuf representation to be sent on the wire.
|
||||
// All the events in a group are encoded and added to the payload
|
||||
func (p *MessageProcessor) EncodeMembershipUpdate(
|
||||
group *v1protocol.Group,
|
||||
chatEntity ChatEntity,
|
||||
) ([]byte, error) {
|
||||
message := v1protocol.MembershipUpdateMessage{
|
||||
ChatID: group.ChatID(),
|
||||
Events: group.Events(),
|
||||
}
|
||||
|
||||
return p.encodeMembershipUpdate(message, chatEntity)
|
||||
}
|
||||
|
||||
// EncodeAbridgedMembershipUpdate takes a group and an optional chat message and returns the protobuf representation to be sent on the wire.
|
||||
// Only the events relevant to the sender are encoded
|
||||
func (p *MessageProcessor) EncodeAbridgedMembershipUpdate(
|
||||
group *v1protocol.Group,
|
||||
chatEntity ChatEntity,
|
||||
) ([]byte, error) {
|
||||
message := v1protocol.MembershipUpdateMessage{
|
||||
ChatID: group.ChatID(),
|
||||
Events: group.AbridgedEvents(&p.identity.PublicKey),
|
||||
}
|
||||
return p.encodeMembershipUpdate(message, chatEntity)
|
||||
}
|
||||
|
||||
// SendPublic takes encoded data, encrypts it and sends through the wire.
|
||||
func (p *MessageProcessor) SendPublic(
|
||||
ctx context.Context,
|
||||
|
|
|
|||
|
|
@ -4538,7 +4538,7 @@ func (m *Messenger) encodeChatEntity(chat *Chat, message common.ChatEntity) ([]b
|
|||
return nil, err
|
||||
}
|
||||
|
||||
encodedMessage, err = m.processor.EncodeMembershipUpdate(group, message)
|
||||
encodedMessage, err = m.processor.EncodeAbridgedMembershipUpdate(group, message)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -42,7 +42,7 @@ func MembershipUpdateEventFromProtobuf(chatID string, raw []byte) (*MembershipUp
|
|||
return nil, errors.Wrap(err, "failed to extract signature")
|
||||
}
|
||||
|
||||
from := types.EncodeHex(crypto.FromECDSAPub(publicKey))
|
||||
from := publicKeyToString(publicKey)
|
||||
|
||||
err = proto.Unmarshal(encodedEvent, &decodedEvent)
|
||||
if err != nil {
|
||||
|
|
@ -147,7 +147,7 @@ func (u *MembershipUpdateEvent) Sign(key *ecdsa.PrivateKey) error {
|
|||
return err
|
||||
}
|
||||
u.Signature = signature
|
||||
u.From = types.EncodeHex(crypto.FromECDSAPub(&key.PublicKey))
|
||||
u.From = publicKeyToString(&key.PublicKey)
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
@ -241,7 +241,7 @@ type Group struct {
|
|||
}
|
||||
|
||||
func groupChatID(creator *ecdsa.PublicKey) string {
|
||||
return uuid.New().String() + "-" + types.EncodeHex(crypto.FromECDSAPub(creator))
|
||||
return uuid.New().String() + "-" + publicKeyToString(creator)
|
||||
}
|
||||
|
||||
func NewGroupWithEvents(chatID string, events []MembershipUpdateEvent) (*Group, error) {
|
||||
|
|
@ -314,6 +314,78 @@ func (g Group) Events() []MembershipUpdateEvent {
|
|||
return g.events
|
||||
}
|
||||
|
||||
func isInSlice(m string, set []string) bool {
|
||||
for _, k := range set {
|
||||
if k == m {
|
||||
return true
|
||||
}
|
||||
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// AbridgedEvents returns the minimum set of events for a user to publish a post
|
||||
func (g Group) AbridgedEvents(publicKey *ecdsa.PublicKey) []MembershipUpdateEvent {
|
||||
var events []MembershipUpdateEvent
|
||||
var nameChangedEventFound bool
|
||||
var joinedEventFound bool
|
||||
memberID := publicKeyToString(publicKey)
|
||||
var addedEventFound bool
|
||||
nextInChain := memberID
|
||||
// Iterate in reverse
|
||||
for i := len(g.events) - 1; i >= 0; i-- {
|
||||
event := g.events[i]
|
||||
switch event.Type {
|
||||
case protobuf.MembershipUpdateEvent_CHAT_CREATED:
|
||||
events = append(events, event)
|
||||
case protobuf.MembershipUpdateEvent_NAME_CHANGED:
|
||||
if nameChangedEventFound {
|
||||
continue
|
||||
}
|
||||
events = append(events, event)
|
||||
nameChangedEventFound = true
|
||||
case protobuf.MembershipUpdateEvent_MEMBERS_ADDED:
|
||||
// If we already have an added event
|
||||
// or the user is not in slice, ignore
|
||||
if addedEventFound {
|
||||
continue
|
||||
}
|
||||
|
||||
areWeTheTarget := isInSlice(nextInChain, event.Members)
|
||||
|
||||
// If it's us, and we have been added by the creator, no more work to do, this is authoritative
|
||||
if areWeTheTarget && g.events[0].From == event.From {
|
||||
addedEventFound = true
|
||||
events = append(events, event)
|
||||
|
||||
} else if areWeTheTarget {
|
||||
// if it's us and we haven't been added by the creator, we follow the history of whoever invited us
|
||||
nextInChain = event.From
|
||||
events = append(events, event)
|
||||
}
|
||||
case protobuf.MembershipUpdateEvent_MEMBER_JOINED:
|
||||
if joinedEventFound || event.From != memberID {
|
||||
continue
|
||||
}
|
||||
joinedEventFound = true
|
||||
events = append(events, event)
|
||||
|
||||
case protobuf.MembershipUpdateEvent_ADMINS_ADDED:
|
||||
if isInSlice(nextInChain, event.Members) {
|
||||
events = append(events, event)
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// Reverse events
|
||||
for i, j := 0, len(events)-1; i < j; i, j = i+1, j-1 {
|
||||
events[i], events[j] = events[j], events[i]
|
||||
}
|
||||
|
||||
return events
|
||||
}
|
||||
|
||||
func (g Group) Members() []string {
|
||||
return g.members.List()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -324,3 +324,171 @@ func TestMembershipUpdateEventEqual(t *testing.T) {
|
|||
u2.Signature = []byte("different-signature")
|
||||
require.False(t, u1.Equal(u2))
|
||||
}
|
||||
|
||||
func TestAbridgedEvents(t *testing.T) {
|
||||
var clock uint64 = 0
|
||||
creator, err := crypto.GenerateKey()
|
||||
require.NoError(t, err)
|
||||
creatorID := publicKeyToString(&creator.PublicKey)
|
||||
|
||||
member1, err := crypto.GenerateKey()
|
||||
require.NoError(t, err)
|
||||
member1ID := publicKeyToString(&member1.PublicKey)
|
||||
|
||||
member2, err := crypto.GenerateKey()
|
||||
require.NoError(t, err)
|
||||
member2ID := publicKeyToString(&member2.PublicKey)
|
||||
|
||||
member3, err := crypto.GenerateKey()
|
||||
require.NoError(t, err)
|
||||
member3ID := publicKeyToString(&member3.PublicKey)
|
||||
|
||||
member4, err := crypto.GenerateKey()
|
||||
require.NoError(t, err)
|
||||
member4ID := publicKeyToString(&member4.PublicKey)
|
||||
|
||||
g, err := NewGroupWithCreator("name-0", clock, creator)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// Full events is only a single one
|
||||
require.Len(t, g.Events(), 1)
|
||||
// same as abridged
|
||||
require.Len(t, g.AbridgedEvents(&creator.PublicKey), 1)
|
||||
|
||||
// We change name of the chat
|
||||
nameChangedEvent1 := NewNameChangedEvent("name-1", clock)
|
||||
nameChangedEvent1.From = creatorID
|
||||
nameChangedEvent1.ChatID = g.chatID
|
||||
err = g.ProcessEvent(nameChangedEvent1)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// We change name of the chat again
|
||||
nameChangedEvent2 := NewNameChangedEvent("name-2", clock)
|
||||
nameChangedEvent2.From = creatorID
|
||||
nameChangedEvent2.ChatID = g.chatID
|
||||
err = g.ProcessEvent(nameChangedEvent2)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// Full events is 3 events
|
||||
require.Len(t, g.Events(), 3)
|
||||
// While abridged should exclude the first name-1 event
|
||||
require.Len(t, g.AbridgedEvents(&creator.PublicKey), 2)
|
||||
require.Equal(t, g.AbridgedEvents(&creator.PublicKey)[1].Name, "name-2")
|
||||
|
||||
// Add a new member
|
||||
newMemberEvent1 := NewMembersAddedEvent([]string{member1ID}, clock)
|
||||
newMemberEvent1.From = creatorID
|
||||
newMemberEvent1.ChatID = g.chatID
|
||||
err = g.ProcessEvent(newMemberEvent1)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// Full events is 4 events
|
||||
require.Len(t, g.Events(), 4)
|
||||
// While abridged, given we are the creator, we only take 2 events and ignore
|
||||
// the member created event
|
||||
require.Len(t, g.AbridgedEvents(&creator.PublicKey), 2)
|
||||
require.Equal(t, g.AbridgedEvents(&creator.PublicKey)[1].Name, "name-2")
|
||||
|
||||
// While abridged, given we are the new member, we take 3 events
|
||||
// that are relevant to us
|
||||
require.Len(t, g.AbridgedEvents(&member1.PublicKey), 3)
|
||||
require.Equal(t, g.AbridgedEvents(&member1.PublicKey)[1].Name, "name-2")
|
||||
require.Equal(t, g.AbridgedEvents(&member1.PublicKey)[2].Members, []string{member1ID})
|
||||
|
||||
// We join the chat
|
||||
joinedEvent1 := NewMemberJoinedEvent(clock)
|
||||
joinedEvent1.From = member1ID
|
||||
joinedEvent1.ChatID = g.chatID
|
||||
err = g.ProcessEvent(joinedEvent1)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// Full events is 5 events
|
||||
require.Len(t, g.Events(), 5)
|
||||
// While abridged, given we are the creator, we only take 2 events and ignore
|
||||
// the member created event
|
||||
require.Len(t, g.AbridgedEvents(&creator.PublicKey), 2)
|
||||
require.Equal(t, g.AbridgedEvents(&creator.PublicKey)[1].Name, "name-2")
|
||||
|
||||
// While abridged, given we are the new member, we take 4 events
|
||||
// that are relevant to us
|
||||
require.Len(t, g.AbridgedEvents(&member1.PublicKey), 4)
|
||||
|
||||
// Next is the tricky case, a user that has been invited by someone
|
||||
// made an admin. We need to follow the history of admins so
|
||||
// that whoever receives the message can see that Creator-> Invited A -> Made A admin -> A Invited B
|
||||
|
||||
// Creator makes member1 Admin
|
||||
addedAdminEvent1 := NewAdminsAddedEvent([]string{member1ID}, clock)
|
||||
addedAdminEvent1.From = creatorID
|
||||
addedAdminEvent1.ChatID = g.chatID
|
||||
err = g.ProcessEvent(addedAdminEvent1)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// member1 adds member2
|
||||
newMemberEvent2 := NewMembersAddedEvent([]string{member2ID}, clock)
|
||||
newMemberEvent2.From = member1ID
|
||||
newMemberEvent2.ChatID = g.chatID
|
||||
err = g.ProcessEvent(newMemberEvent2)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// member1 makes member2 admin
|
||||
addedAdminEvent2 := NewAdminsAddedEvent([]string{member2ID}, clock)
|
||||
addedAdminEvent2.From = member1ID
|
||||
addedAdminEvent2.ChatID = g.chatID
|
||||
err = g.ProcessEvent(addedAdminEvent2)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// member2 adds member3
|
||||
newMemberEvent3 := NewMembersAddedEvent([]string{member3ID}, clock)
|
||||
newMemberEvent3.From = member2ID
|
||||
newMemberEvent3.ChatID = g.chatID
|
||||
err = g.ProcessEvent(newMemberEvent3)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// member1 makes member3 admin
|
||||
addedAdminEvent3 := NewAdminsAddedEvent([]string{member3ID}, clock)
|
||||
addedAdminEvent3.From = member1ID
|
||||
addedAdminEvent3.ChatID = g.chatID
|
||||
err = g.ProcessEvent(addedAdminEvent3)
|
||||
require.NoError(t, err)
|
||||
clock++
|
||||
|
||||
// member3 adds member4
|
||||
newMemberEvent4 := NewMembersAddedEvent([]string{member4ID}, clock)
|
||||
newMemberEvent4.From = member3ID
|
||||
newMemberEvent4.ChatID = g.chatID
|
||||
err = g.ProcessEvent(newMemberEvent4)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Now we check that the history has been correctly followed
|
||||
// Full events is 4 events
|
||||
require.Len(t, g.Events(), 11)
|
||||
// While abridged, given we are the creator, we only take 2 events and ignore
|
||||
// the member created event
|
||||
require.Len(t, g.AbridgedEvents(&creator.PublicKey), 2)
|
||||
require.Equal(t, g.AbridgedEvents(&creator.PublicKey)[1].Name, "name-2")
|
||||
|
||||
// While abridged, given we are the new member, we take 3 events
|
||||
// that are relevant to us
|
||||
require.Len(t, g.AbridgedEvents(&member4.PublicKey), 9)
|
||||
|
||||
// We build a group from the abridged events
|
||||
|
||||
group, err := NewGroupWithEvents(g.chatID, g.AbridgedEvents(&member4.PublicKey))
|
||||
require.NoError(t, err)
|
||||
|
||||
// Make sure the chatID, name is the same
|
||||
require.Equal(t, g.name, group.name)
|
||||
require.Equal(t, g.chatID, group.chatID)
|
||||
// Make sure that user 4 is a member
|
||||
require.True(t, group.IsMember(member4ID))
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue