backfill: serialize history sync portal flag updates

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)
@@ -417,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)
@@ -517,6 +530,11 @@ 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()
+		defer wa.historySyncPortalLock.Unlock()
+		if wa.historySyncPortalGen != gen {
+			return
+		}
 		wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false
 		err = wa.UserLogin.Save(ctx)
 		if err != nil {

pkg/connector/client.go

@@ -128,6 +128,9 @@ 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 -> 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,

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

@@ -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);

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 {