owpengram-server/internal/rpc/channels_updates_rpc_test.go

package rpc

import (
	"context"
	"strings"
	appchannels "telesrv/internal/app/channels"
	appupdates "telesrv/internal/app/updates"
	appusers "telesrv/internal/app/users"
	"telesrv/internal/domain"
	"telesrv/internal/store/memory"
	"testing"

	"github.com/iamxvbaba/td/clock"
	"github.com/iamxvbaba/td/proto"
	"github.com/iamxvbaba/td/tg"
	"go.uber.org/zap/zaptest"
)

func TestChannelRealtimeRecipientsPreferOnlineMembers(t *testing.T) {
	ctx := context.Background()
	channelStore := memory.NewChannelStore()
	channelService := appchannels.NewService(channelStore)
	created, err := channelService.CreateMegagroupFromCreateChat(ctx, 1001, domain.CreateChannelRequest{
		Title:         "Fanout",
		MemberUserIDs: []int64{1002, 1999},
		Date:          10,
	})
	if err != nil {
		t.Fatalf("create channel: %v", err)
	}
	sessions := &captureSessions{
		onlineUserIDs:  []int64{1999, 3000},
		channelViewers: map[int64][]int64{created.Channel.ID: {1999}},
		channelMembers: map[int64][]int64{created.Channel.ID: {1999, 3000}},
	}
	r := New(Config{}, Deps{
		Channels: channelService,
		Sessions: sessions,
	}, zaptest.NewLogger(t), clock.System)
	contains := func(items []int64, want int64) bool {
		for _, item := range items {
			if item == want {
				return true
			}
		}
		return false
	}

	got := r.channelFanoutRecipients(ctx, channelFanoutMembers, created.Channel.ID, []int64{1002})
	if !contains(got, 1999) {
		t.Fatalf("recipients = %v, want online active member 1999", got)
	}
	if contains(got, 3000) {
		t.Fatalf("recipients = %v, non-member online user leaked", got)
	}
	if !contains(got, 1002) {
		t.Fatalf("recipients = %v, want explicit fallback recipient 1002", got)
	}
	onlines, err := r.onMessagesGetOnlines(WithUserID(ctx, 1001), &tg.InputPeerChannel{
		ChannelID:  created.Channel.ID,
		AccessHash: created.Channel.AccessHash,
	})
	if err != nil {
		t.Fatalf("messages.getOnlines: %v", err)
	}
	if onlines.Onlines != 2 {
		t.Fatalf("messages.getOnlines = %d, want caller plus online active member", onlines.Onlines)
	}
}

