Use the roomserver API to query senderID during sync event conversion

syncapi/routing/messages.go

@@ -495,7 +495,7 @@ func (r *messagesReq) handleNonEmptyEventsSlice(streamEvents []types.StreamEvent
 	}
 
 	// Append the events ve previously retrieved locally.
-	events = append(events, r.snapshot.StreamEventsToEvents(nil, streamEvents)...)
+	events = append(events, r.snapshot.StreamEventsToEvents(r.ctx, nil, streamEvents, r.rsAPI)...)
 	sort.Sort(eventsByDepth(events))
 
 	return

syncapi/storage/interface.go

@@ -44,8 +44,8 @@ type DatabaseTransaction interface {
 	MaxStreamPositionForRelations(ctx context.Context) (types.StreamPosition, error)
 
 	CurrentState(ctx context.Context, roomID string, stateFilterPart *synctypes.StateFilter, excludeEventIDs []string) ([]*rstypes.HeaderedEvent, error)
-	GetStateDeltasForFullStateSync(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *synctypes.StateFilter) ([]types.StateDelta, []string, error)
+	GetStateDeltasForFullStateSync(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *synctypes.StateFilter, rsAPI api.SyncRoomserverAPI) ([]types.StateDelta, []string, error)
-	GetStateDeltas(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *synctypes.StateFilter) ([]types.StateDelta, []string, error)
+	GetStateDeltas(ctx context.Context, device *userapi.Device, r types.Range, userID string, stateFilter *synctypes.StateFilter, rsAPI api.SyncRoomserverAPI) ([]types.StateDelta, []string, error)
 	RoomIDsWithMembership(ctx context.Context, userID string, membership string) ([]string, error)
 	MembershipCount(ctx context.Context, roomID, membership string, pos types.StreamPosition) (int, error)
 	GetRoomSummary(ctx context.Context, roomID, userID string) (summary *types.Summary, err error)
@@ -90,7 +90,7 @@ type DatabaseTransaction interface {
 	// StreamEventsToEvents converts streamEvent to Event. If device is non-nil and
 	// matches the streamevent.transactionID device then the transaction ID gets
 	// added to the unsigned section of the output event.
-	StreamEventsToEvents(device *userapi.Device, in []types.StreamEvent) []*rstypes.HeaderedEvent
+	StreamEventsToEvents(ctx context.Context, device *userapi.Device, in []types.StreamEvent, rsAPI api.SyncRoomserverAPI) []*rstypes.HeaderedEvent
 	// SendToDeviceUpdatesForSync returns a list of send-to-device updates. It returns the
 	// relevant events within the given ranges for the supplied user ID and device ID.
 	SendToDeviceUpdatesForSync(ctx context.Context, userID, deviceID string, from, to types.StreamPosition) (pos types.StreamPosition, events []types.SendToDeviceEvent, err error)

syncapi/storage/shared/storage_consumer.go

@@ -99,7 +99,41 @@ func (d *Database) Events(ctx context.Context, eventIDs []string) ([]*rstypes.He
 
 	// We don't include a device here as we only include transaction IDs in
 	// incremental syncs.
-	return d.StreamEventsToEvents(nil, streamEvents), nil
+	return d.StreamEventsToEvents(ctx, nil, streamEvents, nil), nil
+}
+
+func (d *Database) StreamEventsToEvents(ctx context.Context, device *userapi.Device, in []types.StreamEvent, rsAPI api.SyncRoomserverAPI) []*rstypes.HeaderedEvent {
+	out := make([]*rstypes.HeaderedEvent, len(in))
+	for i := 0; i < len(in); i++ {
+		out[i] = in[i].HeaderedEvent
+		if device != nil && in[i].TransactionID != nil {
+			userID, err := spec.NewUserID(device.UserID, true)
+			if err != nil {
+				logrus.WithFields(logrus.Fields{
+					"event_id": out[i].EventID(),
+				}).WithError(err).Warnf("Failed to add transaction ID to event")
+				continue
+			}
+			deviceSenderID, err := rsAPI.QuerySenderIDForUser(ctx, in[i].RoomID(), *userID)
+			if err != nil {
+				logrus.WithFields(logrus.Fields{
+					"event_id": out[i].EventID(),
+				}).WithError(err).Warnf("Failed to add transaction ID to event")
+				continue
+			}
+			if deviceSenderID == in[i].SenderID() && device.SessionID == in[i].TransactionID.SessionID {
+				err := out[i].SetUnsignedField(
+					"transaction_id", in[i].TransactionID.TransactionID,
+				)
+				if err != nil {
+					logrus.WithFields(logrus.Fields{
+						"event_id": out[i].EventID(),
+					}).WithError(err).Warnf("Failed to add transaction ID to event")
+				}
+			}
+		}
+	}
+	return out
 }
 
 // AddInviteEvent stores a new invite event for a user.
@@ -190,45 +224,6 @@ func (d *Database) UpsertAccountData(
 	return
 }
 
-func (d *Database) StreamEventsToEvents(device *userapi.Device, in []types.StreamEvent) []*rstypes.HeaderedEvent {
-	out := make([]*rstypes.HeaderedEvent, len(in))
-	for i := 0; i < len(in); i++ {
-		out[i] = in[i].HeaderedEvent
-		if device != nil && in[i].TransactionID != nil {
-			userID, err := spec.NewUserID(device.UserID, true)
-			if err != nil {
-				logrus.WithFields(logrus.Fields{
-					"event_id": out[i].EventID(),
-				}).WithError(err).Warnf("Failed to add transaction ID to event")
-				continue
-			}
-			deviceSenderID, err := d.getSenderIDForUser(in[i].RoomID(), *userID)
-			if err != nil {
-				logrus.WithFields(logrus.Fields{
-					"event_id": out[i].EventID(),
-				}).WithError(err).Warnf("Failed to add transaction ID to event")
-				continue
-			}
-			if deviceSenderID == in[i].SenderID() && device.SessionID == in[i].TransactionID.SessionID {
-				err := out[i].SetUnsignedField(
-					"transaction_id", in[i].TransactionID.TransactionID,
-				)
-				if err != nil {
-					logrus.WithFields(logrus.Fields{
-						"event_id": out[i].EventID(),
-					}).WithError(err).Warnf("Failed to add transaction ID to event")
-				}
-			}
-		}
-	}
-	return out
-}
-
-func (d *Database) getSenderIDForUser(roomID string, userID spec.UserID) (spec.SenderID, error) { // nolint
-	// TODO: Replace with actual logic for pseudoIDs
-	return spec.SenderID(userID.String()), nil
-}
-
 // handleBackwardExtremities adds this event as a backwards extremity if and only if we do not have all of
 // the events listed in the event's 'prev_events'. This function also updates the backwards extremities table
 // to account for the fact that the given event is no longer a backwards extremity, but may be marked as such.

syncapi/storage/shared/storage_sync.go

@@ -10,6 +10,7 @@ import (
 	"github.com/tidwall/gjson"
 
 	"github.com/matrix-org/dendrite/internal/eventutil"
+	"github.com/matrix-org/dendrite/roomserver/api"
 	rstypes "github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/dendrite/syncapi/synctypes"
 	"github.com/matrix-org/dendrite/syncapi/types"
@@ -186,7 +187,7 @@ func (d *DatabaseTransaction) Events(ctx context.Context, eventIDs []string) ([]
 
 	// We don't include a device here as we only include transaction IDs in
 	// incremental syncs.
-	return d.StreamEventsToEvents(nil, streamEvents), nil
+	return d.StreamEventsToEvents(ctx, nil, streamEvents, nil), nil
 }
 
 func (d *DatabaseTransaction) AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error) {
@@ -325,7 +326,7 @@ func (d *DatabaseTransaction) GetBackwardTopologyPos(
 func (d *DatabaseTransaction) GetStateDeltas(
 	ctx context.Context, device *userapi.Device,
 	r types.Range, userID string,
-	stateFilter *synctypes.StateFilter,
+	stateFilter *synctypes.StateFilter, rsAPI api.SyncRoomserverAPI,
 ) (deltas []types.StateDelta, joinedRoomsIDs []string, err error) {
 	// Implement membership change algorithm: https://github.com/matrix-org/synapse/blob/v0.19.3/synapse/handlers/sync.py#L821
 	// - Get membership list changes for this user in this sync response
@@ -417,7 +418,7 @@ func (d *DatabaseTransaction) GetStateDeltas(
 		if !peek.Deleted {
 			deltas = append(deltas, types.StateDelta{
 				Membership:  spec.Peek,
-				StateEvents: d.StreamEventsToEvents(device, state[peek.RoomID]),
+				StateEvents: d.StreamEventsToEvents(ctx, device, state[peek.RoomID], rsAPI),
 				RoomID:      peek.RoomID,
 			})
 		}
@@ -462,7 +463,7 @@ func (d *DatabaseTransaction) GetStateDeltas(
 			deltas = append(deltas, types.StateDelta{
 				Membership:    membership,
 				MembershipPos: ev.StreamPosition,
-				StateEvents:   d.StreamEventsToEvents(device, stateFiltered[roomID]),
+				StateEvents:   d.StreamEventsToEvents(ctx, device, stateFiltered[roomID], rsAPI),
 				RoomID:        roomID,
 			})
 			break
@@ -474,7 +475,7 @@ func (d *DatabaseTransaction) GetStateDeltas(
 	for _, joinedRoomID := range joinedRoomIDs {
 		deltas = append(deltas, types.StateDelta{
 			Membership:  spec.Join,
-			StateEvents: d.StreamEventsToEvents(device, stateFiltered[joinedRoomID]),
+			StateEvents: d.StreamEventsToEvents(ctx, device, stateFiltered[joinedRoomID], rsAPI),
 			RoomID:      joinedRoomID,
 			NewlyJoined: newlyJoinedRooms[joinedRoomID],
 		})
@@ -490,7 +491,7 @@ func (d *DatabaseTransaction) GetStateDeltas(
 func (d *DatabaseTransaction) GetStateDeltasForFullStateSync(
 	ctx context.Context, device *userapi.Device,
 	r types.Range, userID string,
-	stateFilter *synctypes.StateFilter,
+	stateFilter *synctypes.StateFilter, rsAPI api.SyncRoomserverAPI,
 ) ([]types.StateDelta, []string, error) {
 	// Look up all memberships for the user. We only care about rooms that a
 	// user has ever interacted with — joined to, kicked/banned from, left.
@@ -531,7 +532,7 @@ func (d *DatabaseTransaction) GetStateDeltasForFullStateSync(
 			}
 			deltas[peek.RoomID] = types.StateDelta{
 				Membership:  spec.Peek,
-				StateEvents: d.StreamEventsToEvents(device, s),
+				StateEvents: d.StreamEventsToEvents(ctx, device, s, rsAPI),
 				RoomID:      peek.RoomID,
 			}
 		}
@@ -560,7 +561,7 @@ func (d *DatabaseTransaction) GetStateDeltasForFullStateSync(
 					deltas[roomID] = types.StateDelta{
 						Membership:    membership,
 						MembershipPos: ev.StreamPosition,
-						StateEvents:   d.StreamEventsToEvents(device, stateStreamEvents),
+						StateEvents:   d.StreamEventsToEvents(ctx, device, stateStreamEvents, rsAPI),
 						RoomID:        roomID,
 					}
 				}
@@ -581,7 +582,7 @@ func (d *DatabaseTransaction) GetStateDeltasForFullStateSync(
 		}
 		deltas[joinedRoomID] = types.StateDelta{
 			Membership:  spec.Join,
-			StateEvents: d.StreamEventsToEvents(device, s),
+			StateEvents: d.StreamEventsToEvents(ctx, device, s, rsAPI),
 			RoomID:      joinedRoomID,
 		}
 	}

syncapi/storage/storage_test.go

@@ -214,7 +214,7 @@ func TestGetEventsInRangeWithTopologyToken(t *testing.T) {
 			if err != nil {
 				t.Fatalf("GetEventsInTopologicalRange returned an error: %s", err)
 			}
-			gots := snapshot.StreamEventsToEvents(nil, paginatedEvents)
+			gots := snapshot.StreamEventsToEvents(context.Background(), nil, paginatedEvents, nil)
 			test.AssertEventsEqual(t, gots, test.Reversed(events[len(events)-5:]))
 		})
 	})

syncapi/streams/stream_pdu.go

@@ -175,12 +175,12 @@ func (p *PDUStreamProvider) IncrementalSync(
 	eventFilter := req.Filter.Room.Timeline
 
 	if req.WantFullState {
-		if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltasForFullStateSync(ctx, req.Device, r, req.Device.UserID, &stateFilter); err != nil {
+		if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltasForFullStateSync(ctx, req.Device, r, req.Device.UserID, &stateFilter, p.rsAPI); err != nil {
 			req.Log.WithError(err).Error("p.DB.GetStateDeltasForFullStateSync failed")
 			return from
 		}
 	} else {
-		if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltas(ctx, req.Device, r, req.Device.UserID, &stateFilter); err != nil {
+		if stateDeltas, syncJoinedRooms, err = snapshot.GetStateDeltas(ctx, req.Device, r, req.Device.UserID, &stateFilter, p.rsAPI); err != nil {
 			req.Log.WithError(err).Error("p.DB.GetStateDeltas failed")
 			return from
 		}
@@ -275,7 +275,7 @@ func (p *PDUStreamProvider) addRoomDeltaToResponse(
 	limited := dbEvents[delta.RoomID].Limited
 
 	recEvents := gomatrixserverlib.ReverseTopologicalOrdering(
-		gomatrixserverlib.ToPDUs(snapshot.StreamEventsToEvents(device, recentStreamEvents)),
+		gomatrixserverlib.ToPDUs(snapshot.StreamEventsToEvents(ctx, device, recentStreamEvents, p.rsAPI)),
 		gomatrixserverlib.TopologicalOrderByPrevEvents,
 	)
 	recentEvents := make([]*rstypes.HeaderedEvent, len(recEvents))
@@ -512,7 +512,7 @@ func (p *PDUStreamProvider) getJoinResponseForCompleteSync(
 	// We don't include a device here as we don't need to send down
 	// transaction IDs for complete syncs, but we do it anyway because Sytest demands it for:
 	// "Can sync a room with a message with a transaction id" - which does a complete sync to check.
-	recentEvents := snapshot.StreamEventsToEvents(device, recentStreamEvents)
+	recentEvents := snapshot.StreamEventsToEvents(ctx, device, recentStreamEvents, p.rsAPI)
 
 	events := recentEvents
 	// Only apply history visibility checks if the response is for joined rooms