bridgev2: automatically update old portals when enabling split portals

bridgev2/bridge.go

@@ -120,6 +120,7 @@ func (br *Bridge) StartConnectors() error {
 	if err != nil {
 		return DBUpgradeError{Err: err, Section: "main"}
 	}
+	didSplitPortals := br.MigrateToSplitPortals(ctx)
 	br.Log.Info().Msg("Starting Matrix connector")
 	err = br.Matrix.Start(ctx)
 	if err != nil {
@@ -133,9 +134,41 @@ func (br *Bridge) StartConnectors() error {
 	if br.Network.GetCapabilities().DisappearingMessages {
 		go br.DisappearLoop.Start()
 	}
+	if didSplitPortals || br.Config.ResendBridgeInfo {
+		br.ResendBridgeInfo(ctx)
+	}
 	return nil
 }
 
+func (br *Bridge) ResendBridgeInfo(ctx context.Context) {
+	log := zerolog.Ctx(ctx).With().Str("action", "resend bridge info").Logger()
+	portals, err := br.GetAllPortalsWithMXID(ctx)
+	if err != nil {
+		log.Err(err).Msg("Failed to get portals")
+		return
+	}
+	for _, portal := range portals {
+		portal.UpdateBridgeInfo(ctx)
+	}
+	log.Info().Msg("Resent bridge info to all portals")
+}
+
+func (br *Bridge) MigrateToSplitPortals(ctx context.Context) bool {
+	log := zerolog.Ctx(ctx).With().Str("action", "migrate to split portals").Logger()
+	ctx = log.WithContext(ctx)
+	if !br.Config.SplitPortals || br.DB.KV.Get(ctx, database.KeySplitPortalsEnabled) == "true" {
+		return false
+	}
+	affected, err := br.DB.Portal.MigrateToSplitPortals(ctx)
+	if err != nil {
+		log.Err(err).Msg("Failed to migrate portals")
+		return false
+	}
+	log.Info().Int64("rows_affected", affected).Msg("Migrated to split portals")
+	br.DB.KV.Set(ctx, database.KeySplitPortalsEnabled, "true")
+	return affected > 0
+}
+
 func (br *Bridge) StartLogins() error {
 	ctx := br.Log.WithContext(context.Background())
 

bridgev2/bridgeconfig/config.go

@@ -62,6 +62,7 @@ type BridgeConfig struct {
 	PrivateChatPortalMeta   bool             `yaml:"private_chat_portal_meta"`
 	AsyncEvents             bool             `yaml:"async_events"`
 	SplitPortals            bool             `yaml:"split_portals"`
+	ResendBridgeInfo        bool             `yaml:"resend_bridge_info"`
 	BridgeMatrixLeave       bool             `yaml:"bridge_matrix_leave"`
 	TagOnlyOnCreate         bool             `yaml:"tag_only_on_create"`
 	MuteOnlyOnCreate        bool             `yaml:"mute_only_on_create"`

bridgev2/bridgeconfig/upgrade.go

@@ -27,6 +27,7 @@ func doUpgrade(helper up.Helper) {
 	helper.Copy(up.Bool, "bridge", "private_chat_portal_meta")
 	helper.Copy(up.Bool, "bridge", "async_events")
 	helper.Copy(up.Bool, "bridge", "split_portals")
+	helper.Copy(up.Bool, "bridge", "resend_bridge_info")
 	helper.Copy(up.Bool, "bridge", "bridge_matrix_leave")
 	helper.Copy(up.Bool, "bridge", "tag_only_on_create")
 	helper.Copy(up.Bool, "bridge", "mute_only_on_create")
@@ -168,6 +169,8 @@ func doUpgrade(helper up.Helper) {
 
 var SpacedBlocks = [][]string{
 	{"bridge"},
+	{"bridge", "bridge_matrix_leave"},
+	{"bridge", "cleanup_on_logout"},
 	{"bridge", "relay"},
 	{"bridge", "permissions"},
 	{"database"},

bridgev2/database/database.go

@@ -33,6 +33,7 @@ type Database struct {
 	UserLogin           *UserLoginQuery
 	UserPortal          *UserPortalQuery
 	BackfillTask        *BackfillTaskQuery
+	KV                  *KVQuery
 }
 
 type MetaMerger interface {
@@ -136,6 +137,10 @@ func New(bridgeID networkid.BridgeID, mt MetaTypes, db *dbutil.Database) *Databa
 				return &BackfillTask{}
 			}),
 		},
+		KV: &KVQuery{
+			BridgeID: bridgeID,
+			Database: db,
+		},
 	}
 }
 

bridgev2/database/kvstore.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2024 Tulir Asokan
+//
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+package database
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+
+	"github.com/rs/zerolog"
+	"go.mau.fi/util/dbutil"
+
+	"maunium.net/go/mautrix/bridgev2/networkid"
+)
+
+type Key string
+
+const (
+	KeySplitPortalsEnabled Key = "split_portals_enabled"
+)
+
+type KVQuery struct {
+	BridgeID networkid.BridgeID
+	*dbutil.Database
+}
+
+const (
+	getKVQuery = `SELECT value FROM kv_store WHERE bridge_id = $1 AND key = $2`
+	setKVQuery = `
+		INSERT INTO kv_store (bridge_id, key, value) VALUES ($1, $2, $3)
+		ON CONFLICT (bridge_id, key) DO UPDATE SET value = $3
+	`
+)
+
+func (kvq *KVQuery) Get(ctx context.Context, key Key) string {
+	var value string
+	err := kvq.QueryRow(ctx, getKVQuery, kvq.BridgeID, key).Scan(&value)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		zerolog.Ctx(ctx).Err(err).Str("key", string(key)).Msg("Failed to get key from kvstore")
+	}
+	return value
+}
+
+func (kvq *KVQuery) Set(ctx context.Context, key Key, value string) {
+	_, err := kvq.Exec(ctx, setKVQuery, kvq.BridgeID, key, value)
+	if err != nil {
+		zerolog.Ctx(ctx).Err(err).
+			Str("key", string(key)).
+			Str("value", value).
+			Msg("Failed to set key in kvstore")
+	}
+}

bridgev2/database/portal.go

@@ -102,7 +102,19 @@ const (
 		DELETE FROM portal
 		WHERE bridge_id=$1 AND id=$2 AND receiver=$3
 	`
-	reIDPortalQuery = `UPDATE portal SET id=$4, receiver=$5 WHERE bridge_id=$1 AND id=$2 AND receiver=$3`
+	reIDPortalQuery            = `UPDATE portal SET id=$4, receiver=$5 WHERE bridge_id=$1 AND id=$2 AND receiver=$3`
+	migrateToSplitPortalsQuery = `
+		UPDATE portal
+		SET receiver=COALESCE((
+			SELECT login_id
+			FROM user_portal
+			WHERE bridge_id=portal.bridge_id AND portal_id=portal.id AND portal_receiver=''
+			LIMIT 1
+		), (
+			SELECT id FROM user_login WHERE bridge_id=portal.bridge_id LIMIT 1
+		), '')
+		WHERE receiver='' AND bridge_id=$1
+	`
 )
 
 func (pq *PortalQuery) GetByKey(ctx context.Context, key networkid.PortalKey) (*Portal, error) {
@@ -159,6 +171,14 @@ func (pq *PortalQuery) Delete(ctx context.Context, key networkid.PortalKey) erro
 	return pq.Exec(ctx, deletePortalQuery, pq.BridgeID, key.ID, key.Receiver)
 }
 
+func (pq *PortalQuery) MigrateToSplitPortals(ctx context.Context) (int64, error) {
+	res, err := pq.GetDB().Exec(ctx, migrateToSplitPortalsQuery, pq.BridgeID)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
 func (p *Portal) Scan(row dbutil.Scannable) (*Portal, error) {
 	var mxid, parentID, parentReceiver, relayLoginID, otherUserID, disappearType sql.NullString
 	var disappearTimer sql.NullInt64

bridgev2/database/upgrades/00-latest.sql

@@ -1,4 +1,4 @@
--- v0 -> v17 (compatible with v9+): Latest revision
+-- v0 -> v18 (compatible with v9+): Latest revision
 CREATE TABLE "user" (
 	bridge_id       TEXT NOT NULL,
 	mxid            TEXT NOT NULL,
@@ -198,3 +198,11 @@ CREATE TABLE backfill_task (
 		REFERENCES portal (bridge_id, id, receiver)
 		ON DELETE CASCADE ON UPDATE CASCADE
 );
+
+CREATE TABLE kv_store (
+	bridge_id TEXT NOT NULL,
+	key       TEXT NOT NULL,
+	value     TEXT NOT NULL,
+
+	PRIMARY KEY (bridge_id, key)
+);

bridgev2/database/upgrades/18-kv-store.sql

@@ -0,0 +1,8 @@
+-- v18 (compatible with v9+): Add generic key-value store
+CREATE TABLE kv_store (
+	bridge_id TEXT NOT NULL,
+	key       TEXT NOT NULL,
+	value     TEXT NOT NULL,
+
+	PRIMARY KEY (bridge_id, key)
+);

bridgev2/matrix/mxmain/example-config.yaml

@@ -15,6 +15,8 @@ bridge:
     # in the same Matrix room bridged to that group. If this is set to true,
     # every user will get their own Matrix room instead.
     split_portals: false
+    # Should the bridge resend `m.bridge` events to all portals on startup?
+    resend_bridge_info: false
 
     # Should leaving Matrix rooms be bridged as leaving groups on the remote network?
     bridge_matrix_leave: false