func TestChannelSendHistoryAndDifferenceRPC(t *testing.T) {
	ctx := context.Background()
	userStore := memory.NewUserStore()
	owner, _ := userStore.Create(ctx, domain.User{AccessHash: 11, Phone: "15550002001", FirstName: "Owner"})
	friend, _ := userStore.Create(ctx, domain.User{AccessHash: 22, Phone: "15550002002", FirstName: "Friend"})
	channelStore := memory.NewChannelStore()
	sessions := &captureScopedSessions{captureSessions: &captureSessions{}}
	r := New(Config{}, Deps{
		Users:    appusers.NewService(userStore),
		Channels: appchannels.NewService(channelStore),
		Sessions: sessions,
	}, zaptest.NewLogger(t), clock.System)
	created, err := r.onMessagesCreateChat(WithUserID(ctx, owner.ID), &tg.MessagesCreateChatRequest{
		Users: []tg.InputUserClass{&tg.InputUser{UserID: friend.ID, AccessHash: friend.AccessHash}},
		Title: "RPC Group",
	})
	if err != nil {
		t.Fatalf("create chat: %v", err)
	}
	channel := created.Updates.(*tg.Updates).Chats[0].(*tg.Channel)

	var authKeyID [8]byte
	authKeyID[0] = 9
	sendCtx := WithSessionID(WithAuthKeyID(WithUserID(ctx, owner.ID), authKeyID), 77)
	sent, err := r.onMessagesSendMessage(sendCtx, &tg.MessagesSendMessageRequest{
		Peer:     &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Message:  "hello channel",
		RandomID: 99,
	})
	if err != nil {
		t.Fatalf("send channel message: %v", err)
	}
	sendUpdates := sent.(*tg.Updates)
	if id, ok := sendUpdates.Updates[0].(*tg.UpdateMessageID); !ok || id.ID != 3 || id.RandomID != 99 {
		t.Fatalf("message id update = %#v, want id=3 random_id=99", sendUpdates.Updates[0])
	}
	newMsg, ok := sendUpdates.Updates[1].(*tg.UpdateNewChannelMessage)
	if !ok || newMsg.Pts != 4 || newMsg.PtsCount != 1 {
		t.Fatalf("new channel update = %#v, want pts=4", sendUpdates.Updates[1])
	}
	msg := newMsg.Message.(*tg.Message)
	if msg.PeerID.(*tg.PeerChannel).ChannelID != channel.ID || msg.Message != "hello channel" || !msg.Out {
		t.Fatalf("channel message = %#v, want outgoing channel text", msg)
	}
	pushed := sessions.snapshot()
	if pushed.userID != friend.ID || pushed.sessionID != 77 || pushed.messageType != proto.MessageFromServer {
		t.Fatalf("pushed channel update = user %d exclude session %d type %v, want friend/exclude/from_server", pushed.userID, pushed.sessionID, pushed.messageType)
	}
	if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != authKeyID {
		t.Fatalf("exclude auth_key_id = %x, want %x", gotAuthKeyID, authKeyID)
	}
	pushedUpdates, ok := pushed.message.(*tg.Updates)
	if !ok || len(pushedUpdates.Updates) != 1 {
		t.Fatalf("pushed channel update = %T %+v, want one updates container without updateMessageID", pushed.message, pushed.message)
	}
	pushedNew, ok := pushedUpdates.Updates[0].(*tg.UpdateNewChannelMessage)
	if !ok {
		t.Fatalf("pushed update[0] = %T, want updateNewChannelMessage", pushedUpdates.Updates[0])
	}
	pushedMsg := pushedNew.Message.(*tg.Message)
	if pushedMsg.Out || pushedMsg.Message != "hello channel" {
		t.Fatalf("pushed message = %#v, want incoming channel text for friend", pushedMsg)
	}

	history, err := r.onChannelsGetMessages(WithUserID(ctx, friend.ID), &tg.ChannelsGetMessagesRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:      []tg.InputMessageClass{&tg.InputMessageID{ID: msg.ID}},
	})
	if err != nil {
		t.Fatalf("get channel messages: %v", err)
	}
	messages := history.(*tg.MessagesMessages)
	got := messages.Messages[0].(*tg.Message)
	if got.Message != "hello channel" || got.Out {
		t.Fatalf("history message = %#v, want incoming text for friend", got)
	}

	contentAuthKeyID := [8]byte{0x44}
	contentCtx := WithSessionID(WithAuthKeyID(WithUserID(ctx, friend.ID), contentAuthKeyID), 88)
	if ok, err := r.onChannelsReadMessageContents(contentCtx, &tg.ChannelsReadMessageContentsRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:      []int{msg.ID},
	}); err != nil || !ok {
		t.Fatalf("channels.readMessageContents = ok %v err %v, want true", ok, err)
	}
	contentPush := sessions.snapshot()
	if contentPush.userID != friend.ID || contentPush.sessionID != 88 || contentPush.messageType != proto.MessageFromServer {
		t.Fatalf("content-read push = user %d exclude session %d type %v, want friend/exclude/from_server", contentPush.userID, contentPush.sessionID, contentPush.messageType)
	}
	if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != contentAuthKeyID {
		t.Fatalf("content-read exclude auth_key_id = %x, want %x", gotAuthKeyID, contentAuthKeyID)
	}
	contentUpdates, ok := contentPush.message.(*tg.Updates)
	if !ok || len(contentUpdates.Updates) != 1 {
		t.Fatalf("content-read pushed message = %T %+v, want one update", contentPush.message, contentPush.message)
	}
	contentRead, ok := contentUpdates.Updates[0].(*tg.UpdateChannelReadMessagesContents)
	if !ok || contentRead.ChannelID != channel.ID || len(contentRead.Messages) != 1 || contentRead.Messages[0] != msg.ID {
		t.Fatalf("content-read update = %#v, want channel %d msg %d", contentUpdates.Updates[0], channel.ID, msg.ID)
	}

	diff, err := r.onUpdatesGetChannelDifference(WithUserID(ctx, friend.ID), &tg.UpdatesGetChannelDifferenceRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Filter:  &tg.ChannelMessagesFilterEmpty{},
		Pts:     newMsg.Pts - 1,
		Limit:   10,
	})
	if err != nil {
		t.Fatalf("channel difference: %v", err)
	}
	fullDiff, ok := diff.(*tg.UpdatesChannelDifference)
	if !ok || fullDiff.Pts != newMsg.Pts || len(fullDiff.NewMessages) != 1 {
		t.Fatalf("diff = %T %+v, want one new message at pts=%d", diff, diff, newMsg.Pts)
	}
	if _, err := r.onUpdatesGetChannelDifference(WithUserID(ctx, friend.ID), &tg.UpdatesGetChannelDifferenceRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Filter:  &tg.ChannelMessagesFilterEmpty{},
		Pts:     fullDiff.Pts + 1,
		Limit:   10,
	}); err == nil || !strings.Contains(err.Error(), "PERSISTENT_TIMESTAMP_INVALID") {
		t.Fatalf("future channel pts err = %v, want PERSISTENT_TIMESTAMP_INVALID", err)
	}

	readOK, err := r.onChannelsReadHistory(WithUserID(ctx, friend.ID), &tg.ChannelsReadHistoryRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		MaxID:   msg.ID,
	})
	if err != nil || !readOK {
		t.Fatalf("channels.readHistory = %v err %v, want true", readOK, err)
	}
	readPush := sessions.snapshot()
	if readPush.userID != owner.ID || readPush.messageType != proto.MessageFromServer {
		t.Fatalf("read outbox push = user %d type %v, want owner/from_server", readPush.userID, readPush.messageType)
	}
	readPushUpdates, ok := readPush.message.(*tg.Updates)
	if !ok || len(readPushUpdates.Updates) != 1 {
		t.Fatalf("read outbox pushed message = %T %+v, want one update", readPush.message, readPush.message)
	}
	readOutbox, ok := readPushUpdates.Updates[0].(*tg.UpdateReadChannelOutbox)
	if !ok || readOutbox.ChannelID != channel.ID || readOutbox.MaxID != msg.ID {
		t.Fatalf("read outbox update = %#v, want channel %d max %d", readPushUpdates.Updates[0], channel.ID, msg.ID)
	}
	fullAfterRead, err := r.onChannelsGetFullChannel(WithUserID(ctx, owner.ID), &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash})
	if err != nil {
		t.Fatalf("get full channel after read: %v", err)
	}
	fullChannel := fullAfterRead.FullChat.(*tg.ChannelFull)
	if fullChannel.ReadOutboxMaxID != msg.ID {
		t.Fatalf("full channel read_outbox = %d, want %d", fullChannel.ReadOutboxMaxID, msg.ID)
	}
	readers, err := r.onMessagesGetMessageReadParticipants(WithUserID(ctx, owner.ID), &tg.MessagesGetMessageReadParticipantsRequest{
		Peer:  &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		MsgID: msg.ID,
	})
	if err != nil {
		t.Fatalf("get message read participants: %v", err)
	}
	if len(readers) != 1 || readers[0].UserID != friend.ID || readers[0].Date == 0 {
		t.Fatalf("read participants = %+v, want friend read date", readers)
	}

	editReq := &tg.MessagesEditMessageRequest{
		Peer:    &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:      msg.ID,
		Message: "edited channel",
	}
	editReq.SetMessage("edited channel")
	edited, err := r.onMessagesEditMessage(WithUserID(ctx, owner.ID), editReq)
	if err != nil {
		t.Fatalf("edit channel message: %v", err)
	}
	editUpdates := edited.(*tg.Updates)
	edit, ok := editUpdates.Updates[0].(*tg.UpdateEditChannelMessage)
	if !ok || edit.Pts != newMsg.Pts+1 || edit.PtsCount != 1 {
		t.Fatalf("edit update = %#v, want updateEditChannelMessage pts=%d", editUpdates.Updates[0], newMsg.Pts+1)
	}
	if edit.Message.(*tg.Message).Message != "edited channel" {
		t.Fatalf("edited message = %#v, want edited text", edit.Message)
	}
	editData, err := r.onMessagesGetMessageEditData(WithUserID(ctx, owner.ID), &tg.MessagesGetMessageEditDataRequest{
		Peer: &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:   msg.ID,
	})
	if err != nil {
		t.Fatalf("get channel edit data: %v", err)
	}
	if editData.GetCaption() {
		t.Fatalf("channel edit data caption = true, want false for text-only message")
	}

	forwardReplyTo := &tg.InputReplyToMessage{ReplyToMsgID: msg.ID}
	forwardReplyTo.SetQuoteText("channel")
	forwardReq := &tg.MessagesForwardMessagesRequest{
		FromPeer: &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ToPeer:   &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:       []int{msg.ID},
		RandomID: []int64{100},
	}
	forwardReq.SetReplyTo(forwardReplyTo)
	forwarded, err := r.onMessagesForwardMessages(WithUserID(ctx, friend.ID), forwardReq)
	if err != nil {
		t.Fatalf("forward channel message: %v", err)
	}
	forwardUpdates := forwarded.(*tg.Updates)
	if id, ok := forwardUpdates.Updates[0].(*tg.UpdateMessageID); !ok || id.ID != msg.ID+1 || id.RandomID != 100 {
		t.Fatalf("forward id update = %#v, want id=%d", forwardUpdates.Updates[0], msg.ID+1)
	}
	forwardNew, ok := forwardUpdates.Updates[1].(*tg.UpdateNewChannelMessage)
	if !ok || forwardNew.Pts != edit.Pts+1 || forwardNew.PtsCount != 1 {
		t.Fatalf("forward new update = %#v, want pts=%d", forwardUpdates.Updates[1], edit.Pts+1)
	}
	forwardMsg := forwardNew.Message.(*tg.Message)
	if forwardMsg.Message != "edited channel" || forwardMsg.FwdFrom.FromID == nil {
		t.Fatalf("forward message = %#v, want fwd header and edited body", forwardMsg)
	}
	if header, ok := forwardMsg.ReplyTo.(*tg.MessageReplyHeader); !ok || header.ReplyToMsgID != msg.ID {
		t.Fatalf("forward reply header = %#v, want reply to channel message %d", forwardMsg.ReplyTo, msg.ID)
	}

	deleted, err := r.onChannelsDeleteMessages(WithUserID(ctx, owner.ID), &tg.ChannelsDeleteMessagesRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:      []int{msg.ID, forwardMsg.ID},
	})
	if err != nil {
		t.Fatalf("delete channel messages: %v", err)
	}
	if deleted.Pts != forwardNew.Pts+2 || deleted.PtsCount != 2 {
		t.Fatalf("delete affected = %+v, want pts=%d count=2", deleted, forwardNew.Pts+2)
	}

	diff, err = r.onUpdatesGetChannelDifference(WithUserID(ctx, friend.ID), &tg.UpdatesGetChannelDifferenceRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Filter:  &tg.ChannelMessagesFilterEmpty{},
		Pts:     newMsg.Pts,
		Limit:   10,
	})
	if err != nil {
		t.Fatalf("channel difference after edit/delete: %v", err)
	}
	fullDiff, ok = diff.(*tg.UpdatesChannelDifference)
	if !ok || fullDiff.Pts != deleted.Pts || len(fullDiff.NewMessages) != 1 || len(fullDiff.OtherUpdates) != 3 {
		t.Fatalf("diff after edit/delete = %T %+v, want forward message id mapping plus edit/delete updates", diff, diff)
	}
	// 差量首条 other update 是请求者自己消息的 updateMessageID：断线后
	// 经差量对账本地 pending，避免重复气泡。
	if mapping, ok := fullDiff.OtherUpdates[0].(*tg.UpdateMessageID); !ok || mapping.RandomID == 0 || mapping.ID == 0 {
		t.Fatalf("diff other[0] = %#v, want updateMessageID for own forwarded message", fullDiff.OtherUpdates[0])
	}
}

