Skip to content

Multi-Matrix-server support (Phase 1): registry, namespaced KV, v3 migration #29

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
fmartingr wants to merge 2 commits into from
Closed

Multi-Matrix-server support (Phase 1): registry, namespaced KV, v3 migration #29

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c1c29b1
Add multi-Matrix-server backend groundwork (registry, namespaced KV, …
fmartingr Jul 20, 2026
df81b75
chore: add element matrix client to docker compose
fmartingr Jul 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,12 +125,32 @@ For local development and testing, you can run a Matrix Synapse server using Doc

3. The Matrix server will be available at `http://localhost:8888`

### Accessing the Web Chat Interface (Element)

Synapse is only a homeserver and has no built-in chat UI. The Docker Compose stack
includes an [Element Web](https://element.io/) client for testing:

1. Start the Element service (included in `docker-compose up -d`, or start it alone):

```bash
docker-compose up -d element
```

2. Open `http://localhost:8880` in your browser.

3. It is pre-configured to use the local homeserver (`http://localhost:8888`), so you
can register or sign in with a test user directly. Registration is enabled for
development, so you can create new users from the login screen.

Element's configuration lives in `docker/element-config.json`.

### Configuration Notes

- The Synapse server is configured to use PostgreSQL as the database
- Registration is enabled for development purposes
- App service configuration is loaded from `docker/mattermost-bridge-registration.yaml`
- Room list publication is restricted to the bridge user only
- An Element Web client is available at `http://localhost:8880` for manual testing

### Stopping the Services

Expand Down
10 changes: 10 additions & 0 deletions docker-compose.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,16 @@ services:
- synapse-init
- postgres

element:
image: vectorim/element-web:latest
restart: unless-stopped
ports:
- "8880:80"
volumes:
- ./docker/element-config.json:/app/config.json:ro
depends_on:
- synapse

postgres:
image: postgres:14
restart: unless-stopped
Expand Down
13 changes: 13 additions & 0 deletions docker/element-config.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
{
"default_server_config": {
"m.homeserver": {
"base_url": "http://localhost:8888",
"server_name": "localhost"
}
},
"disable_custom_urls": false,
"disable_guests": false,
"brand": "Element (dev)",
"default_country_code": "US",
"show_labs_settings": true
}
85 changes: 74 additions & 11 deletions server/bridge_utils.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ type BridgeUtilsConfig struct {
API plugin.API
KVStore kvstore.KVStore
MatrixClient *matrix.Client
ServerID string
RemoteID string
MaxProfileImageSize int64
MaxFileSize int64
Expand All @@ -49,6 +50,7 @@ type BridgeUtils struct {
API plugin.API
kvstore kvstore.KVStore
matrixClient *matrix.Client
serverID string
remoteID string
maxProfileImageSize int64
maxFileSize int64
Expand All @@ -62,6 +64,7 @@ func NewBridgeUtils(config BridgeUtilsConfig) *BridgeUtils {
API: config.API,
kvstore: config.KVStore,
matrixClient: config.MatrixClient,
serverID: config.ServerID,
remoteID: config.RemoteID,
maxProfileImageSize: config.MaxProfileImageSize,
maxFileSize: config.MaxFileSize,
Expand All @@ -71,17 +74,33 @@ func NewBridgeUtils(config BridgeUtilsConfig) *BridgeUtils {

// Shared utility methods that both bridge types need

// GetMatrixRoomID retrieves the Matrix room ID for a given Mattermost channel ID
// GetMatrixRoomID retrieves the Matrix room ID for a given Mattermost channel ID.
// An unmapped channel yields ("", nil): the plugin KV API returns no error for a
// missing key, so the value is simply empty. A real KV read failure or a corrupt
// (unparseable) value is returned as an error rather than being masked as
// unmapped, which would silently mis-route or drop messages.
func (s *BridgeUtils) GetMatrixRoomID(channelID string) (string, error) {
roomID, err := s.kvstore.Get(kvstore.BuildChannelMappingKey(channelID))
data, err := s.kvstore.Get(kvstore.BuildChannelMappingKey(channelID))
if err != nil {
// KV store error (typically key not found) - unmapped channels are expected
return "", nil
return "", errors.Wrap(err, "failed to read channel mapping")
}
return string(roomID), nil
mappings, err := kvstore.ParseChannelServerMappings(data)
if err != nil {
// After the v3 migration every stored value is well-formed JSON, so a
// parse failure here is a corrupt record, not an unmapped channel.
return "", errors.Wrapf(err, "corrupt channel mapping value for channel %s", channelID)
}
return kvstore.RoomIDForServer(mappings, s.serverID), nil
}

func (s *BridgeUtils) setChannelRoomMapping(channelID, matrixRoomIdentifier string) error {
// Guard against persisting a mapping with an empty serverID, which would be
// unroutable and would corrupt the room_mapping_ reverse key. This should not
// happen once the registry is reconciled, but fail loudly rather than write it.
if s.serverID == "" {
return errors.New("cannot store channel mapping: server ID not initialized")
}

// Always resolve to room ID for consistent forward mapping storage
var roomID string
var err error
Expand All @@ -94,22 +113,37 @@ func (s *BridgeUtils) setChannelRoomMapping(channelID, matrixRoomIdentifier stri
roomID = matrixRoomIdentifier
}

// Store forward mapping: channel_mapping_<channelID> -> room_id (always room ID)
err = s.kvstore.Set(kvstore.BuildChannelMappingKey(channelID), []byte(roomID))
// Store forward mapping: channel_mapping_<channelID> -> [{serverID, room_id}].
// Upsert only this server's entry so a channel bridged to multiple homeservers
// keeps the others' mappings intact.
existingData, err := s.kvstore.Get(kvstore.BuildChannelMappingKey(channelID))
if err != nil {
return errors.Wrap(err, "failed to read existing channel room mapping")
}
mappings, err := kvstore.ParseChannelServerMappings(existingData)
if err != nil {
return errors.Wrap(err, "failed to parse existing channel room mapping")
}
mappings = kvstore.UpsertChannelServerMapping(mappings, s.serverID, roomID)
mappingValue, err := kvstore.MarshalChannelServerMappings(mappings)
if err != nil {
return errors.Wrap(err, "failed to marshal channel room mapping")
}
err = s.kvstore.Set(kvstore.BuildChannelMappingKey(channelID), mappingValue)
if err != nil {
return errors.Wrap(err, "failed to store channel room mapping")
}

// Store reverse mapping for the room ID
err = s.kvstore.Set(kvstore.BuildRoomMappingKey(roomID), []byte(channelID))
err = s.kvstore.Set(kvstore.BuildRoomMappingKey(s.serverID, roomID), []byte(channelID))
if err != nil {
return errors.Wrap(err, "failed to store reverse room mapping")
}

// If we started with an alias, also create reverse mapping for the alias
// This allows lookups by both alias and room ID
if strings.HasPrefix(matrixRoomIdentifier, "#") && roomID != matrixRoomIdentifier {
err = s.kvstore.Set(kvstore.BuildRoomMappingKey(matrixRoomIdentifier), []byte(channelID))
err = s.kvstore.Set(kvstore.BuildRoomMappingKey(s.serverID, matrixRoomIdentifier), []byte(channelID))
if err != nil {
s.logger.LogWarn("Failed to create alias reverse mapping", "channel_id", channelID, "room_alias", matrixRoomIdentifier, "error", err)
} else {
Expand All @@ -124,6 +158,35 @@ func (s *BridgeUtils) getConfiguration() *configuration {
return s.configGetter.getConfiguration()
}

// matrixUsernamePrefix returns the username prefix for this bridge's Matrix
// server, resolved from the managed server registry, which is the source of
// truth for per-server settings. reconcileServerConfig always populates each
// entry's prefix (defaulting to DefaultMatrixUsernamePrefix) before bridges run,
// so a configured server always resolves here. The static default is only for
// the degenerate case of no registered server (e.g. before the first reconcile).
// The registry is read live so prefix changes take effect without recreating the
// bridge.
func (s *BridgeUtils) matrixUsernamePrefix() string {
data, err := s.kvstore.Get(kvstore.KeyServersConfig)
if err != nil {
// A transient registry read failure must not silently change the prefix:
// a wrong prefix splits Matrix-user identity (ghosts created and matched
// under a different prefix). Surface it loudly, mirroring GetMatrixRoomID.
s.logger.LogError("Failed to read server registry for username prefix; using default", "server_id", s.serverID, "error", err)
return DefaultMatrixUsernamePrefix
}
servers, err := kvstore.ParseServersConfig(data)
if err != nil {
s.logger.LogError("Corrupt server registry; using default username prefix", "server_id", s.serverID, "error", err)
return DefaultMatrixUsernamePrefix
}
if server, ok := kvstore.ServerConfigForID(servers, s.serverID); ok && server.UsernamePrefix != "" {
return server.UsernamePrefix
}
// No registry yet (before the first reconcile) or no entry for this server.
return DefaultMatrixUsernamePrefix
}
Comment on lines +161 to +188

@coderabbitai coderabbitai Bot Jul 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟠 Major | ⚡ Quick win

Transient registry-read failure silently falls back to the exact "wrong prefix" scenario the comment warns against.

The doc comment says a registry-read failure "must not silently change the prefix" and should "surface it loudly, mirroring GetMatrixRoomID" — but unlike GetMatrixRoomID, this method returns only a string, so on KV read/parse failure it can only log and fall back to DefaultMatrixUsernamePrefix, which is a silent prefix change if the registry's actual entry differs from the default. Since this prefix drives ghost-user creation and Matrix-user reconstruction (line 501), a transient KV blip could momentarily split identity resolution.

Consider caching the last successfully-resolved prefix (per serverID) and falling back to that cached value instead of the static default on transient failure, reserving the static default only for the case where no successful read has ever occurred.

🔧 Suggested approach 
 type BridgeUtils struct {
 	logger              Logger
 	API                 plugin.API
 	kvstore             kvstore.KVStore
 	matrixClient        *matrix.Client
 	serverID            string
 	remoteID            string
 	maxProfileImageSize int64
 	maxFileSize         int64
 	configGetter        ConfigurationGetter
+	// cache of the last successfully-resolved username prefix, used as a
+	// safer fallback than the static default on transient registry errors
+	usernamePrefixMu    sync.RWMutex
+	lastKnownUsernamePrefix string
 }

 func (s *BridgeUtils) matrixUsernamePrefix() string {
 	data, err := s.kvstore.Get(kvstore.KeyServersConfig)
 	if err != nil {
 		s.logger.LogError("Failed to read server registry for username prefix; using last known value", "server_id", s.serverID, "error", err)
-		return DefaultMatrixUsernamePrefix
+		return s.fallbackUsernamePrefix()
 	}
 	servers, err := kvstore.ParseServersConfig(data)
 	if err != nil {
 		s.logger.LogError("Corrupt server registry; using last known username prefix", "server_id", s.serverID, "error", err)
-		return DefaultMatrixUsernamePrefix
+		return s.fallbackUsernamePrefix()
 	}
 	if server, ok := kvstore.ServerConfigForID(servers, s.serverID); ok && server.UsernamePrefix != "" {
+		s.usernamePrefixMu.Lock()
+		s.lastKnownUsernamePrefix = server.UsernamePrefix
+		s.usernamePrefixMu.Unlock()
 		return server.UsernamePrefix
 	}
-	return DefaultMatrixUsernamePrefix
+	return s.fallbackUsernamePrefix()
+}
+
+func (s *BridgeUtils) fallbackUsernamePrefix() string {
+	s.usernamePrefixMu.RLock()
+	defer s.usernamePrefixMu.RUnlock()
+	if s.lastKnownUsernamePrefix != "" {
+		return s.lastKnownUsernamePrefix
+	}
+	return DefaultMatrixUsernamePrefix
 }
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@server/bridge_utils.go` around lines 161 - 188, Update
BridgeUtils.matrixUsernamePrefix to cache the last successfully resolved
username prefix per serverID and return that cached value when the registry read
or ParseServersConfig fails. Preserve the static DefaultMatrixUsernamePrefix
only when no successful resolution has ever occurred, while continuing to update
the cache whenever a valid server prefix is found.


func (s *BridgeUtils) extractMattermostMetadata(event MatrixEvent) (postID string, remoteID string) {
if event.Content != nil {
if id, ok := event.Content["mattermost_post_id"].(string); ok {
Expand Down Expand Up @@ -284,7 +347,7 @@ func (s *BridgeUtils) getMattermostUsernameFromMatrix(matrixUserID string) strin
mattermostUserID = ghostMattermostUserID
} else {
// Check if we have a mapping for this regular Matrix user
userMapKey := "matrix_user_" + matrixUserID
userMapKey := kvstore.BuildMatrixUserKey(s.serverID, matrixUserID)
userIDBytes, err := s.kvstore.Get(userMapKey)
if err != nil || len(userIDBytes) == 0 {
s.logger.LogDebug("No Mattermost user found for Matrix mention", "matrix_user_id", matrixUserID)
Expand Down Expand Up @@ -435,7 +498,7 @@ func (s *BridgeUtils) reconstructMatrixUserIDFromUsername(mattermostUsername str
// We need to reverse this to get "@username:server.com"

config := s.configGetter.getConfiguration()
prefix := config.GetMatrixUsernamePrefixForServer(config.GetMatrixServerURL())
prefix := s.matrixUsernamePrefix()

// Check if username has the expected prefix
expectedPrefix := prefix + ":"
Expand Down
68 changes: 53 additions & 15 deletions server/command/command.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ import (
type Configuration interface {
GetMatrixServerURL() string
GetMatrixServerName() string
GetMatrixUsernamePrefixForServer(serverURL string) string
}

// MigrationResult holds the results of a migration operation
Expand Down Expand Up @@ -54,6 +53,9 @@ type PluginAccessor interface {
// Shared channel access
GetRemoteID() string

// Server registry access
GetServerID() string

// Migration access
RunKVStoreMigrations() error
RunKVStoreMigrationsWithResults() (*MigrationResult, error)
Expand Down Expand Up @@ -388,8 +390,12 @@ func (c *Handler) executeMapCommand(args *model.CommandArgs, roomIdentifier stri
}

// Save both directions of the mapping
serverID := c.plugin.GetServerID()
mappingKey := kvstore.BuildChannelMappingKey(args.ChannelId)
err := c.kvstore.Set(mappingKey, []byte(roomIdentifier))
mappingValue, err := kvstore.BuildSingleChannelMapping(serverID, roomIdentifier)
if err == nil {
err = c.kvstore.Set(mappingKey, mappingValue)
}
if err != nil {
c.client.Log.Error("Failed to save channel mapping", "error", err, "channel_id", args.ChannelId, "room_identifier", roomIdentifier)
return &model.CommandResponse{
Expand All @@ -398,8 +404,8 @@ func (c *Handler) executeMapCommand(args *model.CommandArgs, roomIdentifier stri
}
}

// Store reverse mapping: room_mapping_<roomIdentifier> -> channelID
roomMappingKey := kvstore.BuildRoomMappingKey(roomIdentifier)
// Store reverse mapping: room_mapping_<serverID>_<roomIdentifier> -> channelID
roomMappingKey := kvstore.BuildRoomMappingKey(serverID, roomIdentifier)
err = c.kvstore.Set(roomMappingKey, []byte(args.ChannelId))
if err != nil {
c.client.Log.Error("Failed to save room mapping", "error", err, "room_identifier", roomIdentifier, "channel_id", args.ChannelId)
Expand All @@ -409,7 +415,7 @@ func (c *Handler) executeMapCommand(args *model.CommandArgs, roomIdentifier stri
// If roomIdentifier is an alias, also resolve to room ID and store that mapping
if strings.HasPrefix(roomIdentifier, "#") {
if resolvedRoomID, err := matrixClient.ResolveRoomAlias(roomIdentifier); err == nil {
roomIDMappingKey := kvstore.BuildRoomMappingKey(resolvedRoomID)
roomIDMappingKey := kvstore.BuildRoomMappingKey(serverID, resolvedRoomID)
if err := c.kvstore.Set(roomIDMappingKey, []byte(args.ChannelId)); err != nil {
c.client.Log.Error("Failed to save room ID mapping", "error", err, "room_id", resolvedRoomID, "channel_id", args.ChannelId)
}
Expand Down Expand Up @@ -503,18 +509,36 @@ func (c *Handler) executeUnmapCommand(args *model.CommandArgs) *model.CommandRes
}

// Check if this channel has a Matrix room mapping
serverID := c.plugin.GetServerID()
channelMappingKey := kvstore.BuildChannelMappingKey(args.ChannelId)
roomIDBytes, err := c.kvstore.Get(channelMappingKey)
if err != nil {

// A corrupt (unparseable) value is distinct from an unmapped channel. Clear
// the bad record so the admin can recover with /matrix map, rather than being
// told the channel is not mapped with no way to fix it.
if err == nil && len(roomIDBytes) > 0 {
if _, parseErr := kvstore.ParseChannelServerMappings(roomIDBytes); parseErr != nil {
c.client.Log.Error("Corrupt channel mapping value; clearing it", "error", parseErr, "channel_id", args.ChannelId)
if delErr := c.kvstore.Delete(channelMappingKey); delErr != nil {
c.client.Log.Error("Failed to clear corrupt channel mapping", "error", delErr, "channel_id", args.ChannelId)
}
return &model.CommandResponse{
ResponseType: model.CommandResponseTypeEphemeral,
Text: fmt.Sprintf("⚠️ **Corrupt Mapping Cleared**\n\nChannel `%s` had an unreadable Matrix room mapping, which has been removed. Use `/matrix map` to remap it if needed.", channelName),
}
Comment on lines +522 to +528

@coderabbitai coderabbitai Bot Jul 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Report failure when the corrupt record cannot be deleted.

Line 522 logs a deletion failure, but Lines 525-528 still report that the mapping was cleared. Return an error response instead; the corrupt value remains and subsequent /matrix unmap calls will repeat this state.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@server/command/command.go` around lines 522 - 528, Update the corrupt channel
mapping cleanup branch around kvstore.Delete so a deletion failure returns an
error response instead of the “Corrupt Mapping Cleared” success response.
Preserve the existing success response only when Delete succeeds, and use the
existing command error-response pattern and symbols in this handler.

}
}

mappings, _ := kvstore.ParseChannelServerMappings(roomIDBytes)
matrixRoomIdentifier := kvstore.RoomIDForServer(mappings, serverID)
if err != nil || matrixRoomIdentifier == "" {
// Key not found is expected for unmapped channels
return &model.CommandResponse{
ResponseType: model.CommandResponseTypeEphemeral,
Text: fmt.Sprintf("❌ **No Mapping Found**\n\nChannel `%s` is not currently mapped to any Matrix room.", channelName),
}
}

matrixRoomIdentifier := string(roomIDBytes)

// Clear the Matrix room state to prevent fallback lookups - this is critical
matrixClient := c.plugin.GetMatrixClient()
if matrixClient == nil {
Expand Down Expand Up @@ -545,7 +569,7 @@ func (c *Handler) executeUnmapCommand(args *model.CommandArgs) *model.CommandRes
}

// Remove the room->channel mapping
roomMappingKey := kvstore.BuildRoomMappingKey(matrixRoomIdentifier)
roomMappingKey := kvstore.BuildRoomMappingKey(serverID, matrixRoomIdentifier)
if err := c.kvstore.Delete(roomMappingKey); err != nil {
c.client.Log.Warn("Failed to remove room mapping", "error", err, "room_identifier", matrixRoomIdentifier, "channel_id", args.ChannelId)
// Continue - the main mapping was removed
Expand Down Expand Up @@ -634,17 +658,22 @@ func (c *Handler) executeCreateRoomCommand(args *model.CommandArgs, roomName str
}

// Automatically map the created room to this channel (both directions)
serverID := c.plugin.GetServerID()
mappingKey := kvstore.BuildChannelMappingKey(args.ChannelId)
if err := c.kvstore.Set(mappingKey, []byte(roomID)); err != nil {
c.client.Log.Error("Failed to save channel mapping", "error", err, "channel_id", args.ChannelId, "room_id", roomID)
mappingValue, mErr := kvstore.BuildSingleChannelMapping(serverID, roomID)
if mErr == nil {
mErr = c.kvstore.Set(mappingKey, mappingValue)
}
if mErr != nil {
c.client.Log.Error("Failed to save channel mapping", "error", mErr, "channel_id", args.ChannelId, "room_id", roomID)
return &model.CommandResponse{
ResponseType: model.CommandResponseTypeEphemeral,
Text: fmt.Sprintf("✅ **Matrix Room Created:** `%s`\n\n❌ Failed to save channel mapping. Use `/matrix map %s` to map manually.", roomID, roomID),
}
}

// Store reverse mapping: room_mapping_<roomID> -> channelID
roomMappingKey := kvstore.BuildRoomMappingKey(roomID)
// Store reverse mapping: room_mapping_<serverID>_<roomID> -> channelID
roomMappingKey := kvstore.BuildRoomMappingKey(serverID, roomID)
if err := c.kvstore.Set(roomMappingKey, []byte(args.ChannelId)); err != nil {
c.client.Log.Error("Failed to save room mapping", "error", err, "room_id", roomID, "channel_id", args.ChannelId)
// Continue anyway - the forward mapping was saved successfully
Expand Down Expand Up @@ -672,6 +701,7 @@ func (c *Handler) executeListMappingsCommand(args *model.CommandArgs) *model.Com
responseText.WriteString("**Channel-to-Room Mappings:**\n\n")

// Get channel mapping keys using efficient prefix filtering
serverID := c.plugin.GetServerID()
mappings := make(map[string]string)
channelMappingPrefix := kvstore.KeyPrefixChannelMapping
page := 0
Expand All @@ -696,8 +726,16 @@ func (c *Handler) executeListMappingsCommand(args *model.CommandArgs) *model.Com
for _, key := range keys {
channelID := strings.TrimPrefix(key, channelMappingPrefix)
roomIDBytes, err := c.kvstore.Get(key)
if err == nil && len(roomIDBytes) > 0 {
mappings[channelID] = string(roomIDBytes)
if err != nil || len(roomIDBytes) == 0 {
continue
}
channelMappings, parseErr := kvstore.ParseChannelServerMappings(roomIDBytes)
if parseErr != nil {
c.client.Log.Warn("Failed to parse channel mapping value", "channel_id", channelID, "error", parseErr)
continue
}
if roomID := kvstore.RoomIDForServer(channelMappings, serverID); roomID != "" {
mappings[channelID] = roomID
}
}

Expand Down
Loading
Loading