From f4555782cf476d029e527e7202b14e6277f237e7 Mon Sep 17 00:00:00 2001 From: Tulir Asokan Date: Thu, 2 Apr 2026 00:28:10 +0300 Subject: [PATCH] handlematrix: fetch sponsored messages in channels after read receipt --- go.mod | 2 +- go.sum | 4 +- pkg/connector/handlematrix.go | 137 +++++++++++++++++++++++++++++++++- pkg/connector/metadata.go | 10 +++ 4 files changed, 147 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index 5c679f87..2098fd0b 100644 --- a/go.mod +++ b/go.mod @@ -27,7 +27,7 @@ require ( github.com/rs/zerolog v1.35.0 github.com/stretchr/testify v1.11.1 github.com/tidwall/gjson v1.18.0 - go.mau.fi/util v0.9.7 + go.mau.fi/util v0.9.8-0.20260401212211-58f3ab44ddae go.mau.fi/webp v0.2.0 go.mau.fi/zerozap v0.1.2 go.opentelemetry.io/otel v1.42.0 diff --git a/go.sum b/go.sum index 54194190..1ee7ea6c 100644 --- a/go.sum +++ b/go.sum @@ -112,8 +112,8 @@ github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/yuin/goldmark v1.8.2 h1:kEGpgqJXdgbkhcOgBxkC0X0PmoPG1ZyoZ117rDVp4zE= github.com/yuin/goldmark v1.8.2/go.mod h1:ip/1k0VRfGynBgxOz0yCqHrbZXhcjxyuS66Brc7iBKg= -go.mau.fi/util v0.9.7 h1:AWGNbJfz1zRcQOKeOEYhKUG2fT+/26Gy6kyqcH8tnBg= -go.mau.fi/util v0.9.7/go.mod h1:5T2f3ZWZFAGgmFwg3dGw7YK6kIsb9lryDzvynoR98pE= +go.mau.fi/util v0.9.8-0.20260401212211-58f3ab44ddae h1:93oGd3AZSzXIoSwdoWHvacSDZ5O+JcFNQwydDgGQYMM= +go.mau.fi/util v0.9.8-0.20260401212211-58f3ab44ddae/go.mod h1:5T2f3ZWZFAGgmFwg3dGw7YK6kIsb9lryDzvynoR98pE= go.mau.fi/webp v0.2.0 h1:QVMenHw7JDb4vall5sV75JNBQj9Hw4u8AKbi1QetHvg= go.mau.fi/webp v0.2.0/go.mod h1:VSg9MyODn12Mb5pyG0NIyNFhujrmoFSsZBs8syOZD1Q= go.mau.fi/zeroconfig v0.2.0 h1:e/OGEERqVRRKlgaro7E6bh8xXiKFSXB3eNNIud7FUjU= diff --git a/pkg/connector/handlematrix.go b/pkg/connector/handlematrix.go index 97c2bd18..84520b7b 100644 --- a/pkg/connector/handlematrix.go +++ b/pkg/connector/handlematrix.go @@ -17,6 +17,7 @@ package connector import ( + "bytes" "context" "crypto/sha256" "encoding/base64" @@ -37,11 +38,14 @@ import ( "time" "github.com/rs/zerolog" + "go.mau.fi/util/exsync" "go.mau.fi/util/ffmpeg" + "go.mau.fi/util/jsontime" "go.mau.fi/util/variationselector" "go.mau.fi/webp" "golang.org/x/exp/maps" _ "golang.org/x/image/webp" + "golang.org/x/net/html" "maunium.net/go/mautrix/bridgev2" "maunium.net/go/mautrix/bridgev2/database" "maunium.net/go/mautrix/bridgev2/networkid" @@ -109,7 +113,117 @@ func (t *TelegramClient) HandleMatrixViewingChat(ctx context.Context, msg *bridg GetChatInfoFunc: t.GetChatInfo, }) } - return t.maybePollForReactions(ctx, msg.Portal) + err := t.maybePollForReactions(ctx, msg.Portal) + if err != nil { + return err + } + err = t.pollSponsoredMessage(ctx, msg.Portal) + if err != nil { + return err + } + return nil +} + +func (t *TelegramClient) pollSponsoredMessage(ctx context.Context, portal *bridgev2.Portal) error { + if t.metadata.IsBot { + return nil + } + meta := portal.Metadata.(*PortalMetadata) + peerType, id, topicID, err := ids.ParsePortalID(portal.ID) + if err != nil { + return err + } else if peerType != ids.PeerTypeChannel || meta.IsSuperGroup || topicID != 0 { + return nil + } + meta.sponsoredMessageLock.Lock() + defer meta.sponsoredMessageLock.Unlock() + if time.Since(meta.SponsoredMessagePollTS.Time) < 5*time.Minute { + return nil + } + latestMessage, err := t.main.Bridge.DB.Message.GetLastNonFakePartAtOrBeforeTime(ctx, portal.PortalKey, time.Now()) + if err != nil { + return fmt.Errorf("failed to get latest message for portal: %w", err) + } else if latestMessage != nil && latestMessage.ID == meta.LastMessageOnSponsorFetch { + meta.SponsoredMessagePollTS = jsontime.UnixNow() + return nil + } + accessHash, err := t.ScopedStore.GetAccessHash(ctx, ids.PeerTypeChannel, id) + if err != nil { + return err + } + resp, err := t.client.API().MessagesGetSponsoredMessages(ctx, &tg.MessagesGetSponsoredMessagesRequest{ + Peer: &tg.InputPeerChannel{ChannelID: id, AccessHash: accessHash}, + }) + if err != nil { + return fmt.Errorf("failed to get sponsored messages: %w", err) + } + meta.SponsoredMessagePollTS = jsontime.UnixNow() + if latestMessage != nil { + meta.LastMessageOnSponsorFetch = latestMessage.ID + } + msgs, ok := resp.(*tg.MessagesSponsoredMessages) + if !ok || len(msgs.Messages) == 0 || (len(msgs.Messages) == 1 && bytes.Equal(msgs.Messages[0].RandomID, meta.SponsoredMessageRandomID)) { + err = portal.Save(ctx) + if err != nil { + return fmt.Errorf("failed to save portal after polling sponsored messages: %w", err) + } + return nil + } + if meta.sponsoredMessageSeen == nil { + meta.sponsoredMessageSeen = exsync.NewSet[int64]() + } else { + meta.sponsoredMessageSeen.Clear() + } + msg := msgs.Messages[0] + if bytes.Equal(msg.RandomID, meta.SponsoredMessageRandomID) && len(msgs.Messages) > 1 { + msg = msgs.Messages[1] + } + meta.SponsoredMessageRandomID = msg.RandomID + content := t.parseBodyAndHTML(ctx, msg.Message, msg.Entities) + content.MsgType = event.MsgNotice + content.EnsureHasHTML() + extra := map[string]any{ + "external_url": msg.URL, + "fi.mau.telegram.sponsored": map[string]any{ + "random_id": msg.RandomID, + "url": msg.URL, + "button_text": msg.ButtonText, + "title": msg.Title, + "content": content.FormattedBody, + "sponsor_info": msg.SponsorInfo, + "additional_info": msg.AdditionalInfo, + "recommended": msg.Recommended, + }, + } + var fromStr string + if msg.SponsorInfo != "" { + fromStr = fmt.Sprintf(" from %s", html.EscapeString(msg.SponsorInfo)) + } + prefix := "Ad" + if msg.Recommended { + prefix = "Recommended" + } + content.FormattedBody = fmt.Sprintf( + `%s: %s
%s

Sponsored message%s - %s

`, + prefix, html.EscapeString(msg.Title), content.FormattedBody, fromStr, msg.URL, msg.ButtonText, + ) + sendResp, err := t.main.Bridge.Bot.SendMessage(ctx, portal.MXID, event.EventMessage, &event.Content{ + Raw: extra, + Parsed: content, + }, &bridgev2.MatrixSendExtra{Timestamp: time.Now()}) + if err != nil { + return fmt.Errorf("failed to send sponsored message: %w", err) + } + meta.SponsoredMessageEventID = sendResp.EventID + zerolog.Ctx(ctx).Debug(). + Stringer("event_id", sendResp.EventID). + Str("random_id", base64.StdEncoding.EncodeToString(msg.RandomID)). + Msg("Sent sponsored message to Matrix") + err = portal.Save(ctx) + if err != nil { + return fmt.Errorf("failed to save portal after sending sponsored messages: %w", err) + } + return nil } func (t *TelegramClient) transferMediaToTelegram(ctx context.Context, content *event.MessageEventContent, sticker bool) (tg.InputMediaClass, error) { @@ -829,8 +943,21 @@ func (t *TelegramClient) HandleMatrixReadReceipt(ctx context.Context, msg *bridg MaxID: maxID, }) - if !msg.Portal.Metadata.(*PortalMetadata).IsSuperGroup { - // TODO handle sponsored message read receipts + meta := msg.Portal.Metadata.(*PortalMetadata) + randomID := meta.SponsoredMessageRandomID + if !t.metadata.IsBot && + randomID != nil && + time.Since(meta.SponsoredMessagePollTS.Time) < 15*time.Minute && + (meta.SponsoredMessageEventID == msg.EventID || msg.Receipt.Timestamp.After(meta.SponsoredMessagePollTS.Time)) && + meta.sponsoredMessageSeen.Add(t.telegramUserID) { + _, viewSponsoredErr := t.client.API().MessagesViewSponsoredMessage(ctx, randomID) + if viewSponsoredErr != nil { + log.Err(viewSponsoredErr).Msg("Failed to mark sponsored message as viewed after read receipt") + } else { + log.Debug(). + Str("random_id", base64.StdEncoding.EncodeToString(randomID)). + Msg("Marked sponsored message as viewed after read receipt") + } } default: readMessagesErr = fmt.Errorf("unknown peer type %s", peerType) @@ -843,6 +970,10 @@ func (t *TelegramClient) HandleMatrixReadReceipt(ctx context.Context, msg *bridg if err != nil { log.Err(err).Msg("failed to poll for reactions after read receipt") } + err = t.pollSponsoredMessage(ctx, msg.Portal) + if err != nil { + log.Err(err).Msg("failed to poll for sponsored message after read receipt") + } }() if peerType == ids.PeerTypeChannel && !msg.Portal.Metadata.(*PortalMetadata).FullSynced { diff --git a/pkg/connector/metadata.go b/pkg/connector/metadata.go index cb47c9cf..c7aeef7e 100644 --- a/pkg/connector/metadata.go +++ b/pkg/connector/metadata.go @@ -18,7 +18,9 @@ package connector import ( "context" + "sync" + "go.mau.fi/util/exsync" "go.mau.fi/util/jsontime" "maunium.net/go/mautrix/bridgev2" "maunium.net/go/mautrix/bridgev2/database" @@ -60,6 +62,14 @@ type PortalMetadata struct { LastSync jsontime.Unix `json:"last_sync,omitempty"` FullSynced bool `json:"full_synced,omitempty"` ParticipantsCount int `json:"member_count,omitempty"` + + SponsoredMessagePollTS jsontime.Unix `json:"sponsored_message_poll_ts,omitempty"` + SponsoredMessageEventID id.EventID `json:"sponsored_message_event_id,omitempty"` + SponsoredMessageRandomID []byte `json:"sponsored_message_random_id,omitempty"` + LastMessageOnSponsorFetch networkid.MessageID `json:"last_message_on_sponsor_fetch,omitempty"` + + sponsoredMessageLock sync.Mutex + sponsoredMessageSeen *exsync.Set[int64] } func (pm *PortalMetadata) SetIsSuperGroup(isSupergroup bool) (changed bool) {