func TestChannelsReadMessageContentsClearsUnreadReactionAndPushesUpdate(t *testing.T) {
	ctx := context.Background()
	userStore := memory.NewUserStore()
	owner, _ := userStore.Create(ctx, domain.User{AccessHash: 41, Phone: "15550002141", FirstName: "Owner"})
	friend, _ := userStore.Create(ctx, domain.User{AccessHash: 42, Phone: "15550002142", FirstName: "Friend"})
	channelStore := memory.NewChannelStore()
	sessions := &captureScopedSessions{captureSessions: &captureSessions{}}
	r := New(Config{}, Deps{
		Users:    appusers.NewService(userStore),
		Channels: appchannels.NewService(channelStore),
		Sessions: sessions,
	}, zaptest.NewLogger(t), clock.System)
	created, err := r.onMessagesCreateChat(WithUserID(ctx, owner.ID), &tg.MessagesCreateChatRequest{
		Users: []tg.InputUserClass{&tg.InputUser{UserID: friend.ID, AccessHash: friend.AccessHash}},
		Title: "Reaction Read",
	})
	if err != nil {
		t.Fatalf("create chat: %v", err)
	}
	channel := created.Updates.(*tg.Updates).Chats[0].(*tg.Channel)
	sent, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), &tg.MessagesSendMessageRequest{
		Peer:     &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Message:  "owner message",
		RandomID: 21041,
	})
	if err != nil {
		t.Fatalf("send channel message: %v", err)
	}
	msgID := sent.(*tg.Updates).Updates[1].(*tg.UpdateNewChannelMessage).Message.(*tg.Message).ID
	req := &tg.MessagesSendReactionRequest{
		Peer:     &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		MsgID:    msgID,
		Reaction: []tg.ReactionClass{&tg.ReactionEmoji{Emoticon: "\U0001f525"}},
	}
	req.SetReaction(req.Reaction)
	if _, err := r.onMessagesSendReaction(WithUserID(ctx, friend.ID), req); err != nil {
		t.Fatalf("friend send reaction: %v", err)
	}
	unread, err := r.onMessagesGetUnreadReactions(WithUserID(ctx, owner.ID), &tg.MessagesGetUnreadReactionsRequest{
		Peer:  &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Limit: 10,
	})
	if err != nil {
		t.Fatalf("get unread reactions: %v", err)
	}
	unreadMessages, _, _ := searchMessagesPayload(t, unread)
	if len(unreadMessages) != 1 {
		t.Fatalf("unread reactions = %+v, want one message", unread)
	}

	contentAuthKeyID := [8]byte{0x66}
	contentCtx := WithSessionID(WithAuthKeyID(WithUserID(ctx, owner.ID), contentAuthKeyID), 99)
	if ok, err := r.onChannelsReadMessageContents(contentCtx, &tg.ChannelsReadMessageContentsRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:      []int{msgID},
	}); err != nil || !ok {
		t.Fatalf("channels.readMessageContents = ok %v err %v, want true", ok, err)
	}
	pushed := sessions.snapshot()
	if pushed.userID != owner.ID || pushed.sessionID != 99 || pushed.messageType != proto.MessageFromServer {
		t.Fatalf("reaction read push = user %d session %d type %v, want owner/exclude/from_server", pushed.userID, pushed.sessionID, pushed.messageType)
	}
	if gotAuthKeyID := sessions.scopedAuthKey(); gotAuthKeyID != contentAuthKeyID {
		t.Fatalf("reaction read exclude auth_key_id = %x, want %x", gotAuthKeyID, contentAuthKeyID)
	}
	pushedUpdates, ok := pushed.message.(*tg.Updates)
	if !ok || len(pushedUpdates.Updates) != 1 {
		t.Fatalf("reaction read push = %T %+v, want one updateMessageReactions", pushed.message, pushed.message)
	}
	reactionUpdate, ok := pushedUpdates.Updates[0].(*tg.UpdateMessageReactions)
	if !ok || reactionUpdate.MsgID != msgID {
		t.Fatalf("reaction read update = %#v, want updateMessageReactions for %d", pushedUpdates.Updates[0], msgID)
	}
	for _, recent := range reactionUpdate.Reactions.RecentReactions {
		if recent.Unread {
			t.Fatalf("reaction read update recent = %+v, want unread cleared", reactionUpdate.Reactions.RecentReactions)
		}
	}
	unreadAfter, err := r.onMessagesGetUnreadReactions(WithUserID(ctx, owner.ID), &tg.MessagesGetUnreadReactionsRequest{
		Peer:  &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Limit: 10,
	})
	if err != nil {
		t.Fatalf("get unread reactions after read contents: %v", err)
	}
	unreadAfterMessages, _, _ := searchMessagesPayload(t, unreadAfter)
	if len(unreadAfterMessages) != 0 {
		t.Fatalf("unread reactions after read contents = %+v, want empty", unreadAfter)
	}
}

