diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index e7748ac8..27b23a5b 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -96,6 +96,16 @@ func (wa *WhatsAppClient) historySyncLoop(ctx context.Context) { } if resetTimer { timerPending.Store(true) + wa.historySyncPortalLock.Lock() + wa.historySyncPortalGen++ + if meta := wa.UserLogin.Metadata.(*waid.UserLoginMetadata); !meta.HistorySyncPortalsNeedCreating { + meta.HistorySyncPortalsNeedCreating = true + err := wa.UserLogin.Save(ctx) + if err != nil { + wa.UserLogin.Log.Err(err).Msg("Failed to save user login history sync portals need creating flag") + } + } + wa.historySyncPortalLock.Unlock() } if timerPending.Load() { dispatchTimer.Reset(wa.Main.Config.HistorySync.DispatchWait) @@ -405,8 +415,6 @@ func (wa *WhatsAppClient) handleWAHistorySync( return nil } -const historySyncQueuedTimeout = 10 * time.Minute - func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { log := wa.UserLogin.Log.With(). Str("action", "create portals from history sync"). @@ -419,6 +427,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { } }() ctx = log.WithContext(ctx) + wa.historySyncPortalLock.Lock() + gen := wa.historySyncPortalGen + wa.historySyncPortalLock.Unlock() limit := wa.Main.Config.HistorySync.MaxInitialConversations loginTS := wa.UserLogin.Metadata.(*waid.UserLoginMetadata).LoggedInAt conversations, err := wa.Main.DB.Conversation.GetRecent(ctx, wa.UserLogin.ID, limit, loginTS) @@ -450,9 +461,6 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { // We don't currently support new PSAs, so don't bother backfilling them either wg.Done() continue - } else if queuedAt, ok := wa.historySyncQueued.Get(conv.ChatJID); ok && time.Since(queuedAt) < historySyncQueuedTimeout { - wg.Done() - continue } // TODO can the chat info fetch be avoided entirely? select { @@ -493,7 +501,6 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { wg.Done() continue } - wa.historySyncQueued.Set(conv.ChatJID, time.Now()) res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -509,7 +516,6 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { if err != nil { zerolog.Ctx(ctx).Err(err).Msg("Failed to mark conversation as bridged") } - wa.historySyncQueued.Delete(conv.ChatJID) wg.Done() }, }, @@ -517,7 +523,6 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { LatestMessageTS: conv.LastMessageTimestamp, }) if !res.Success { - wa.historySyncQueued.Delete(conv.ChatJID) log.Debug().Msg("Cancelling history sync portal creation loop") return } @@ -525,7 +530,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { log.Info().Int("conversation_count", len(conversations)).Msg("Finished creating portals from history sync") go func() { wg.Wait() - if wa.historySyncQueued.Len() > 0 { + wa.historySyncPortalLock.Lock() + defer wa.historySyncPortalLock.Unlock() + if wa.historySyncPortalGen != gen { return } wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 9b3f190a..cfb48708 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -50,7 +50,6 @@ func (wa *WhatsAppConnector) LoadUserLogin(ctx context.Context, login *bridgev2. MC: noopMCInstance, historySyncWakeup: make(chan struct{}, 1), - historySyncQueued: exsync.NewMap[types.JID, time.Time](), resyncQueue: make(map[types.JID]resyncQueueItem), directMediaRetries: make(map[networkid.MessageID]*directMediaRetry), mediaRetryLock: semaphore.NewWeighted(wa.Config.HistorySync.MediaRequests.MaxAsyncHandle), @@ -115,7 +114,6 @@ type WhatsAppClient struct { MC mClient historySyncWakeup chan struct{} - historySyncQueued *exsync.Map[types.JID, time.Time] stopLoops atomic.Pointer[context.CancelFunc] resyncQueue map[types.JID]resyncQueueItem resyncQueueLock sync.Mutex @@ -130,6 +128,9 @@ type WhatsAppClient struct { lastPresence types.Presence blocklistLock sync.Mutex + historySyncPortalLock sync.Mutex + historySyncPortalGen uint64 + disableNewsletter bool appStateRecoveryLock sync.Mutex diff --git a/pkg/connector/wadb/upgrades/00-latest-schema.sql b/pkg/connector/wadb/upgrades/00-latest-schema.sql index cd02333f..c4172d09 100644 --- a/pkg/connector/wadb/upgrades/00-latest-schema.sql +++ b/pkg/connector/wadb/upgrades/00-latest-schema.sql @@ -1,4 +1,4 @@ --- v0 -> v12 (compatible with v3+): Latest revision +-- v0 -> v11 (compatible with v3+): Latest revision CREATE TABLE whatsapp_poll_option_id ( bridge_id TEXT NOT NULL, @@ -50,8 +50,6 @@ CREATE TABLE whatsapp_history_sync_message ( REFERENCES whatsapp_history_sync_conversation (bridge_id, user_login_id, chat_jid) ON UPDATE CASCADE ON DELETE CASCADE ); -CREATE INDEX whatsapp_history_sync_message_chat_ts_idx ON whatsapp_history_sync_message (bridge_id, user_login_id, chat_jid, timestamp); - CREATE TABLE whatsapp_media_backfill_request ( bridge_id TEXT NOT NULL, user_login_id TEXT NOT NULL, diff --git a/pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql b/pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql deleted file mode 100644 index 0d3b910a..00000000 --- a/pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql +++ /dev/null @@ -1,2 +0,0 @@ --- v12 (compatible with v3+): Add index for history sync message timestamp lookups -CREATE INDEX whatsapp_history_sync_message_chat_ts_idx ON whatsapp_history_sync_message (bridge_id, user_login_id, chat_jid, timestamp); diff --git a/pkg/msgconv/from-whatsapp.go b/pkg/msgconv/from-whatsapp.go index e2ab92cd..ec273076 100644 --- a/pkg/msgconv/from-whatsapp.go +++ b/pkg/msgconv/from-whatsapp.go @@ -76,7 +76,8 @@ func (mc *MessageConverter) getBasicUserInfo(ctx context.Context, user types.JID var pnJID types.JID if user.Server == types.DefaultUserServer { pnJID = user - } else if cli := getClient(ctx); user.Server == types.HiddenUserServer && cli != nil { + } else if user.Server == types.HiddenUserServer { + cli := getClient(ctx) if user.User == cli.Store.GetLID().User { pnJID = cli.Store.GetJID() } else {