Merge pull request #675 from tdakkota/fix/sort-updates

fix(updates): sort updates before applying

telegram/updates/sort_pts.go

@@ -0,0 +1,82 @@
+package updates
+
+import (
+	"sort"
+
+	"github.com/gotd/td/tg"
+)
+
+func sortUpdatesByPts(u []tg.UpdateClass) {
+	sort.Stable(ptsSorter(u))
+}
+
+type ptsSorter []tg.UpdateClass
+
+func (p ptsSorter) Len() int {
+	return len(p)
+}
+
+func (p ptsSorter) Less(i, j int) bool {
+	type (
+		ptsType int
+		compare struct {
+			typ       ptsType
+			channelID int64
+			ptsDiff   int
+		}
+	)
+	// Sorting order
+	//
+	// 0) Common updates without PTS.
+	// 1) Common PTS updates
+	// 2) Channel PTS updates (by channelID and by pts)
+	// 3) QTS updates
+	const (
+		other ptsType = iota
+		commonPts
+		channelPts
+		commonQts
+	)
+	getType := func(u tg.UpdateClass) compare {
+		if pts, ptsCount, ok := isCommonPtsUpdate(u); ok {
+			return compare{typ: commonPts, ptsDiff: pts - ptsCount}
+		}
+		if channelID, pts, ptsCount, ok, _ := isChannelPtsUpdate(u); ok {
+			return compare{typ: channelPts, channelID: channelID, ptsDiff: pts - ptsCount}
+		}
+		if qts, ok := isCommonQtsUpdate(u); ok {
+			return compare{typ: commonQts, ptsDiff: qts}
+		}
+		return compare{typ: other}
+	}
+
+	a, b := getType(p[i]), getType(p[j])
+	switch {
+	case a.typ < b.typ:
+		return true
+	case a.typ > b.typ:
+		return false
+	}
+
+	// a.typ == b.typ
+	switch a.typ {
+	case other:
+		// Keep original order
+	case commonPts, commonQts:
+		return a.ptsDiff < b.ptsDiff
+	case channelPts:
+		if a.channelID < b.channelID {
+			return true
+		}
+		if a.channelID > b.channelID {
+			return false
+		}
+		return a.ptsDiff < b.ptsDiff
+	}
+
+	return false
+}
+
+func (p ptsSorter) Swap(i, j int) {
+	p[i], p[j] = p[j], p[i]
+}

telegram/updates/sort_pts_test.go

@@ -0,0 +1,96 @@
+package updates
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/gotd/td/tg"
+)
+
+func Test_sortUpdatesByPts(t *testing.T) {
+	channelNewMessage := func(pts int, id int64) *tg.UpdateNewChannelMessage {
+		return &tg.UpdateNewChannelMessage{
+			Message: &tg.Message{
+				PeerID: &tg.PeerChannel{ChannelID: id},
+			},
+			Pts:      pts - 1,
+			PtsCount: 1,
+		}
+	}
+	newMessage := func(pts int) *tg.UpdateNewMessage {
+		return &tg.UpdateNewMessage{
+			Message: &tg.Message{
+				PeerID: &tg.PeerUser{UserID: 10},
+			},
+			Pts:      pts - 1,
+			PtsCount: 1,
+		}
+	}
+	encryptedNewMessage := func(pts int) *tg.UpdateNewEncryptedMessage {
+		return &tg.UpdateNewEncryptedMessage{
+			Qts: pts,
+		}
+	}
+	channelReadInbox := func(pts int, id int64) *tg.UpdateReadChannelInbox {
+		return &tg.UpdateReadChannelInbox{ChannelID: id, MaxID: 25, Pts: pts}
+	}
+
+	tests := []struct {
+		input  []tg.UpdateClass
+		result []tg.UpdateClass
+	}{
+		{
+			[]tg.UpdateClass{
+				channelReadInbox(26, 1),
+				channelNewMessage(25, 1),
+			},
+			[]tg.UpdateClass{
+				channelNewMessage(25, 1),
+				channelReadInbox(26, 1),
+			},
+		},
+		{
+			[]tg.UpdateClass{
+				channelReadInbox(26, 1),
+				channelNewMessage(25, 2),
+				newMessage(26),
+				encryptedNewMessage(26),
+				newMessage(25),
+				encryptedNewMessage(25),
+				encryptedNewMessage(27),
+				channelNewMessage(25, 1),
+			},
+			[]tg.UpdateClass{
+				newMessage(25),
+				newMessage(26),
+				channelNewMessage(25, 1),
+				channelReadInbox(26, 1),
+				channelNewMessage(25, 2),
+				encryptedNewMessage(25),
+				encryptedNewMessage(26),
+				encryptedNewMessage(27),
+			},
+		},
+		{
+			[]tg.UpdateClass{
+				channelReadInbox(26, 1),
+				&tg.UpdateConfig{},
+				channelNewMessage(25, 1),
+			},
+			[]tg.UpdateClass{
+				&tg.UpdateConfig{},
+				channelNewMessage(25, 1),
+				channelReadInbox(26, 1),
+			},
+		},
+	}
+	for i, tt := range tests {
+		tt := tt
+		t.Run(fmt.Sprintf("Test%d", i+1), func(t *testing.T) {
+			sortUpdatesByPts(tt.input)
+			require.Equal(t, tt.result, tt.input)
+		})
+	}
+}

telegram/updates/state_apply.go

@@ -40,6 +40,7 @@ func (s *state) applyCombined(ctx context.Context, comb *tg.UpdatesCombined) (pt
 		}
 		others []tg.UpdateClass
 	)
+	sortUpdatesByPts(comb.Updates)
 
 	for _, u := range comb.Updates {
 		switch u := u.(type) {