func TestChannelReadHistoryProducesReadChannelInboxDifference(t *testing.T) {
	ctx := context.Background()
	userStore := memory.NewUserStore()
	owner, _ := userStore.Create(ctx, domain.User{AccessHash: 31, Phone: "15550002131", FirstName: "Owner"})
	friend, _ := userStore.Create(ctx, domain.User{AccessHash: 32, Phone: "15550002132", FirstName: "Friend"})
	channelStore := memory.NewChannelStore()
	updates := appupdates.NewService(memory.NewUpdateStateStore(), memory.NewUpdateEventStore())
	r := New(Config{}, Deps{
		Users:    appusers.NewService(userStore),
		Channels: appchannels.NewService(channelStore),
		Updates:  updates,
		Sessions: &captureSessions{},
	}, zaptest.NewLogger(t), clock.System)
	created, err := r.onMessagesCreateChat(WithUserID(ctx, owner.ID), &tg.MessagesCreateChatRequest{
		Users: []tg.InputUserClass{&tg.InputUser{UserID: friend.ID, AccessHash: friend.AccessHash}},
		Title: "Read Channel Inbox",
	})
	if err != nil {
		t.Fatalf("create chat: %v", err)
	}
	channel := created.Updates.(*tg.Updates).Chats[0].(*tg.Channel)
	sent, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), &tg.MessagesSendMessageRequest{
		Peer:     &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Message:  "read me",
		RandomID: 301,
	})
	if err != nil {
		t.Fatalf("send channel message: %v", err)
	}
	newUpdate := sent.(*tg.Updates).Updates[1].(*tg.UpdateNewChannelMessage)
	msg := newUpdate.Message.(*tg.Message)
	readOK, err := r.onChannelsReadHistory(WithUserID(ctx, friend.ID), &tg.ChannelsReadHistoryRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		MaxID:   msg.ID,
	})
	if err != nil {
		t.Fatalf("channels.readHistory: %v", err)
	}
	if !readOK {
		t.Fatalf("channels.readHistory = false, want true")
	}
	diff, err := r.onUpdatesGetDifference(WithUserID(ctx, friend.ID), &tg.UpdatesGetDifferenceRequest{Pts: 0})
	if err != nil {
		t.Fatalf("updates.getDifference: %v", err)
	}
	full, ok := diff.(*tg.UpdatesDifference)
	if !ok || len(full.OtherUpdates) != 1 {
		t.Fatalf("difference = %T %+v, want one read channel inbox update", diff, diff)
	}
	read, ok := full.OtherUpdates[0].(*tg.UpdateReadChannelInbox)
	if !ok || read.ChannelID != channel.ID || read.MaxID != msg.ID || read.StillUnreadCount != 0 {
		t.Fatalf("difference update = %#v, want updateReadChannelInbox channel %d max %d", full.OtherUpdates[0], channel.ID, msg.ID)
	}
	if read.Pts != newUpdate.Pts {
		t.Fatalf("difference channel read pts = %d, want channel pts %d", read.Pts, newUpdate.Pts)
	}
	if len(full.Chats) != 1 {
		t.Fatalf("difference chats = %d, want channel context", len(full.Chats))
	}
}

