backfill: restart portal creation if interrupted soon after login

go.mod

@@ -18,7 +18,7 @@ require (
 	golang.org/x/sync v0.8.0
 	google.golang.org/protobuf v1.35.1
 	gopkg.in/yaml.v3 v3.0.1
-	maunium.net/go/mautrix v0.21.2-0.20241106121131-f588c35d8b1c
+	maunium.net/go/mautrix v0.21.2-0.20241106135041-5967fe7b0f82
 )
 
 require (

go.sum

@@ -101,5 +101,5 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 maunium.net/go/mauflag v1.0.0 h1:YiaRc0tEI3toYtJMRIfjP+jklH45uDHtT80nUamyD4M=
 maunium.net/go/mauflag v1.0.0/go.mod h1:nLivPOpTpHnpzEh8jEdSL9UqO9+/KBJFmNRlwKfkPeA=
-maunium.net/go/mautrix v0.21.2-0.20241106121131-f588c35d8b1c h1:X8EZp6mkboTX3ACDpYQgFW2Vb98DF2pgcnqrAfqSPrI=
-maunium.net/go/mautrix v0.21.2-0.20241106121131-f588c35d8b1c/go.mod h1:UBuBMbPJfh1AqYc1K1Lr0eQclx5vs1k1iiLVO/iMyw4=
+maunium.net/go/mautrix v0.21.2-0.20241106135041-5967fe7b0f82 h1:Dc70Xr6WFJj5XYtJhVGfcIU4hfcgIy2DGc4wWnzGSbA=
+maunium.net/go/mautrix v0.21.2-0.20241106135041-5967fe7b0f82/go.mod h1:UBuBMbPJfh1AqYc1K1Lr0eQclx5vs1k1iiLVO/iMyw4=

pkg/connector/backfill.go

@@ -7,6 +7,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/rs/zerolog"
@@ -31,7 +32,12 @@ const historySyncDispatchWait = 30 * time.Second
 
 func (wa *WhatsAppClient) historySyncLoop(ctx context.Context) {
 	dispatchTimer := time.NewTimer(historySyncDispatchWait)
-	dispatchTimer.Stop()
+
+	if !wa.isNewLogin && wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating {
+		dispatchTimer.Reset(5 * time.Second)
+	} else {
+		dispatchTimer.Stop()
+	}
 	wa.UserLogin.Log.Debug().Msg("Starting history sync loop")
 	for {
 		select {
@@ -196,9 +202,12 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 	}
 	log.Info().Int("conversation_count", len(conversations)).Msg("Creating portals from history sync")
 	rateLimitErrors := 0
+	var wg sync.WaitGroup
+	wg.Add(len(conversations))
 	for i := 0; i < len(conversations); i++ {
 		conv := conversations[i]
 		if conv.ChatJID == types.StatusBroadcastJID && !wa.Main.Config.EnableStatusBroadcast {
+			wg.Done()
 			continue
 		}
 		// TODO can the chat info fetch be avoided entirely?
@@ -207,10 +216,15 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 		if errors.Is(err, whatsmeow.ErrNotInGroup) {
 			log.Debug().Stringer("chat_jid", conv.ChatJID).
 				Msg("Skipping creating room because the user is not a participant")
-			err = wa.Main.DB.Message.DeleteAllInChat(ctx, wa.UserLogin.ID, conv.ChatJID)
+			//err = wa.Main.DB.Message.DeleteAllInChat(ctx, wa.UserLogin.ID, conv.ChatJID)
+			//if err != nil {
+			//	log.Err(err).Msg("Failed to delete historical messages for portal")
+			//}
+			err = wa.Main.DB.Conversation.Delete(ctx, wa.UserLogin.ID, conv.ChatJID)
 			if err != nil {
-				log.Err(err).Msg("Failed to delete historical messages for portal")
+				log.Err(err).Msg("Failed to delete conversation user is not in")
 			}
+			wg.Done()
 			continue
 		} else if errors.Is(err, whatsmeow.ErrIQRateOverLimit) {
 			rateLimitErrors++
@@ -222,6 +236,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 			continue
 		} else if err != nil {
 			log.Err(err).Stringer("chat_jid", conv.ChatJID).Msg("Failed to get chat info")
+			wg.Done()
 			continue
 		}
 		wa.Main.Bridge.QueueRemoteEvent(wa.UserLogin, &simplevent.ChatResync{
@@ -230,12 +245,28 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 				LogContext:   nil,
 				PortalKey:    wa.makeWAPortalKey(conv.ChatJID),
 				CreatePortal: true,
+				PostHandleFunc: func(ctx context.Context, portal *bridgev2.Portal) {
+					err := wa.Main.DB.Conversation.MarkBridged(ctx, wa.UserLogin.ID, conv.ChatJID)
+					if err != nil {
+						zerolog.Ctx(ctx).Err(err).Msg("Failed to mark conversation as bridged")
+					}
+					wg.Done()
+				},
 			},
 			ChatInfo:        wrappedInfo,
 			LatestMessageTS: conv.LastMessageTimestamp,
 		})
 	}
 	log.Info().Int("conversation_count", len(conversations)).Msg("Finished creating portals from history sync")
+	go func() {
+		wg.Wait()
+		wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false
+		err = wa.UserLogin.Save(ctx)
+		if err != nil {
+			log.Err(err).Msg("Failed to save user login history sync portals created flag")
+		}
+		log.Info().Msg("Finished processing all history sync chat resync events")
+	}()
 }
 
 func (wa *WhatsAppClient) FetchMessages(ctx context.Context, params bridgev2.FetchMessagesParams) (*bridgev2.FetchMessagesResponse, error) {

pkg/connector/client.go

@@ -98,6 +98,7 @@ type WhatsAppClient struct {
 	mediaRetryLock     *semaphore.Weighted
 
 	lastPhoneOfflineWarning time.Time
+	isNewLogin              bool
 }
 
 var (

pkg/connector/login.go

@@ -308,6 +308,8 @@ func (wl *WALogin) Wait(ctx context.Context) (*bridgev2.LoginStep, error) {
 		Metadata: &waid.UserLoginMetadata{
 			WADeviceID: wl.LoginSuccess.ID.Device,
 			Timezone:   wl.Timezone,
+
+			HistorySyncPortalsNeedCreating: true,
 		},
 	}, &bridgev2.NewLoginParams{
 		DeleteOnConflict: true,
@@ -316,6 +318,7 @@ func (wl *WALogin) Wait(ctx context.Context) (*bridgev2.LoginStep, error) {
 		return nil, fmt.Errorf("failed to create user login: %w", err)
 	}
 
+	ul.Client.(*WhatsAppClient).isNewLogin = true
 	err = ul.Client.Connect(ul.Log.WithContext(context.Background()))
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect after login: %w", err)

pkg/connector/wadb/conversation.go

@@ -30,6 +30,7 @@ type Conversation struct {
 	EphemeralSettingTimestamp *int64
 	MarkedAsUnread            *bool
 	UnreadCount               *uint32
+	Bridged                   bool
 }
 
 func parseHistoryTime(ts *uint64) time.Time {
@@ -63,9 +64,10 @@ const (
 	upsertHistorySyncConversationQuery = `
 		INSERT INTO whatsapp_history_sync_conversation (
 			bridge_id, user_login_id, chat_jid, last_message_timestamp, archived, pinned, mute_end_time,
-			end_of_history_transfer_type, ephemeral_expiration, ephemeral_setting_timestamp, marked_as_unread, unread_count
+			end_of_history_transfer_type, ephemeral_expiration, ephemeral_setting_timestamp, marked_as_unread,
+			unread_count, bridged
 		)
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)
 		ON CONFLICT (bridge_id, user_login_id, chat_jid)
 		DO UPDATE SET
 			last_message_timestamp=CASE
@@ -81,21 +83,24 @@ const (
 			ephemeral_expiration=COALESCE(excluded.ephemeral_expiration, whatsapp_history_sync_conversation.ephemeral_expiration),
 			ephemeral_setting_timestamp=COALESCE(excluded.ephemeral_setting_timestamp, whatsapp_history_sync_conversation.ephemeral_setting_timestamp),
 			marked_as_unread=COALESCE(excluded.marked_as_unread, whatsapp_history_sync_conversation.marked_as_unread),
-			unread_count=COALESCE(excluded.unread_count, whatsapp_history_sync_conversation.unread_count)
+			unread_count=COALESCE(excluded.unread_count, whatsapp_history_sync_conversation.unread_count),
+			bridged=false
 	`
 	getRecentConversations = `
 		SELECT
 			bridge_id, user_login_id, chat_jid, last_message_timestamp, archived, pinned, mute_end_time,
-			end_of_history_transfer_type, ephemeral_expiration, ephemeral_setting_timestamp, marked_as_unread, unread_count
+			end_of_history_transfer_type, ephemeral_expiration, ephemeral_setting_timestamp, marked_as_unread,
+			unread_count, bridged
 		FROM whatsapp_history_sync_conversation
-		WHERE bridge_id=$1 AND user_login_id=$2
+		WHERE bridge_id=$1 AND user_login_id=$2 AND bridged=false
 		ORDER BY last_message_timestamp DESC
 		LIMIT $3
 	`
 	getConversationByJID = `
 		SELECT
 			bridge_id, user_login_id, chat_jid, last_message_timestamp, archived, pinned, mute_end_time,
-			end_of_history_transfer_type, ephemeral_expiration, ephemeral_setting_timestamp, marked_as_unread, unread_count
+			end_of_history_transfer_type, ephemeral_expiration, ephemeral_setting_timestamp, marked_as_unread,
+			unread_count, bridged
 		FROM whatsapp_history_sync_conversation
 		WHERE bridge_id=$1 AND user_login_id=$2 AND chat_jid=$3
 	`
@@ -104,6 +109,11 @@ const (
 		DELETE FROM whatsapp_history_sync_conversation
 		WHERE bridge_id=$1 AND user_login_id=$2 AND chat_jid=$3
 	`
+	markConversationBridged = `
+		UPDATE whatsapp_history_sync_conversation
+		SET bridged=true
+		WHERE bridge_id=$1 AND user_login_id=$2 AND chat_jid=$3
+	`
 )
 
 func (cq *ConversationQuery) Put(ctx context.Context, conv *Conversation) error {
@@ -120,6 +130,10 @@ func (cq *ConversationQuery) GetRecent(ctx context.Context, loginID networkid.Us
 	return cq.QueryMany(ctx, getRecentConversations, cq.BridgeID, loginID, limitPtr)
 }
 
+func (cq *ConversationQuery) MarkBridged(ctx context.Context, loginID networkid.UserLoginID, chatJID types.JID) error {
+	return cq.Exec(ctx, markConversationBridged, cq.BridgeID, loginID, chatJID)
+}
+
 func (cq *ConversationQuery) Get(ctx context.Context, loginID networkid.UserLoginID, chatJID types.JID) (*Conversation, error) {
 	return cq.QueryOne(ctx, getConversationByJID, cq.BridgeID, loginID, chatJID)
 }
@@ -153,6 +167,7 @@ func (c *Conversation) sqlVariables() []any {
 		c.EphemeralSettingTimestamp,
 		c.MarkedAsUnread,
 		c.UnreadCount,
+		c.Bridged,
 	}
 }
 
@@ -171,6 +186,7 @@ func (c *Conversation) Scan(row dbutil.Scannable) (*Conversation, error) {
 		&c.EphemeralSettingTimestamp,
 		&c.MarkedAsUnread,
 		&c.UnreadCount,
+		&c.Bridged,
 	)
 	if err != nil {
 		return nil, err

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

@@ -1,4 +1,4 @@
--- v0 -> v3 (compatible with v3+): Latest revision
+-- v0 -> v4 (compatible with v3+): Latest revision
 
 CREATE TABLE whatsapp_poll_option_id (
     bridge_id TEXT  NOT NULL,
@@ -13,9 +13,9 @@ CREATE TABLE whatsapp_poll_option_id (
 );
 
 CREATE TABLE whatsapp_history_sync_conversation (
-    bridge_id                    TEXT NOT NULL,
-    user_login_id                TEXT NOT NULL,
-    chat_jid                     TEXT NOT NULL,
+    bridge_id                    TEXT    NOT NULL,
+    user_login_id                TEXT    NOT NULL,
+    chat_jid                     TEXT    NOT NULL,
 
     last_message_timestamp       BIGINT,
     archived                     BOOLEAN,
@@ -27,6 +27,8 @@ CREATE TABLE whatsapp_history_sync_conversation (
     marked_as_unread             BOOLEAN,
     unread_count                 INTEGER,
 
+    bridged                      BOOLEAN NOT NULL DEFAULT false,
+
     PRIMARY KEY (bridge_id, user_login_id, chat_jid),
     CONSTRAINT whatsapp_history_sync_conversation_user_login_fkey FOREIGN KEY (bridge_id, user_login_id)
         REFERENCES user_login (bridge_id, id) ON UPDATE CASCADE ON DELETE CASCADE

pkg/connector/wadb/upgrades/04-conversation-bridged-flag.sql

@@ -0,0 +1,2 @@
+-- v4 (compatible with v3+): Add bridged flag for history sync conversations
+ALTER TABLE history_sync_conversation ADD COLUMN bridged BOOLEAN NOT NULL DEFAULT false;

pkg/waid/dbmeta.go

@@ -33,6 +33,8 @@ type UserLoginMetadata struct {
 	PhoneLastPinged jsontime.Unix `json:"phone_last_pinged"`
 	Timezone        string        `json:"timezone"`
 	PushKeys        *PushKeys     `json:"push_keys,omitempty"`
+
+	HistorySyncPortalsNeedCreating bool `json:"history_sync_portals_need_creating,omitempty"`
 }
 
 type PushKeys struct {