lovely-projects/whatsapp
lovely-projects/whatsapp
1
0
Fork 0
mirror of https://github.com/mautrix/whatsapp.git synced 2025-03-14 14:15:38 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

{chat,user}info: add resync queue

Browse source
This commit is contained in:
Tulir Asokan 2024-09-26 15:23:15 +03:00
parent c4a466b7b9
commit d07fcc80e3
8 changed files with 239 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
cmd/mautrix-whatsapp/legacymigrate.sql
View file

@ -87,7 +87,13 @@ SELECT
END, -- room_type
CASE WHEN expiration_time>0 THEN 'after_read' END, -- disappear_type
CASE WHEN expiration_time > 0 THEN expiration_time * 1000000000 END, -- disappear_timer TODO check multiplier
'{}' -- metadata
-- only: postgres
jsonb_build_object
-- only: sqlite (line commented)
-- json_object
(
'last_sync', last_sync,
) -- metadata
FROM portal_old;
INSERT INTO user_portal (bridge_id, user_mxid, login_id, portal_id, portal_receiver, in_space, preferred, last_read)

8
pkg/connector/backfill.go
View file

@ -27,13 +27,7 @@ var _ bridgev2.BackfillingNetworkAPI = (*WhatsAppClient)(nil)
const historySyncDispatchWait = 30 * time.Second
func (wa *WhatsAppClient) historySyncLoop() {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
oldStop := wa.stopHistorySyncLoop.Swap(&cancel)
if oldStop != nil {
(*oldStop)()
}
func (wa *WhatsAppClient) historySyncLoop(ctx context.Context) {
dispatchTimer := time.NewTimer(historySyncDispatchWait)
dispatchTimer.Stop()
wa.UserLogin.Log.Debug().Msg("Starting history sync loop")

2
pkg/connector/capabilities.go
View file

@ -10,7 +10,7 @@ import (
var WhatsAppGeneralCaps = &bridgev2.NetworkGeneralCapabilities{
DisappearingMessages: true,
AggressiveUpdateInfo: false,
AggressiveUpdateInfo: true,
}
func (wa *WhatsAppConnector) GetCapabilities() *bridgev2.NetworkGeneralCapabilities {

37
pkg/connector/chatinfo.go
View file

@ -2,10 +2,12 @@ package connector
import (
"context"
"errors"
"fmt"
"time"
"github.com/rs/zerolog"
"go.mau.fi/util/jsontime"
"go.mau.fi/util/ptr"
"go.mau.fi/whatsmeow"
"go.mau.fi/whatsmeow/types"
@ -43,6 +45,7 @@ func (wa *WhatsAppClient) getChatInfo(ctx context.Context, portalJID types.JID,
return nil, err
}
wrapped = wa.wrapGroupInfo(info)
wrapped.ExtraUpdates = bridgev2.MergeExtraUpdaters(wrapped.ExtraUpdates, updatePortalLastSyncAt)
case types.NewsletterServer:
info, err := wa.Client.GetNewsletterInfo(portalJID)
if err != nil {
@ -65,6 +68,13 @@ func (wa *WhatsAppClient) getChatInfo(ctx context.Context, portalJID types.JID,
return wrapped, nil
}
func updatePortalLastSyncAt(_ context.Context, portal *bridgev2.Portal) bool {
meta := portal.Metadata.(*waid.PortalMetadata)
forceSave := time.Since(meta.LastSync.Time) > 24*time.Hour
meta.LastSync = jsontime.UnixNow()
return forceSave
}
func updateDisappearingTimerSetAt(ts int64) bridgev2.ExtraUpdater[*bridgev2.Portal] {
return func(_ context.Context, portal *bridgev2.Portal) bool {
meta := portal.Metadata.(*waid.PortalMetadata)
@ -331,15 +341,33 @@ func (wa *WhatsAppClient) makePortalAvatarFetcher(avatarID string, sender types.
if existingID == "remove" || existingID == "unauthorized" {
existingID = ""
}
var wrappedAvatar *bridgev2.Avatar
avatar, err := wa.Client.GetProfilePictureInfo(jid, &whatsmeow.GetProfilePictureParams{
ExistingID: existingID,
IsCommunity: portal.RoomType == database.RoomTypeSpace,
})
if err != nil {
if errors.Is(err, whatsmeow.ErrProfilePictureNotSet) {
wrappedAvatar = &bridgev2.Avatar{
ID: "remove",
Remove: true,
}
} else if errors.Is(err, whatsmeow.ErrProfilePictureUnauthorized) {
wrappedAvatar = &bridgev2.Avatar{
ID: "unauthorized",
Remove: true,
}
} else if err != nil {
zerolog.Ctx(ctx).Err(err).Msg("Failed to get avatar info")
return false
} else if avatar == nil {
return false
} else {
wrappedAvatar = &bridgev2.Avatar{
ID: networkid.AvatarID(avatar.ID),
Get: func(ctx context.Context) ([]byte, error) {
return wa.Client.DownloadMediaWithPath(avatar.DirectPath, nil, nil, nil, 0, "", "")
},
}
}
var evtSender bridgev2.EventSender
if !sender.IsEmpty() {
@ -347,12 +375,7 @@ func (wa *WhatsAppClient) makePortalAvatarFetcher(avatarID string, sender types.
}
senderIntent := portal.GetIntentFor(ctx, evtSender, wa.UserLogin, bridgev2.RemoteEventChatInfoChange)
//lint:ignore SA1019 TODO invent a cleaner way to fetch avatar metadata before updating?
return portal.Internal().UpdateAvatar(ctx, &bridgev2.Avatar{
ID: networkid.AvatarID(avatar.ID),
Get: func(ctx context.Context) ([]byte, error) {
return wa.Client.DownloadMediaWithPath(avatar.DirectPath, nil, nil, nil, 0, "", "")
},
}, senderIntent, ts)
return portal.Internal().UpdateAvatar(ctx, wrappedAvatar, senderIntent, ts)
}
}

29
pkg/connector/client.go
View file

@ -3,7 +3,9 @@ package connector
import (
"context"
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/rs/zerolog"
"go.mau.fi/whatsmeow"
@ -24,6 +26,7 @@ func (wa *WhatsAppConnector) LoadUserLogin(_ context.Context, login *bridgev2.Us
UserLogin: login,
historySyncs: make(chan *waHistorySync.HistorySync, 64),
resyncQueue: make(map[types.JID]resyncQueueItem),
}
login.Client = w
@ -52,6 +55,11 @@ func (wa *WhatsAppConnector) LoadUserLogin(_ context.Context, login *bridgev2.Us
return nil
}
type resyncQueueItem struct {
portal *bridgev2.Portal
ghost *bridgev2.Ghost
}
type WhatsAppClient struct {
Main *WhatsAppConnector
UserLogin *bridgev2.UserLogin
@ -59,8 +67,11 @@ type WhatsAppClient struct {
Device *store.Device
JID types.JID
historySyncs chan *waHistorySync.HistorySync
stopHistorySyncLoop atomic.Pointer[context.CancelFunc]
historySyncs chan *waHistorySync.HistorySync
stopLoops atomic.Pointer[context.CancelFunc]
resyncQueue map[types.JID]resyncQueueItem
resyncQueueLock sync.Mutex
nextResync time.Time
}
var _ bridgev2.NetworkAPI = (*WhatsAppClient)(nil)
@ -103,12 +114,22 @@ func (wa *WhatsAppClient) Connect(ctx context.Context) error {
if err := wa.Main.updateProxy(ctx, wa.Client, false); err != nil {
zerolog.Ctx(ctx).Err(err).Msg("Failed to update proxy")
}
go wa.historySyncLoop()
wa.startLoops()
return wa.Client.Connect()
}
func (wa *WhatsAppClient) startLoops() {
ctx, cancel := context.WithCancel(context.Background())
oldStop := wa.stopLoops.Swap(&cancel)
if oldStop != nil {
(*oldStop)()
}
go wa.historySyncLoop(ctx)
go wa.ghostResyncLoop(ctx)
}
func (wa *WhatsAppClient) Disconnect() {
if stopHistorySyncLoop := wa.stopHistorySyncLoop.Swap(nil); stopHistorySyncLoop != nil {
if stopHistorySyncLoop := wa.stopLoops.Swap(nil); stopHistorySyncLoop != nil {
(*stopHistorySyncLoop)()
}
if cli := wa.Client; cli != nil {

1
pkg/connector/events.go
View file

@ -182,6 +182,7 @@ func (evt *WAMessageEvent) HandleExisting(ctx context.Context, portal *bridgev2.
}
func (evt *WAMessageEvent) ConvertMessage(ctx context.Context, portal *bridgev2.Portal, intent bridgev2.MatrixAPI) (*bridgev2.ConvertedMessage, error) {
evt.wa.EnqueuePortalResync(portal)
converted := evt.wa.Main.MsgConv.ToMatrix(ctx, portal, evt.wa.Client, intent, evt.Message, &evt.Info)
return converted, nil
}

179
pkg/connector/userinfo.go
View file

@ -2,19 +2,164 @@ package connector
import (
"context"
"errors"
"fmt"
"math/rand/v2"
"time"
"github.com/rs/zerolog"
"go.mau.fi/util/exzerolog"
"go.mau.fi/util/jsontime"
"go.mau.fi/util/ptr"
"go.mau.fi/whatsmeow"
"go.mau.fi/whatsmeow/types"
"maunium.net/go/mautrix/bridgev2"
"maunium.net/go/mautrix/bridgev2/networkid"
"maunium.net/go/mautrix/bridgev2/simplevent"
"maunium.net/go/mautrix-whatsapp/pkg/waid"
)
const resyncMinInterval = 7 * 24 * time.Hour
const resyncLoopInterval = 4 * time.Hour
func (wa *WhatsAppClient) EnqueueGhostResync(ghost *bridgev2.Ghost) {
if ghost.Metadata.(*waid.GhostMetadata).LastSync.Add(resyncMinInterval).After(time.Now()) {
return
}
wa.resyncQueueLock.Lock()
jid := waid.ParseUserID(ghost.ID)
if _, exists := wa.resyncQueue[jid]; !exists {
wa.resyncQueue[jid] = resyncQueueItem{ghost: ghost}
wa.UserLogin.Log.Debug().
Stringer("jid", jid).
Stringer("next_resync_in", time.Until(wa.nextResync)).
Msg("Enqueued resync for ghost")
}
wa.resyncQueueLock.Unlock()
}
func (wa *WhatsAppClient) EnqueuePortalResync(portal *bridgev2.Portal) {
jid, _ := waid.ParsePortalID(portal.ID)
if jid.Server != types.GroupServer || portal.Metadata.(*waid.PortalMetadata).LastSync.Add(resyncMinInterval).After(time.Now()) {
return
}
wa.resyncQueueLock.Lock()
if _, exists := wa.resyncQueue[jid]; !exists {
wa.resyncQueue[jid] = resyncQueueItem{portal: portal}
wa.UserLogin.Log.Debug().
Stringer("jid", jid).
Stringer("next_resync_in", time.Until(wa.nextResync)).
Msg("Enqueued resync for portal")
}
wa.resyncQueueLock.Unlock()
}
func (wa *WhatsAppClient) ghostResyncLoop(ctx context.Context) {
log := wa.UserLogin.Log.With().Str("action", "ghost resync loop").Logger()
ctx = log.WithContext(ctx)
wa.nextResync = time.Now().Add(resyncLoopInterval).Add(-time.Duration(rand.IntN(3600)) * time.Second)
timer := time.NewTimer(time.Until(wa.nextResync))
log.Info().Time("first_resync", wa.nextResync).Msg("Ghost resync queue starting")
for {
select {
case <-ctx.Done():
timer.Stop()
return
case <-timer.C:
}
queue := wa.rotateResyncQueue()
timer.Reset(time.Until(wa.nextResync))
if len(queue) > 0 {
wa.doGhostResync(ctx, queue)
} else {
log.Trace().Msg("Nothing in background resync queue")
}
}
}
func (wa *WhatsAppClient) rotateResyncQueue() map[types.JID]resyncQueueItem {
wa.resyncQueueLock.Lock()
defer wa.resyncQueueLock.Unlock()
wa.nextResync = time.Now().Add(resyncLoopInterval)
if len(wa.resyncQueue) == 0 {
return nil
}
queue := wa.resyncQueue
wa.resyncQueue = make(map[types.JID]resyncQueueItem)
return queue
}
func (wa *WhatsAppClient) doGhostResync(ctx context.Context, queue map[types.JID]resyncQueueItem) {
log := zerolog.Ctx(ctx)
log.Debug().Msg("Starting background resyncs")
defer log.Debug().Msg("Background resyncs finished")
var ghostJIDs []types.JID
var ghosts []*bridgev2.Ghost
var portals []*bridgev2.Portal
for jid, item := range queue {
var lastSync time.Time
if item.ghost != nil {
lastSync = item.ghost.Metadata.(*waid.GhostMetadata).LastSync.Time
} else if item.portal != nil {
lastSync = item.portal.Metadata.(*waid.PortalMetadata).LastSync.Time
}
if lastSync.Add(resyncMinInterval).After(time.Now()) {
log.Debug().
Stringer("jid", jid).
Time("last_sync", lastSync).
Msg("Not resyncing, last sync was too recent")
continue
}
if item.ghost != nil {
ghosts = append(ghosts, item.ghost)
ghostJIDs = append(ghostJIDs, jid)
} else if item.portal != nil {
portals = append(portals, item.portal)
}
}
for _, portal := range portals {
wa.Main.Bridge.QueueRemoteEvent(wa.UserLogin, &simplevent.ChatResync{
EventMeta: simplevent.EventMeta{
Type: bridgev2.RemoteEventChatResync,
LogContext: func(c zerolog.Context) zerolog.Context {
return c.Str("sync_reason", "queue")
},
PortalKey: portal.PortalKey,
},
GetChatInfoFunc: wa.GetChatInfo,
})
}
if len(ghostJIDs) == 0 {
return
}
log.Debug().Array("jids", exzerolog.ArrayOfStringers(ghostJIDs)).Msg("Doing background sync for users")
infos, err := wa.Client.GetUserInfo(ghostJIDs)
if err != nil {
log.Err(err).Msg("Failed to get user info for background sync")
return
}
for _, ghost := range ghosts {
jid := waid.ParseUserID(ghost.ID)
info, ok := infos[jid]
if !ok {
log.Warn().Stringer("jid", jid).Msg("Didn't get info for puppet in background sync")
continue
}
userInfo, err := wa.getUserInfo(ctx, jid, info.PictureID != "" && string(ghost.AvatarID) != info.PictureID)
if err != nil {
log.Err(err).Stringer("jid", jid).Msg("Failed to get user info for puppet in background sync")
continue
}
ghost.UpdateInfo(ctx, userInfo)
}
}
func (wa *WhatsAppClient) GetUserInfo(ctx context.Context, ghost *bridgev2.Ghost) (*bridgev2.UserInfo, error) {
if ghost.Name != "" {
wa.EnqueueGhostResync(ghost)
return nil, nil
}
jid := waid.ParseUserID(ghost.ID)
fetchAvatar := !ghost.Metadata.(*waid.GhostMetadata).AvatarFetchAttempted
return wa.getUserInfo(ctx, jid, fetchAvatar)
@ -37,8 +182,15 @@ func (wa *WhatsAppClient) contactToUserInfo(jid types.JID, contact types.Contact
if getAvatar {
ui.ExtraUpdates = bridgev2.MergeExtraUpdaters(ui.ExtraUpdates, wa.fetchGhostAvatar)
}
ui.ExtraUpdates = bridgev2.MergeExtraUpdaters(ui.ExtraUpdates, updateGhostLastSyncAt)
return ui
}
func updateGhostLastSyncAt(_ context.Context, ghost *bridgev2.Ghost) bool {
meta := ghost.Metadata.(*waid.GhostMetadata)
forceSave := time.Since(meta.LastSync.Time) > 24*time.Hour
meta.LastSync = jsontime.UnixNow()
return forceSave
}
func (wa *WhatsAppClient) fetchGhostAvatar(ctx context.Context, ghost *bridgev2.Ghost) bool {
jid := waid.ParseUserID(ghost.ID)
@ -49,17 +201,30 @@ func (wa *WhatsAppClient) fetchGhostAvatar(ctx context.Context, ghost *bridgev2.
}
wasAttempted := meta.AvatarFetchAttempted
meta.AvatarFetchAttempted = true
var wrappedAvatar *bridgev2.Avatar
avatar, err := wa.Client.GetProfilePictureInfo(jid, &whatsmeow.GetProfilePictureParams{ExistingID: existingID})
if err != nil {
if errors.Is(err, whatsmeow.ErrProfilePictureNotSet) {
wrappedAvatar = &bridgev2.Avatar{
ID: "remove",
Remove: true,
}
} else if errors.Is(err, whatsmeow.ErrProfilePictureUnauthorized) {
wrappedAvatar = &bridgev2.Avatar{
ID: "unauthorized",
Remove: true,
}
} else if err != nil {
zerolog.Ctx(ctx).Err(err).Msg("Failed to get avatar info")
return !wasAttempted
} else if avatar == nil {
return !wasAttempted
} else {
wrappedAvatar = &bridgev2.Avatar{
ID: networkid.AvatarID(avatar.ID),
Get: func(ctx context.Context) ([]byte, error) {
return wa.Client.DownloadMediaWithPath(avatar.DirectPath, nil, nil, nil, 0, "", "")
},
}
}
return ghost.UpdateAvatar(ctx, &bridgev2.Avatar{
ID: networkid.AvatarID(avatar.ID),
Get: func(ctx context.Context) ([]byte, error) {
return wa.Client.DownloadMediaWithPath(avatar.DirectPath, nil, nil, nil, 0, "", "")
},
}) || !wasAttempted
return ghost.UpdateAvatar(ctx, wrappedAvatar) || !wasAttempted
}

3
pkg/waid/dbmeta.go
View file

@ -55,7 +55,8 @@ type ReactionMetadata struct {
}
type PortalMetadata struct {
DisappearingTimerSetAt int64 `json:"disappearing_timer_set_at,omitempty"`
DisappearingTimerSetAt int64 `json:"disappearing_timer_set_at,omitempty"`
LastSync jsontime.Unix `json:"last_sync,omitempty"`
}
type GhostMetadata struct {