func TestChannelDifferenceTooLongCarriesDialogPts(t *testing.T) {
	ctx := context.Background()
	userStore := memory.NewUserStore()
	owner, _ := userStore.Create(ctx, domain.User{AccessHash: 61, Phone: "15550002161", FirstName: "Owner"})
	friend, _ := userStore.Create(ctx, domain.User{AccessHash: 62, Phone: "15550002162", FirstName: "Friend"})
	channelStore := memory.NewChannelStore()
	r := New(Config{}, Deps{
		Users:    appusers.NewService(userStore),
		Channels: appchannels.NewService(channelStore),
	}, zaptest.NewLogger(t), clock.System)
	created, err := r.onMessagesCreateChat(WithUserID(ctx, owner.ID), &tg.MessagesCreateChatRequest{
		Users: []tg.InputUserClass{&tg.InputUser{UserID: friend.ID, AccessHash: friend.AccessHash}},
		Title: "RPC TooLong Group",
	})
	if err != nil {
		t.Fatalf("create chat: %v", err)
	}
	channel := created.Updates.(*tg.Updates).Chats[0].(*tg.Channel)
	sourceCreated, err := r.onChannelsCreateChannel(WithUserID(ctx, owner.ID), &tg.ChannelsCreateChannelRequest{
		Title:     "RPC TooLong Source",
		About:     "forward source",
		Broadcast: true,
	})
	if err != nil {
		t.Fatalf("create source channel: %v", err)
	}
	sourceChannel := sourceCreated.(*tg.Updates).Chats[0].(*tg.Channel)
	if _, err := r.onChannelsInviteToChannel(WithUserID(ctx, owner.ID), &tg.ChannelsInviteToChannelRequest{
		Channel: &tg.InputChannel{ChannelID: sourceChannel.ID, AccessHash: sourceChannel.AccessHash},
		Users:   []tg.InputUserClass{&tg.InputUser{UserID: friend.ID, AccessHash: friend.AccessHash}},
	}); err != nil {
		t.Fatalf("invite friend to source channel: %v", err)
	}
	sourceSent, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), &tg.MessagesSendMessageRequest{
		Peer:     &tg.InputPeerChannel{ChannelID: sourceChannel.ID, AccessHash: sourceChannel.AccessHash},
		Message:  "forward source",
		RandomID: 7000,
	})
	if err != nil {
		t.Fatalf("send source channel message: %v", err)
	}
	sourceMsgID := sourceSent.(*tg.Updates).Updates[0].(*tg.UpdateMessageID).ID
	if _, err := r.onMessagesForwardMessages(WithUserID(ctx, owner.ID), &tg.MessagesForwardMessagesRequest{
		FromPeer: &tg.InputPeerChannel{ChannelID: sourceChannel.ID, AccessHash: sourceChannel.AccessHash},
		ToPeer:   &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		ID:       []int{sourceMsgID},
		RandomID: []int64{7001},
	}); err != nil {
		t.Fatalf("forward source channel to target channel: %v", err)
	}
	for i := 0; i < 12; i++ {
		if _, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), &tg.MessagesSendMessageRequest{
			Peer:     &tg.InputPeerChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
			Message:  "too long page",
			RandomID: int64(i + 1),
		}); err != nil {
			t.Fatalf("send channel message %d: %v", i, err)
		}
	}
	diff, err := r.onUpdatesGetChannelDifference(WithUserID(ctx, friend.ID), &tg.UpdatesGetChannelDifferenceRequest{
		Channel: &tg.InputChannel{ChannelID: channel.ID, AccessHash: channel.AccessHash},
		Filter:  &tg.ChannelMessagesFilterEmpty{},
		Pts:     0,
		Limit:   3,
	})
	if err != nil {
		t.Fatalf("channel difference: %v", err)
	}
	tooLong, ok := diff.(*tg.UpdatesChannelDifferenceTooLong)
	if !ok {
		t.Fatalf("diff = %T %+v, want channelDifferenceTooLong", diff, diff)
	}
	dialog, ok := tooLong.Dialog.(*tg.Dialog)
	if !ok {
		t.Fatalf("tooLong dialog = %T, want dialog", tooLong.Dialog)
	}
	pts, ok := dialog.GetPts()
	if !ok || pts == 0 {
		t.Fatalf("tooLong dialog pts = %d ok=%v, want current channel pts", pts, ok)
	}
	if len(tooLong.Messages) == 0 || len(tooLong.Messages) > domain.MaxChannelDifferenceTooLongMessages {
		t.Fatalf("tooLong messages = %d, want bounded latest snapshot", len(tooLong.Messages))
	}
	if len(tooLong.Chats) == 0 || tooLong.Chats[0].(*tg.Channel).ID != channel.ID {
		t.Fatalf("tooLong chats = %+v, want source channel context", tooLong.Chats)
	}
	hasSourceChannel := false
	for _, chat := range tooLong.Chats {
		if ch, ok := chat.(*tg.Channel); ok && ch.ID == sourceChannel.ID {
			hasSourceChannel = true
			break
		}
	}
	if !hasSourceChannel {
		t.Fatalf("tooLong chats = %+v, want forwarded source channel context %d", tooLong.Chats, sourceChannel.ID)
	}
	hasOwnerUser := false
	for _, user := range tooLong.Users {
		if u, ok := user.(*tg.User); ok && u.ID == owner.ID {
			hasOwnerUser = true
			break
		}
	}
	if !hasOwnerUser {
		t.Fatalf("tooLong users = %+v, want sender user context", tooLong.Users)
	}
}

