diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 75a452a5..8870e034 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -106,7 +106,8 @@ type TelegramClient struct { stopTakeoutTimer *time.Timer takeoutDialogsOnce sync.Once - prevReactionPoll map[networkid.PortalKey]time.Time + prevReactionPoll map[networkid.PortalKey]time.Time + prevReactionPollLock sync.Mutex } var _ bridgev2.NetworkAPI = (*TelegramClient)(nil) diff --git a/pkg/connector/handlematrix.go b/pkg/connector/handlematrix.go index c73e9f06..8ab67e52 100644 --- a/pkg/connector/handlematrix.go +++ b/pkg/connector/handlematrix.go @@ -104,7 +104,7 @@ func (t *TelegramClient) HandleMatrixViewingChat(ctx context.Context, msg *bridg GetChatInfoFunc: t.GetChatInfo, }) } - return nil + return t.maybePollForReactions(ctx, msg.Portal) } func (t *TelegramClient) transferMediaToTelegram(ctx context.Context, content *event.MessageEventContent, sticker bool) (tg.InputMediaClass, error) { @@ -721,20 +721,7 @@ func (t *TelegramClient) HandleMatrixReadReceipt(ctx context.Context, msg *bridg wg.Add(1) go func() { defer wg.Done() - if peerType != ids.PeerTypeChannel || msg.Portal.Metadata.(*PortalMetadata).IsSuperGroup { - log.Debug().Msg("Not polling reactions because peer is not a channel or is a super-group") - return - } - - // If it hasn't been 20 seconds since the last poll, skip - now := time.Now() - if prev, ok := t.prevReactionPoll[msg.Portal.PortalKey]; ok && now.Before(prev.Add(20*time.Second)) { - log.Debug().Msg("Not polling reactions because last poll was less than 20 seconds ago") - return - } - t.prevReactionPoll[msg.Portal.PortalKey] = now - - reactionPollErr = t.pollForReactions(ctx, msg.Portal.PortalKey, inputPeer) + reactionPollErr = t.maybePollForReactions(ctx, msg.Portal) }() if peerType == ids.PeerTypeChannel && !msg.Portal.Metadata.(*PortalMetadata).FullSynced { diff --git a/pkg/connector/reactions.go b/pkg/connector/reactions.go index b1958739..c21e26c2 100644 --- a/pkg/connector/reactions.go +++ b/pkg/connector/reactions.go @@ -205,7 +205,30 @@ func (t *TelegramClient) getReactionLimit(ctx context.Context, sender networkid. } } -func (t *TelegramClient) pollForReactions(ctx context.Context, portalKey networkid.PortalKey, inputPeer tg.InputPeerClass) error { +func (t *TelegramClient) maybePollForReactions(ctx context.Context, portal *bridgev2.Portal) error { + // Only poll for reactions in supergroups + if portal == nil || !portal.Metadata.(*PortalMetadata).IsSuperGroup { + return nil + } + + t.prevReactionPollLock.Lock() + prev, ok := t.prevReactionPoll[portal.PortalKey] + if ok && time.Since(prev) > 20*time.Second { + ok = false + t.prevReactionPoll[portal.PortalKey] = time.Now() + } + t.prevReactionPollLock.Unlock() + if ok { + return nil + } + return t.pollForReactions(ctx, portal.PortalKey) +} + +func (t *TelegramClient) pollForReactions(ctx context.Context, portalKey networkid.PortalKey) error { + inputPeer, parseErr := t.inputPeerForPortalID(ctx, portalKey.ID) + if parseErr != nil { + return parseErr + } log := zerolog.Ctx(ctx).With(). Stringer("portal_key", portalKey). Str("action", "poll_for_reactions").