796 lines
31 KiB
Go
796 lines
31 KiB
Go
package updates
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"testing"
|
||
|
||
"telesrv/internal/domain"
|
||
"telesrv/internal/store/memory"
|
||
)
|
||
|
||
func TestRecordNewMessageFeedsGetDifference(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 1
|
||
svc := NewService(memory.NewUpdateStateStore(), memory.NewUpdateEventStore())
|
||
msg := domain.Message{
|
||
ID: 10,
|
||
OwnerUserID: 1000000001,
|
||
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
|
||
From: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
|
||
Date: 1700000000,
|
||
Body: "Login code: 12345",
|
||
}
|
||
|
||
event, state, err := svc.RecordNewMessage(ctx, authKeyID, msg.OwnerUserID, msg)
|
||
if err != nil {
|
||
t.Fatalf("RecordNewMessage: %v", err)
|
||
}
|
||
if event.Pts != 1 || event.PtsCount != 1 || state.Pts != 1 || state.Seq != 0 {
|
||
t.Fatalf("event/state = %+v / %+v, want first pts event with seq=0", event, state)
|
||
}
|
||
|
||
diff, err := svc.GetDifference(ctx, authKeyID, msg.OwnerUserID, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State != state || len(diff.Events) != 1 || diff.Events[0].Message.ID != msg.ID {
|
||
t.Fatalf("diff = %+v, want recorded login message event and state %+v", diff, state)
|
||
}
|
||
|
||
diff, err = svc.GetDifference(ctx, authKeyID, msg.OwnerUserID, state)
|
||
if err != nil {
|
||
t.Fatalf("GetDifference current: %v", err)
|
||
}
|
||
if len(diff.Events) != 0 || diff.State != state {
|
||
t.Fatalf("current diff = %+v, want empty events and same state", diff)
|
||
}
|
||
}
|
||
|
||
func TestPublishNewMessageIsIdempotentByMessageBoxID(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 11
|
||
events := memory.NewUpdateEventStore()
|
||
svc := NewService(memory.NewUpdateStateStore(), events)
|
||
msg := domain.Message{
|
||
ID: 10,
|
||
OwnerUserID: 1000000001,
|
||
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
|
||
From: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
|
||
Date: 1700000000,
|
||
Body: "Login code: 12345",
|
||
}
|
||
|
||
firstEvent, firstState, err := svc.PublishNewMessage(ctx, msg.OwnerUserID, msg)
|
||
if err != nil {
|
||
t.Fatalf("PublishNewMessage first: %v", err)
|
||
}
|
||
secondEvent, secondState, err := svc.PublishNewMessage(ctx, msg.OwnerUserID, msg)
|
||
if err != nil {
|
||
t.Fatalf("PublishNewMessage retry: %v", err)
|
||
}
|
||
if firstEvent.Pts != 1 || firstState.Pts != 1 {
|
||
t.Fatalf("first event/state = %+v / %+v, want pts=1", firstEvent, firstState)
|
||
}
|
||
if secondEvent.Pts != firstEvent.Pts || secondState.Pts != firstState.Pts {
|
||
t.Fatalf("retry event/state = %+v / %+v, want same pts as first %+v / %+v", secondEvent, secondState, firstEvent, firstState)
|
||
}
|
||
|
||
diff, err := svc.GetDifference(ctx, authKeyID, msg.OwnerUserID, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State.Pts != 1 || len(diff.Events) != 1 || diff.Events[0].Message.ID != msg.ID {
|
||
t.Fatalf("diff = %+v, want one durable login message event", diff)
|
||
}
|
||
}
|
||
|
||
func TestRecordReadHistoryFeedsGetDifference(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 2
|
||
svc := NewService(memory.NewUpdateStateStore(), memory.NewUpdateEventStore())
|
||
peer := domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID}
|
||
|
||
ownerUserID := int64(1000000001)
|
||
event, state, err := svc.RecordReadHistory(ctx, authKeyID, ownerUserID, domain.ReadHistoryResult{
|
||
OwnerUserID: ownerUserID,
|
||
Peer: peer,
|
||
MaxID: 10,
|
||
Changed: true,
|
||
}, [8]byte{}, 0)
|
||
if err != nil {
|
||
t.Fatalf("RecordReadHistory: %v", err)
|
||
}
|
||
if event.Type != domain.UpdateEventReadHistoryInbox || event.Pts != 1 || event.PtsCount != 1 || state.Pts != 1 {
|
||
t.Fatalf("event/state = %+v / %+v, want read history event with first pts", event, state)
|
||
}
|
||
|
||
diff, err := svc.GetDifference(ctx, authKeyID, ownerUserID, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State != state || len(diff.Events) != 1 || diff.Events[0].Peer != peer || diff.Events[0].MaxID != 10 {
|
||
t.Fatalf("diff = %+v, want recorded read history event and state %+v", diff, state)
|
||
}
|
||
}
|
||
|
||
func TestRecordChannelReadHistoryKeepsChannelPtsPayload(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 7
|
||
svc := NewService(memory.NewUpdateStateStore(), memory.NewUpdateEventStore())
|
||
ownerUserID := int64(1000000001)
|
||
peer := domain.Peer{Type: domain.PeerTypeChannel, ID: 7001}
|
||
|
||
event, state, err := svc.RecordReadHistory(ctx, authKeyID, ownerUserID, domain.ReadHistoryResult{
|
||
OwnerUserID: ownerUserID,
|
||
Peer: peer,
|
||
MaxID: 11,
|
||
StillUnreadCount: 3,
|
||
ChannelPts: 77,
|
||
Changed: true,
|
||
}, [8]byte{}, 0)
|
||
if err != nil {
|
||
t.Fatalf("RecordReadHistory: %v", err)
|
||
}
|
||
if event.Pts != 1 || state.Pts != 1 || event.ChannelPts != 77 {
|
||
t.Fatalf("event/state = %+v / %+v, want account pts=1 and channel pts payload=77", event, state)
|
||
}
|
||
|
||
diff, err := svc.GetDifference(ctx, authKeyID, ownerUserID, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if len(diff.Events) != 1 || diff.Events[0].Peer != peer || diff.Events[0].ChannelPts != 77 {
|
||
t.Fatalf("diff = %+v, want recorded channel read payload with channel pts=77", diff)
|
||
}
|
||
}
|
||
|
||
func TestRecordSettingsEventsFeedGetDifference(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 3
|
||
svc := NewService(memory.NewUpdateStateStore(), memory.NewUpdateEventStore())
|
||
ownerUserID := int64(1000000001)
|
||
peer := domain.Peer{Type: domain.PeerTypeUser, ID: 1000000002}
|
||
|
||
if _, _, err := svc.RecordContactsReset(ctx, authKeyID, ownerUserID, [8]byte{}, 0); err != nil {
|
||
t.Fatalf("RecordContactsReset: %v", err)
|
||
}
|
||
if _, _, err := svc.RecordDialogPinned(ctx, authKeyID, ownerUserID, peer, true, 0, [8]byte{}, 0); err != nil {
|
||
t.Fatalf("RecordDialogPinned: %v", err)
|
||
}
|
||
order := []domain.Peer{peer}
|
||
if _, _, err := svc.RecordPinnedDialogs(ctx, authKeyID, ownerUserID, 0, order, [8]byte{}, 0); err != nil {
|
||
t.Fatalf("RecordPinnedDialogs: %v", err)
|
||
}
|
||
if _, _, err := svc.RecordDialogUnreadMark(ctx, authKeyID, ownerUserID, peer, false, [8]byte{}, 0); err != nil {
|
||
t.Fatalf("RecordDialogUnreadMark: %v", err)
|
||
}
|
||
settings := domain.PeerSettings{ShareContact: true}
|
||
if _, _, err := svc.RecordPeerSettings(ctx, authKeyID, ownerUserID, peer, settings, [8]byte{}, 0); err != nil {
|
||
t.Fatalf("RecordPeerSettings: %v", err)
|
||
}
|
||
stateEvent, state, err := svc.RecordPeerStoryBlocked(ctx, authKeyID, ownerUserID, peer, true, [8]byte{}, 0)
|
||
if err != nil {
|
||
t.Fatalf("RecordPeerStoryBlocked: %v", err)
|
||
}
|
||
if stateEvent.Pts != 6 || state.Pts != 6 {
|
||
t.Fatalf("last event/state = %+v / %+v, want pts=6", stateEvent, state)
|
||
}
|
||
|
||
diff, err := svc.GetDifference(ctx, authKeyID, ownerUserID, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State.Pts != 6 || len(diff.Events) != 6 {
|
||
t.Fatalf("diff = %+v, want six settings events", diff)
|
||
}
|
||
wantTypes := []domain.UpdateEventType{
|
||
domain.UpdateEventContactsReset,
|
||
domain.UpdateEventDialogPinned,
|
||
domain.UpdateEventPinnedDialogs,
|
||
domain.UpdateEventDialogUnreadMark,
|
||
domain.UpdateEventPeerSettings,
|
||
domain.UpdateEventPeerStoryBlocked,
|
||
}
|
||
for i, typ := range wantTypes {
|
||
if diff.Events[i].Type != typ || diff.Events[i].Pts != i+1 || diff.Events[i].PtsCount != 1 {
|
||
t.Fatalf("event[%d] = %+v, want type=%s pts=%d pts_count=1", i, diff.Events[i], typ, i+1)
|
||
}
|
||
}
|
||
if diff.Events[1].Peer != peer || !diff.Events[1].Bool {
|
||
t.Fatalf("dialog pinned event = %+v, want peer and pinned=true", diff.Events[1])
|
||
}
|
||
if diff.Events[3].Peer != peer || diff.Events[3].Bool {
|
||
t.Fatalf("unread mark event = %+v, want peer and unread=false", diff.Events[3])
|
||
}
|
||
if len(diff.Events[2].Peers) != 1 || diff.Events[2].Peers[0] != peer {
|
||
t.Fatalf("pinned dialogs event = %+v, want order peer", diff.Events[2])
|
||
}
|
||
if diff.Events[4].Peer != peer || !diff.Events[4].Settings.ShareContact {
|
||
t.Fatalf("peer settings event = %+v, want peer and settings", diff.Events[4])
|
||
}
|
||
if diff.Events[5].Peer != peer || !diff.Events[5].Bool {
|
||
t.Fatalf("peer story blocked event = %+v, want peer and blocked=true", diff.Events[5])
|
||
}
|
||
}
|
||
|
||
func TestRecordSettingsEventUsesDispatchAppender(t *testing.T) {
|
||
ctx := context.Background()
|
||
authKeyID := [8]byte{4}
|
||
rawAuthKeyID := [8]byte{4, 9}
|
||
events := &captureDispatchAppender{UpdateEventStore: memory.NewUpdateEventStore()}
|
||
states := &captureStateStore{}
|
||
svc := NewService(states, events)
|
||
peer := domain.Peer{Type: domain.PeerTypeUser, ID: 1000000002}
|
||
|
||
event, state, err := svc.RecordDialogPinned(ctx, authKeyID, 1000000001, peer, true, 0, rawAuthKeyID, 42)
|
||
if err != nil {
|
||
t.Fatalf("RecordDialogPinned: %v", err)
|
||
}
|
||
if event.Pts != 1 || state.Pts != 1 {
|
||
t.Fatalf("event/state = %+v / %+v, want first pts", event, state)
|
||
}
|
||
if !events.dispatched || events.excludeAuthKeyID != rawAuthKeyID || events.excludeSessionID != 42 || events.event.Type != domain.UpdateEventDialogPinned || events.event.Peer != peer {
|
||
t.Fatalf("dispatch capture = %+v exclude_auth=%v exclude_session=%d dispatched=%v, want dialog_pinned outbox", events.event, events.excludeAuthKeyID, events.excludeSessionID, events.dispatched)
|
||
}
|
||
if states.lastSaveAuthKeyID != authKeyID {
|
||
t.Fatalf("device state auth key = %x, want business/perm %x", states.lastSaveAuthKeyID, authKeyID)
|
||
}
|
||
if _, found, err := states.Get(ctx, rawAuthKeyID, 1000000001); err != nil || found {
|
||
t.Fatalf("raw temp key unexpectedly owns device state: found=%v err=%v", found, err)
|
||
}
|
||
}
|
||
|
||
func TestRecordSettingsEventDispatchFailureDoesNotRecordEvent(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 6
|
||
events := &failingDispatchAppender{UpdateEventStore: memory.NewUpdateEventStore()}
|
||
svc := NewService(memory.NewUpdateStateStore(), events)
|
||
|
||
_, _, err := svc.RecordDialogPinned(ctx, authKeyID, 1000000001, domain.Peer{Type: domain.PeerTypeUser, ID: 1000000002}, true, 0, authKeyID, 42)
|
||
if !errors.Is(err, errDispatchFailed) {
|
||
t.Fatalf("RecordDialogPinned err = %v, want dispatch failure", err)
|
||
}
|
||
diff, err := svc.GetDifference(ctx, authKeyID, 1000000001, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State.Pts != 0 || len(diff.Events) != 0 {
|
||
t.Fatalf("diff after dispatch failure = %+v, want no durable event before allocated append commits", diff)
|
||
}
|
||
}
|
||
|
||
func TestRecordPeerStoryBlockedUsesDispatchAppender(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 7
|
||
events := &captureDispatchAppender{UpdateEventStore: memory.NewUpdateEventStore()}
|
||
svc := NewService(memory.NewUpdateStateStore(), events)
|
||
peer := domain.Peer{Type: domain.PeerTypeUser, ID: 1000000002}
|
||
|
||
event, state, err := svc.RecordPeerStoryBlocked(ctx, authKeyID, 1000000001, peer, true, authKeyID, 91)
|
||
if err != nil {
|
||
t.Fatalf("RecordPeerStoryBlocked: %v", err)
|
||
}
|
||
if event.Pts != 1 || state.Pts != 1 || !event.LacksWirePts() {
|
||
t.Fatalf("event/state = %+v / %+v, want first aux pts event", event, state)
|
||
}
|
||
if !events.dispatched || events.excludeAuthKeyID != authKeyID || events.excludeSessionID != 91 || events.event.Type != domain.UpdateEventPeerStoryBlocked || events.event.Peer != peer || !events.event.Bool {
|
||
t.Fatalf("dispatch capture = %+v exclude_auth=%v exclude_session=%d dispatched=%v, want peer_story_blocked outbox", events.event, events.excludeAuthKeyID, events.excludeSessionID, events.dispatched)
|
||
}
|
||
}
|
||
|
||
func TestRecordStoryUsesDispatchAppenderExcludeCurrentSession(t *testing.T) {
|
||
ctx := context.Background()
|
||
authKeyID := [8]byte{8, 1, 0}
|
||
events := &captureDispatchAppender{UpdateEventStore: memory.NewUpdateEventStore()}
|
||
svc := NewService(memory.NewUpdateStateStore(), events)
|
||
owner := domain.Peer{Type: domain.PeerTypeUser, ID: 1000000001}
|
||
story := domain.Story{
|
||
Owner: owner,
|
||
ID: 3,
|
||
Date: 1700000100,
|
||
ExpireDate: 1700086500,
|
||
Public: true,
|
||
Caption: "owner story",
|
||
}
|
||
|
||
event, state, err := svc.RecordStory(ctx, authKeyID, owner.ID, story, authKeyID, 1234)
|
||
if err != nil {
|
||
t.Fatalf("RecordStory: %v", err)
|
||
}
|
||
if event.Type != domain.UpdateEventStory || event.Pts != 1 || event.PtsCount != 1 || state.Pts != 1 {
|
||
t.Fatalf("event/state = %+v / %+v, want first story pts event", event, state)
|
||
}
|
||
if !events.dispatched || events.userID != owner.ID || events.excludeAuthKeyID != authKeyID || events.excludeSessionID != 1234 {
|
||
t.Fatalf("dispatch capture = user %d exclude_auth=%v exclude_session=%d dispatched=%v, want current session excluded", events.userID, events.excludeAuthKeyID, events.excludeSessionID, events.dispatched)
|
||
}
|
||
if events.event.Type != domain.UpdateEventStory || events.event.Peer != owner || events.event.Story.ID != story.ID {
|
||
t.Fatalf("dispatch event = %+v, want story update for owner story", events.event)
|
||
}
|
||
}
|
||
|
||
func TestRecordStoryReadAndSentReactionExcludeCurrentSession(t *testing.T) {
|
||
ctx := context.Background()
|
||
authKeyID := [8]byte{8, 1, 4}
|
||
events := &captureDispatchAppender{UpdateEventStore: memory.NewUpdateEventStore()}
|
||
svc := NewService(memory.NewUpdateStateStore(), events)
|
||
viewerID := int64(1000000002)
|
||
peer := domain.Peer{Type: domain.PeerTypeUser, ID: 1000000001}
|
||
story := domain.Story{
|
||
Owner: peer,
|
||
ID: 5,
|
||
Date: 1700000200,
|
||
ExpireDate: 1700086600,
|
||
Public: true,
|
||
}
|
||
|
||
event, state, err := svc.RecordReadStories(ctx, authKeyID, viewerID, domain.StoryReadResult{
|
||
ViewerID: viewerID,
|
||
Peer: peer,
|
||
MaxReadID: story.ID,
|
||
Advanced: true,
|
||
Date: 1700000201,
|
||
}, authKeyID, 2233)
|
||
if err != nil {
|
||
t.Fatalf("RecordReadStories: %v", err)
|
||
}
|
||
if event.Type != domain.UpdateEventReadStories || event.Pts != 1 || state.Pts != 1 {
|
||
t.Fatalf("read event/state = %+v / %+v, want first read story pts event", event, state)
|
||
}
|
||
if !events.dispatched || events.userID != viewerID || events.excludeAuthKeyID != authKeyID || events.excludeSessionID != 2233 || events.event.Type != domain.UpdateEventReadStories || events.event.MaxID != story.ID {
|
||
t.Fatalf("read dispatch capture = %+v user %d exclude_auth=%v exclude_session=%d dispatched=%v, want current session excluded", events.event, events.userID, events.excludeAuthKeyID, events.excludeSessionID, events.dispatched)
|
||
}
|
||
|
||
reaction := &domain.MessageReaction{Type: domain.MessageReactionEmoji, Emoticon: "🔥"}
|
||
event, state, err = svc.RecordSentStoryReaction(ctx, authKeyID, viewerID, domain.StoryReactionResult{
|
||
ViewerID: viewerID,
|
||
Peer: peer,
|
||
StoryID: story.ID,
|
||
Story: story,
|
||
Reaction: reaction,
|
||
Changed: true,
|
||
Date: 1700000202,
|
||
}, authKeyID, 2233)
|
||
if err != nil {
|
||
t.Fatalf("RecordSentStoryReaction: %v", err)
|
||
}
|
||
if event.Type != domain.UpdateEventSentStoryReaction || event.Pts != 2 || state.Pts != 2 {
|
||
t.Fatalf("reaction event/state = %+v / %+v, want second sent story reaction pts event", event, state)
|
||
}
|
||
if !events.dispatched || events.userID != viewerID || events.excludeAuthKeyID != authKeyID || events.excludeSessionID != 2233 || events.event.Type != domain.UpdateEventSentStoryReaction || events.event.Reaction == nil || events.event.Reaction.Emoticon != "🔥" {
|
||
t.Fatalf("reaction dispatch capture = %+v user %d exclude_auth=%v exclude_session=%d dispatched=%v, want current session excluded", events.event, events.userID, events.excludeAuthKeyID, events.excludeSessionID, events.dispatched)
|
||
}
|
||
}
|
||
|
||
func TestRecordNewStoryReactionDispatchesWithoutSavingDeviceState(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 5
|
||
states := &captureStateStore{}
|
||
events := &captureDispatchAppender{UpdateEventStore: memory.NewUpdateEventStore()}
|
||
svc := NewService(states, events)
|
||
ownerID := int64(1000000001)
|
||
viewerID := int64(1000000002)
|
||
reaction := &domain.MessageReaction{Type: domain.MessageReactionEmoji, Emoticon: "👍"}
|
||
|
||
event, state, err := svc.RecordNewStoryReaction(ctx, authKeyID, 0, domain.StoryReactionResult{
|
||
ViewerID: viewerID,
|
||
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: ownerID},
|
||
StoryID: 3,
|
||
Story: domain.Story{
|
||
Owner: domain.Peer{Type: domain.PeerTypeUser, ID: ownerID},
|
||
ID: 3,
|
||
Date: 1700000100,
|
||
},
|
||
Reaction: reaction,
|
||
Date: 1700000101,
|
||
}, [8]byte{}, 0)
|
||
if err != nil {
|
||
t.Fatalf("RecordNewStoryReaction: %v", err)
|
||
}
|
||
if event.Type != domain.UpdateEventNewStoryReaction || event.UserID != ownerID || event.Peer.ID != viewerID || event.Reaction == nil || event.Reaction.Emoticon != "👍" {
|
||
t.Fatalf("event = %+v, want owner-side new story reaction from viewer", event)
|
||
}
|
||
if state.Pts != 1 || state.Seq != 0 {
|
||
t.Fatalf("state = %+v, want first account pts", state)
|
||
}
|
||
if states.saveCount != 0 {
|
||
t.Fatalf("state saves = %d, want no device state save for remote owner notification", states.saveCount)
|
||
}
|
||
if !events.dispatched || events.userID != ownerID || events.event.Type != domain.UpdateEventNewStoryReaction {
|
||
t.Fatalf("dispatch capture = %+v user=%d dispatched=%v, want owner outbox event", events.event, events.userID, events.dispatched)
|
||
}
|
||
|
||
diff, err := svc.GetDifference(ctx, authKeyID, ownerID, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State.Pts != 1 || len(diff.Events) != 1 || diff.Events[0].Type != domain.UpdateEventNewStoryReaction {
|
||
t.Fatalf("diff = %+v, want one durable new story reaction", diff)
|
||
}
|
||
}
|
||
|
||
func TestClearAuthKeyDropsStateAndEvents(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 8
|
||
states := memory.NewUpdateStateStore()
|
||
events := memory.NewUpdateEventStore()
|
||
svc := NewService(states, events)
|
||
msg := domain.Message{
|
||
ID: 1,
|
||
OwnerUserID: 1000000001,
|
||
Peer: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
|
||
From: domain.Peer{Type: domain.PeerTypeUser, ID: domain.OfficialSystemUserID},
|
||
Date: 1700000000,
|
||
}
|
||
if _, _, err := svc.RecordNewMessage(ctx, authKeyID, msg.OwnerUserID, msg); err != nil {
|
||
t.Fatalf("RecordNewMessage: %v", err)
|
||
}
|
||
if err := svc.ClearAuthKey(ctx, authKeyID); err != nil {
|
||
t.Fatalf("ClearAuthKey: %v", err)
|
||
}
|
||
diff, err := svc.GetDifference(ctx, authKeyID, msg.OwnerUserID, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State.Pts != 1 || len(diff.Events) != 1 {
|
||
t.Fatalf("difference after clear = %+v, want durable user events to remain", diff)
|
||
}
|
||
diff, err = svc.GetDifference(ctx, authKeyID, msg.OwnerUserID+1, domain.UpdateState{})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference other user: %v", err)
|
||
}
|
||
if diff.State.Pts != 0 || len(diff.Events) != 0 {
|
||
t.Fatalf("difference for other user after clear = %+v, want no cross-account events", diff)
|
||
}
|
||
}
|
||
|
||
func TestDeleteMessagesPtsRangeFeedsGetDifference(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 9
|
||
userID := int64(1000000001)
|
||
events := memory.NewUpdateEventStore()
|
||
svc := NewService(memory.NewUpdateStateStore(), events)
|
||
for _, event := range []domain.UpdateEvent{
|
||
{UserID: userID, Type: domain.UpdateEventNewMessage, Pts: 1, PtsCount: 1, Date: 1700000001, Message: domain.Message{ID: 1, OwnerUserID: userID}},
|
||
{UserID: userID, Type: domain.UpdateEventNewMessage, Pts: 2, PtsCount: 1, Date: 1700000002, Message: domain.Message{ID: 2, OwnerUserID: userID}},
|
||
{UserID: userID, Type: domain.UpdateEventDeleteMessages, Pts: 4, PtsCount: 2, Date: 1700000003, MessageIDs: []int{1, 2}},
|
||
} {
|
||
if err := events.Append(ctx, userID, event); err != nil {
|
||
t.Fatalf("append event pts=%d: %v", event.Pts, err)
|
||
}
|
||
}
|
||
|
||
state, err := svc.GetState(ctx, authKeyID, userID)
|
||
if err != nil {
|
||
t.Fatalf("GetState: %v", err)
|
||
}
|
||
if state.Pts != 4 {
|
||
t.Fatalf("state = %+v, want contiguous pts=4 across delete range", state)
|
||
}
|
||
diff, err := svc.GetDifference(ctx, authKeyID, userID, domain.UpdateState{Pts: 2})
|
||
if err != nil {
|
||
t.Fatalf("GetDifference: %v", err)
|
||
}
|
||
if diff.State.Pts != 4 || len(diff.Events) != 1 {
|
||
t.Fatalf("diff = %+v, want one delete event ending at pts=4", diff)
|
||
}
|
||
got := diff.Events[0]
|
||
if got.Type != domain.UpdateEventDeleteMessages || got.Pts != 4 || got.PtsCount != 2 || len(got.MessageIDs) != 2 {
|
||
t.Fatalf("delete event = %+v, want pts=4 pts_count=2 ids", got)
|
||
}
|
||
}
|
||
|
||
// TestCurrentStateCommitsAuditedBaselineOnlyAfterDelivery verifies that
|
||
// computing a getState result is side-effect free and that its physically
|
||
// delivered baseline advances confirmed+observed atomically.
|
||
func TestCurrentStateCommitsAuditedBaselineOnlyAfterDelivery(t *testing.T) {
|
||
ctx := context.Background()
|
||
var authKeyID [8]byte
|
||
authKeyID[0] = 11
|
||
userID := int64(1000000001)
|
||
events := memory.NewUpdateEventStore()
|
||
states := memory.NewUpdateStateStore()
|
||
svc := NewService(states, events)
|
||
if err := events.Append(ctx, userID, domain.UpdateEvent{
|
||
UserID: userID, Type: domain.UpdateEventNewMessage, Pts: 1, PtsCount: 1,
|
||
Date: 1700000001, Message: domain.Message{ID: 1, OwnerUserID: userID},
|
||
}); err != nil {
|
||
t.Fatalf("append: %v", err)
|
||
}
|
||
// 设备确认水位停在 pts=1 后账号又推进两格。
|
||
if err := states.Save(ctx, authKeyID, userID, domain.UpdateState{Pts: 1, Date: 1700000001}); err != nil {
|
||
t.Fatalf("seed confirmed state: %v", err)
|
||
}
|
||
for pts := 2; pts <= 3; pts++ {
|
||
if err := events.Append(ctx, userID, domain.UpdateEvent{
|
||
UserID: userID, Type: domain.UpdateEventNewMessage, Pts: pts, PtsCount: 1,
|
||
Date: 1700000001 + pts, Message: domain.Message{ID: pts, OwnerUserID: userID},
|
||
}); err != nil {
|
||
t.Fatalf("append pts=%d: %v", pts, err)
|
||
}
|
||
}
|
||
|
||
st, err := svc.CurrentState(ctx, userID)
|
||
if err != nil {
|
||
t.Fatalf("CurrentState: %v", err)
|
||
}
|
||
if st.Pts != 3 {
|
||
t.Fatalf("current state pts = %d, want account current 3", st.Pts)
|
||
}
|
||
confirmed, _, err := svc.ConfirmedState(ctx, authKeyID, userID)
|
||
if err != nil {
|
||
t.Fatalf("ConfirmedState before delivery: %v", err)
|
||
}
|
||
if confirmed.Pts != 1 {
|
||
t.Fatalf("confirmed before delivery = %d, want 1", confirmed.Pts)
|
||
}
|
||
if _, ok := states.ObservedClientState(authKeyID, userID); ok {
|
||
t.Fatal("computed getState unexpectedly advanced observed")
|
||
}
|
||
if err := svc.CommitDeliveredState(ctx, authKeyID, userID, st, domain.UpdateStateCommitDeliveredAndObservedBaseline); err != nil {
|
||
t.Fatalf("CommitDeliveredState: %v", err)
|
||
}
|
||
confirmed, _, err = svc.ConfirmedState(ctx, authKeyID, userID)
|
||
if err != nil || confirmed.Pts != 3 {
|
||
t.Fatalf("confirmed after delivery = %+v err=%v, want pts=3", confirmed, err)
|
||
}
|
||
observed, ok := states.ObservedClientState(authKeyID, userID)
|
||
if !ok || observed.Pts != 3 {
|
||
t.Fatalf("observed after delivered baseline = %+v/%v, want pts=3", observed, ok)
|
||
}
|
||
}
|
||
|
||
func TestGetDifferenceRetainsOnlyClientObservedInputCursor(t *testing.T) {
|
||
ctx := context.Background()
|
||
authKeyID := [8]byte{12}
|
||
const userID int64 = 1000000012
|
||
events := memory.NewUpdateEventStore()
|
||
states := memory.NewUpdateStateStore()
|
||
svc := NewService(states, events)
|
||
for pts := 1; pts <= 2; pts++ {
|
||
if err := events.Append(ctx, userID, domain.UpdateEvent{
|
||
UserID: userID, Type: domain.UpdateEventNewMessage, Pts: pts, PtsCount: 1,
|
||
Date: 1700000100 + pts, Message: domain.Message{ID: pts, OwnerUserID: userID},
|
||
}); err != nil {
|
||
t.Fatalf("append pts=%d: %v", pts, err)
|
||
}
|
||
}
|
||
|
||
// 服务端把 pts=1..2 放进 response，并不证明客户端收到了 response；observed 只能
|
||
// 保持在本次 request 实际携带的 pts=0。
|
||
from, err := svc.ObserveDifferenceRequest(ctx, authKeyID, userID, domain.UpdateState{Pts: 0, Date: 1700000100})
|
||
if err != nil {
|
||
t.Fatalf("observe first request: %v", err)
|
||
}
|
||
diff, err := svc.GetDifference(ctx, authKeyID, userID, from)
|
||
if err != nil {
|
||
t.Fatalf("first difference: %v", err)
|
||
}
|
||
if diff.State.Pts != 2 || len(diff.Events) != 2 {
|
||
t.Fatalf("first difference = %+v, want response through pts=2", diff)
|
||
}
|
||
observed, ok := states.ObservedClientState(authKeyID, userID)
|
||
if !ok || observed.Pts != 0 {
|
||
t.Fatalf("observed after merely sending response = %+v/%v, want pts=0", observed, ok)
|
||
}
|
||
if _, found, err := svc.ConfirmedState(ctx, authKeyID, userID); err != nil || found {
|
||
t.Fatalf("computed response advanced confirmed: found=%v err=%v", found, err)
|
||
}
|
||
if err := svc.CommitDeliveredState(ctx, authKeyID, userID, diff.State, domain.UpdateStateCommitDeliveredOnly); err != nil {
|
||
t.Fatalf("commit delivered difference: %v", err)
|
||
}
|
||
if confirmed, found, err := svc.ConfirmedState(ctx, authKeyID, userID); err != nil || !found || confirmed.Pts != 2 {
|
||
t.Fatalf("confirmed after delivery = %+v/%v err=%v, want pts=2", confirmed, found, err)
|
||
}
|
||
observed, _ = states.ObservedClientState(authKeyID, userID)
|
||
if observed.Pts != 0 {
|
||
t.Fatalf("delivered difference advanced observed to %d, want 0", observed.Pts)
|
||
}
|
||
|
||
// 客户端下一次明确带回 pts=2 后，才允许 retention 把共同安全水位推进到 2。
|
||
if _, err := svc.ObserveDifferenceRequest(ctx, authKeyID, userID, domain.UpdateState{Pts: 2, Date: 1700000102}); err != nil {
|
||
t.Fatalf("observing next request: %v", err)
|
||
}
|
||
observed, ok = states.ObservedClientState(authKeyID, userID)
|
||
if !ok || observed.Pts != 2 {
|
||
t.Fatalf("observed after client carried cursor = %+v/%v, want pts=2", observed, ok)
|
||
}
|
||
}
|
||
|
||
type retentionCheckpointEvents struct {
|
||
*memory.UpdateEventStore
|
||
pts int
|
||
date int
|
||
current int
|
||
missFirst bool
|
||
calls int
|
||
}
|
||
|
||
func (s *retentionCheckpointEvents) UserUpdateRetentionCheckpoint(_ context.Context, _ [8]byte, _ int64) (int, int, bool, error) {
|
||
s.calls++
|
||
if s.missFirst && s.calls == 1 {
|
||
return 0, 0, false, nil
|
||
}
|
||
return s.pts, s.date, s.pts > 0, nil
|
||
}
|
||
|
||
func (s *retentionCheckpointEvents) MaxContiguousPts(_ context.Context, _ int64) (int, error) {
|
||
return s.current, nil
|
||
}
|
||
|
||
func TestGetDifferenceBelowRetainedFloorUsesEmptySliceCheckpoint(t *testing.T) {
|
||
ctx := context.Background()
|
||
authKeyID := [8]byte{13}
|
||
const userID int64 = 1000000013
|
||
base := memory.NewUpdateEventStore()
|
||
events := &retentionCheckpointEvents{UpdateEventStore: base, pts: 2, date: 1700000202, current: 3}
|
||
states := memory.NewUpdateStateStore()
|
||
svc := NewService(states, events)
|
||
// Retention already removed pts 1..2; only the live tail remains.
|
||
if err := base.Append(ctx, userID, domain.UpdateEvent{
|
||
UserID: userID, Type: domain.UpdateEventNoop, Pts: 3, PtsCount: 1, Date: 1700000203,
|
||
}); err != nil {
|
||
t.Fatalf("append live tail: %v", err)
|
||
}
|
||
|
||
checkpoint, err := svc.GetDifference(ctx, authKeyID, userID, domain.UpdateState{Pts: 0, Date: 1700000200})
|
||
if err != nil {
|
||
t.Fatalf("difference below retained floor: %v", err)
|
||
}
|
||
if !checkpoint.Partial || len(checkpoint.Events) != 0 || checkpoint.State.Pts != 2 || checkpoint.State.Date != 1700000202 {
|
||
t.Fatalf("checkpoint difference = %+v, want empty differenceSlice at pts/date 2/1700000202", checkpoint)
|
||
}
|
||
if _, found, err := svc.ConfirmedState(ctx, authKeyID, userID); err != nil || found {
|
||
t.Fatalf("computed checkpoint advanced confirmed: found=%v err=%v", found, err)
|
||
}
|
||
if err := svc.CommitDeliveredState(ctx, authKeyID, userID, checkpoint.State, domain.UpdateStateCommitDeliveredOnly); err != nil {
|
||
t.Fatalf("commit delivered checkpoint: %v", err)
|
||
}
|
||
if confirmed, found, err := svc.ConfirmedState(ctx, authKeyID, userID); err != nil || !found || confirmed.Pts != 2 {
|
||
t.Fatalf("confirmed checkpoint = %+v/%v err=%v, want pts=2", confirmed, found, err)
|
||
}
|
||
|
||
tail, err := svc.GetDifference(ctx, authKeyID, userID, checkpoint.State)
|
||
if err != nil {
|
||
t.Fatalf("difference from retained floor: %v", err)
|
||
}
|
||
if tail.Partial || len(tail.Events) != 1 || tail.Events[0].Pts != 3 || tail.State.Pts != 3 {
|
||
t.Fatalf("tail difference = %+v, want normal event pts=3", tail)
|
||
}
|
||
}
|
||
|
||
func TestGetDifferenceRechecksCheckpointWhenRetentionRacesEventRead(t *testing.T) {
|
||
ctx := context.Background()
|
||
authKeyID := [8]byte{14}
|
||
const userID int64 = 1000000014
|
||
base := memory.NewUpdateEventStore()
|
||
events := &retentionCheckpointEvents{
|
||
UpdateEventStore: base,
|
||
pts: 2,
|
||
date: 1700000302,
|
||
current: 3,
|
||
missFirst: true,
|
||
}
|
||
if err := base.Append(ctx, userID, domain.UpdateEvent{
|
||
UserID: userID, Type: domain.UpdateEventNoop, Pts: 3, PtsCount: 1, Date: 1700000303,
|
||
}); err != nil {
|
||
t.Fatalf("append live tail: %v", err)
|
||
}
|
||
|
||
diff, err := NewService(memory.NewUpdateStateStore(), events).GetDifference(
|
||
ctx,
|
||
authKeyID,
|
||
userID,
|
||
domain.UpdateState{Pts: 0, Date: 1700000300},
|
||
)
|
||
if err != nil {
|
||
t.Fatalf("difference across retention race: %v", err)
|
||
}
|
||
if events.calls != 2 {
|
||
t.Fatalf("checkpoint probes = %d, want pre-read plus post-gap recheck", events.calls)
|
||
}
|
||
if !diff.Partial || len(diff.Events) != 0 || diff.State.Pts != 2 || diff.State.Date != 1700000302 {
|
||
t.Fatalf("race checkpoint difference = %+v, want empty differenceSlice at retained floor", diff)
|
||
}
|
||
}
|
||
|
||
type captureDispatchAppender struct {
|
||
*memory.UpdateEventStore
|
||
dispatched bool
|
||
userID int64
|
||
event domain.UpdateEvent
|
||
excludeAuthKeyID [8]byte
|
||
excludeSessionID int64
|
||
}
|
||
|
||
func (s *captureDispatchAppender) AppendAllocatedWithDispatch(ctx context.Context, userID int64, event domain.UpdateEvent, excludeAuthKeyID [8]byte, excludeSessionID int64) (domain.UpdateEvent, error) {
|
||
s.dispatched = true
|
||
s.userID = userID
|
||
s.excludeAuthKeyID = excludeAuthKeyID
|
||
s.excludeSessionID = excludeSessionID
|
||
event, err := s.UpdateEventStore.AppendAllocated(ctx, userID, event)
|
||
s.event = event
|
||
return event, err
|
||
}
|
||
|
||
var errDispatchFailed = errors.New("dispatch failed")
|
||
|
||
type failingDispatchAppender struct {
|
||
*memory.UpdateEventStore
|
||
}
|
||
|
||
func (s *failingDispatchAppender) AppendAllocatedWithDispatch(context.Context, int64, domain.UpdateEvent, [8]byte, int64) (domain.UpdateEvent, error) {
|
||
return domain.UpdateEvent{}, errDispatchFailed
|
||
}
|
||
|
||
type captureStateStore struct {
|
||
saveCount int
|
||
lastSaveAuthKeyID [8]byte
|
||
states map[[16]byte]domain.UpdateState
|
||
}
|
||
|
||
func (s *captureStateStore) Get(_ context.Context, authKeyID [8]byte, userID int64) (domain.UpdateState, bool, error) {
|
||
if s.states == nil {
|
||
return domain.UpdateState{}, false, nil
|
||
}
|
||
st, ok := s.states[captureStateKey(authKeyID, userID)]
|
||
return st, ok, nil
|
||
}
|
||
|
||
func (s *captureStateStore) Save(_ context.Context, authKeyID [8]byte, userID int64, state domain.UpdateState) error {
|
||
if s.states == nil {
|
||
s.states = make(map[[16]byte]domain.UpdateState)
|
||
}
|
||
s.saveCount++
|
||
s.lastSaveAuthKeyID = authKeyID
|
||
s.states[captureStateKey(authKeyID, userID)] = state
|
||
return nil
|
||
}
|
||
|
||
func (s *captureStateStore) CommitDeliveredState(ctx context.Context, authKeyID [8]byte, userID int64, state domain.UpdateState, _ domain.UpdateStateCommitMode) error {
|
||
return s.Save(ctx, authKeyID, userID, state)
|
||
}
|
||
|
||
func (s *captureStateStore) ObserveClientState(_ context.Context, _ [8]byte, _ int64, _ domain.UpdateState) error {
|
||
return nil
|
||
}
|
||
|
||
func (s *captureStateStore) Delete(_ context.Context, authKeyID [8]byte, userID int64) error {
|
||
if s.states != nil {
|
||
delete(s.states, captureStateKey(authKeyID, userID))
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (s *captureStateStore) DeleteAuthKey(_ context.Context, authKeyID [8]byte) error {
|
||
if s.states == nil {
|
||
return nil
|
||
}
|
||
for key := range s.states {
|
||
var got [8]byte
|
||
copy(got[:], key[:8])
|
||
if got == authKeyID {
|
||
delete(s.states, key)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func captureStateKey(authKeyID [8]byte, userID int64) [16]byte {
|
||
var key [16]byte
|
||
copy(key[:8], authKeyID[:])
|
||
for i := 0; i < 8; i++ {
|
||
key[8+i] = byte(userID >> (8 * i))
|
||
}
|
||
return key
|
||
}
|