func TestChannelUnreadMentionsRPCUsesMentionState(t *testing.T) {
	ctx := context.Background()
	userStore := memory.NewUserStore()
	owner, _ := userStore.Create(ctx, domain.User{AccessHash: 9101, Phone: "15550009101", FirstName: "Owner", Username: "owner_mention"})
	member, _ := userStore.Create(ctx, domain.User{AccessHash: 9102, Phone: "15550009102", FirstName: "Mentioned", Username: "mention_friend"})
	channelStore := memory.NewChannelStore()
	channelService := appchannels.NewService(channelStore)
	r := New(Config{}, Deps{
		Users:    appusers.NewService(userStore),
		Channels: channelService,
	}, zaptest.NewLogger(t), clock.System)
	created, err := channelService.CreateMegagroupFromCreateChat(ctx, owner.ID, domain.CreateChannelRequest{
		Title:         "Mention RPC",
		MemberUserIDs: []int64{member.ID},
		Date:          1700009101,
	})
	if err != nil {
		t.Fatalf("create megagroup: %v", err)
	}
	peer := &tg.InputPeerChannel{ChannelID: created.Channel.ID, AccessHash: created.Channel.AccessHash}
	if _, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), &tg.MessagesSendMessageRequest{
		Peer:     peer,
		Message:  "hello @mention_friend",
		RandomID: 9102001,
	}); err != nil {
		t.Fatalf("send mention: %v", err)
	}
	mentions, err := r.onMessagesGetUnreadMentions(WithUserID(ctx, member.ID), &tg.MessagesGetUnreadMentionsRequest{
		Peer:      peer,
		OffsetID:  1,
		AddOffset: -10,
		Limit:     10,
	})
	if err != nil {
		t.Fatalf("messages.getUnreadMentions: %v", err)
	}
	mentionMessages, _, _ := searchMessagesPayload(t, mentions)
	if len(mentionMessages) != 1 {
		t.Fatalf("messages.getUnreadMentions = %T %+v, want one unread mention", mentions, mentions)
	}
	if msg := mentionMessages[0].(*tg.Message); msg.Message != "hello @mention_friend" {
		t.Fatalf("mention message = %#v, want sent mention", msg)
	}
	read, err := r.onMessagesReadMentions(WithUserID(ctx, member.ID), &tg.MessagesReadMentionsRequest{Peer: peer})
	if err != nil {
		t.Fatalf("messages.readMentions: %v", err)
	}
	if read.Pts <= 0 || read.PtsCount != 0 || read.Offset != 0 {
		t.Fatalf("messages.readMentions = %+v, want current channel pts and no offset", read)
	}
	mentions, err = r.onMessagesGetUnreadMentions(WithUserID(ctx, member.ID), &tg.MessagesGetUnreadMentionsRequest{
		Peer:  peer,
		Limit: 10,
	})
	if err != nil {
		t.Fatalf("messages.getUnreadMentions after read: %v", err)
	}
	mentionMessages, _, _ = searchMessagesPayload(t, mentions)
	if got := len(mentionMessages); got != 0 {
		t.Fatalf("unread mentions after read = %d, want 0", got)
	}
}

