backfill: don't requeue history sync conversations that are still being created (#1035)

Co-authored-by: Indent <noreply@indent.com>

pkg/connector/backfill.go

@@ -96,16 +96,6 @@ 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)
@@ -415,6 +405,8 @@ 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").
@@ -427,9 +419,6 @@ 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)
@@ -461,6 +450,9 @@ 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 {
@@ -501,6 +493,7 @@ 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,
@@ -516,6 +509,7 @@ 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()
 				},
 			},
@@ -523,6 +517,7 @@ 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
 		}
@@ -530,9 +525,7 @@ 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()
-		wa.historySyncPortalLock.Lock()
+		if wa.historySyncQueued.Len() > 0 {
-		defer wa.historySyncPortalLock.Unlock()
-		if wa.historySyncPortalGen != gen {
 			return
 		}
 		wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false

pkg/connector/client.go

@@ -50,6 +50,7 @@ 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),
@@ -114,6 +115,7 @@ 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
@@ -128,9 +130,6 @@ type WhatsAppClient struct {
 	lastPresence       types.Presence
 	blocklistLock      sync.Mutex
 
-	historySyncPortalLock sync.Mutex
-	historySyncPortalGen  uint64
-
 	disableNewsletter bool
 
 	appStateRecoveryLock      sync.Mutex

pkg/connector/wadb/upgrades/00-latest-schema.sql

@@ -1,4 +1,4 @@
--- v0 -> v11 (compatible with v3+): Latest revision
+-- v0 -> v12 (compatible with v3+): Latest revision
 
 CREATE TABLE whatsapp_poll_option_id (
     bridge_id TEXT  NOT NULL,
@@ -50,6 +50,8 @@ 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,

pkg/connector/wadb/upgrades/12-history-sync-message-timestamp-index.sql

@@ -0,0 +1,2 @@
+-- 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);

pkg/msgconv/from-whatsapp.go

@@ -76,8 +76,7 @@ func (mc *MessageConverter) getBasicUserInfo(ctx context.Context, user types.JID
 	var pnJID types.JID
 	if user.Server == types.DefaultUserServer {
 		pnJID = user
-	} else if user.Server == types.HiddenUserServer {
+	} else if cli := getClient(ctx); user.Server == types.HiddenUserServer && cli != nil {
-		cli := getClient(ctx)
 		if user.User == cli.Store.GetLID().User {
 			pnJID = cli.Store.GetJID()
 		} else {