// A channel post containing an @token that is not a syntactically valid
// username (too short, leading digit, etc.) must still send. Real Telegram
// renders it as a mention and only fails when the reader taps it; it never
// rejects the send. Regression for a 500 INTERNAL_SERVER_ERROR.
func TestChannelSendMessageWithUnresolvableMentionSucceeds(t *testing.T) {
	ctx := context.Background()
	userStore := memory.NewUserStore()
	owner, _ := userStore.Create(ctx, domain.User{AccessHash: 9201, Phone: "15550009201", FirstName: "Owner", Username: "owner_badmention"})
	channelStore := memory.NewChannelStore()
	channelService := appchannels.NewService(channelStore)
	r := New(Config{}, Deps{
		Users:    appusers.NewService(userStore),
		Channels: channelService,
	}, zaptest.NewLogger(t), clock.System)
	created, err := channelService.CreateMegagroupFromCreateChat(ctx, owner.ID, domain.CreateChannelRequest{
		Title: "Bad Mention",
		Date:  1700009201,
	})
	if err != nil {
		t.Fatalf("create megagroup: %v", err)
	}
	peer := &tg.InputPeerChannel{ChannelID: created.Channel.ID, AccessHash: created.Channel.AccessHash}
	for i, text := range []string{"look at @ziodotsh", "hi @2cool and @_x"} {
		if _, err := r.onMessagesSendMessage(WithUserID(ctx, owner.ID), &tg.MessagesSendMessageRequest{
			Peer:     peer,
			Message:  text,
			RandomID: int64(9202001 + i),
		}); err != nil {
			t.Fatalf("send %q: unexpected error %v", text, err)
		}
	}
}

func TestChannelDifferenceIncludesExtraForwardSourceChannel(t *testing.T) {
	channel := domain.Channel{ID: 2000000100, AccessHash: 9010, Title: "Megagroup", Megagroup: true, Date: 1700000000, Pts: 3}
	source := domain.Channel{ID: 2000000101, AccessHash: 9011, Title: "Source", Broadcast: true, Date: 1700000000}
	got, ok := tgChannelDifference(1000000001, domain.ChannelDifference{
		Channel: channel,
		NewMessages: []domain.ChannelMessage{{
			ChannelID:    channel.ID,
			ID:           3,
			SenderUserID: 1000000002,
			From:         domain.Peer{Type: domain.PeerTypeUser, ID: 1000000002},
			Date:         1700000103,
			Body:         "forwarded",
			Forward:      &domain.MessageForward{From: domain.Peer{Type: domain.PeerTypeChannel, ID: source.ID}, Date: 1700000000},
			Pts:          3,
		}},
		Users:    []domain.User{{ID: 1000000002, AccessHash: 42, FirstName: "Bob"}},
		Channels: []domain.Channel{source},
		Pts:      3,
		Final:    true,
		Timeout:  30,
	}).(*tg.UpdatesChannelDifference)
	if !ok {
		t.Fatalf("difference = %T, want *tg.UpdatesChannelDifference", got)
	}
	if len(got.Users) != 1 || len(got.Chats) != 2 {
		t.Fatalf("difference users/chats = %d/%d, want 1/2", len(got.Users), len(got.Chats))
	}
	if ch, ok := got.Chats[1].(*tg.Channel); !ok || ch.ID != source.ID {
		t.Fatalf("extra chat = %#v, want source channel", got.Chats[1])
	}
}

func TestChannelReadHistoryWithReliableDispatchPushesCurrentSessionReadUpdate(t *testing.T) {
	var authKeyID [8]byte
	authKeyID[0] = 10
	updates := &captureUpdates{
		state:            domain.UpdateState{Pts: 900, Date: 1700000102, Seq: 3},
		reliableDispatch: true,
	}
	sessions := &captureSessions{}
	r := New(Config{}, Deps{Updates: updates, Sessions: sessions}, zaptest.NewLogger(t), clock.System)
	ctx := WithSessionID(WithAuthKeyID(context.Background(), authKeyID), 77)

	recorded, err := r.recordChannelReadInbox(ctx, 1000000001, domain.ReadChannelHistoryResult{
		ChannelID:        12345,
		MaxID:            27,
		StillUnreadCount: 2,
		Changed:          true,
		Pts:              42,
	})
	if err != nil {
		t.Fatalf("record channel read inbox: %v", err)
	}
	if recorded.Pts != 900 || recorded.ChannelPts != 42 || updates.excludeSessionID != 77 {
		t.Fatalf("recorded event pts/channel_pts/session = %d/%d/%d, want durable pts 900, channel pts 42 and exclude session 77", recorded.Pts, recorded.ChannelPts, updates.excludeSessionID)
	}
	snap := sessions.snapshot()
	if snap.sessionID != 77 || snap.messageType != proto.MessageFromServer {
		t.Fatalf("current-session push target = session %d type %v, want session 77 server message", snap.sessionID, snap.messageType)
	}
	updatesMsg, ok := snap.message.(*tg.Updates)
	if !ok || len(updatesMsg.Updates) != 2 {
		t.Fatalf("current-session push = %T %+v, want read update plus account pts bookkeeping", snap.message, snap.message)
	}
	update, ok := updatesMsg.Updates[0].(*tg.UpdateReadChannelInbox)
	if !ok {
		t.Fatalf("current-session update = %T, want *tg.UpdateReadChannelInbox", updatesMsg.Updates[0])
	}
	if update.ChannelID != 12345 || update.Pts != 42 || update.MaxID != 27 || update.StillUnreadCount != 2 {
		t.Fatalf("current-session channel read update = %+v, want channel=12345 pts=42 max=27 still=2", update)
	}
	bookkeeping, ok := updatesMsg.Updates[1].(*tg.UpdateDeleteMessages)
	if !ok || len(bookkeeping.Messages) != 0 || bookkeeping.Pts != 900 || bookkeeping.PtsCount != 1 {
		t.Fatalf("current-session bookkeeping = %#v, want empty updateDeleteMessages at account pts 900", updatesMsg.Updates[1])
	}
}

func TestProjectChannelMentionForViewer(t *testing.T) {
	event := domain.ChannelUpdateEvent{
		ChannelID: 500,
		Pts:       7,
		PtsCount:  1,
		Message: domain.ChannelMessage{
			ChannelID:    500,
			ID:           42,
			SenderUserID: 1001,
			Body:         "hi @bob",
			Media:        &domain.MessageMedia{Kind: domain.MessageMediaKindPhoto, Photo: &domain.Photo{ID: 9}},
		},
	}
	mentioned := projectChannelMentionForViewer(event, []int64{1002}, 1002)
	if !mentioned.Message.Mentioned || !mentioned.Message.MediaUnread {
		t.Fatalf("mentioned viewer = %+v, want mentioned+media_unread set in realtime push", mentioned.Message)
	}
	other := projectChannelMentionForViewer(event, []int64{1002}, 1003)
	if other.Message.Mentioned || other.Message.MediaUnread {
		t.Fatalf("other viewer = %+v, must not inherit mention flags", other.Message)
	}
	sender := projectChannelMentionForViewer(event, []int64{1001}, 1001)
	if sender.Message.Mentioned {
		t.Fatalf("sender = %+v, must not be marked mentioned by own message", sender.Message)
	}
	if event.Message.Mentioned {
		t.Fatalf("source event mutated: projection must copy, not alias")
	}
}