Use PDU in even more places (#3074)

- No longer rely on *Event returning from NewEventFrom... functions
 
Requires matrix-org/gomatrixserverlib#377

clientapi/routing/createroom.go

@@ -453,7 +453,7 @@ func createRoom(
 		if i > 0 {
 			builder.PrevEvents = []gomatrixserverlib.EventReference{builtEvents[i-1].EventReference()}
 		}
-		var ev *gomatrixserverlib.Event
+		var ev gomatrixserverlib.PDU
 		ev, err = builder.AddAuthEventsAndBuild(userDomain, &authEvents, evTime, roomVersion, cfg.Matrix.KeyID, cfg.Matrix.PrivateKey)
 		if err != nil {
 			util.GetLogger(ctx).WithError(err).Error("buildEvent failed")

federationapi/federationapi.go

@@ -53,7 +53,6 @@ func AddPublicRoutes(
 	keyRing gomatrixserverlib.JSONVerifier,
 	rsAPI roomserverAPI.FederationRoomserverAPI,
 	fedAPI federationAPI.FederationInternalAPI,
-	servers federationAPI.ServersInRoomProvider,
 	enableMetrics bool,
 ) {
 	cfg := &dendriteConfig.FederationAPI
@@ -87,7 +86,7 @@ func AddPublicRoutes(
 		dendriteConfig,
 		rsAPI, f, keyRing,
 		federation, userAPI, mscCfg,
-		servers, producer, enableMetrics,
+		producer, enableMetrics,
 	)
 }
 

federationapi/federationapi_test.go

@@ -313,7 +313,7 @@ func TestRoomsV3URLEscapeDoNot404(t *testing.T) {
 	natsInstance := jetstream.NATSInstance{}
 	// TODO: This is pretty fragile, as if anything calls anything on these nils this test will break.
 	// Unfortunately, it makes little sense to instantiate these dependencies when we just want to test routing.
-	federationapi.AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, nil, keyRing, nil, &internal.FederationInternalAPI{}, nil, caching.DisableMetrics)
+	federationapi.AddPublicRoutes(processCtx, routers, cfg, &natsInstance, nil, nil, keyRing, nil, &internal.FederationInternalAPI{}, caching.DisableMetrics)
 	baseURL, cancel := test.ListenAndServe(t, routers.Federation, true)
 	defer cancel()
 	serverName := spec.ServerName(strings.TrimPrefix(baseURL, "https://"))

federationapi/routing/profile_test.go

@@ -71,7 +71,7 @@ func TestHandleQueryProfile(t *testing.T) {
 		if !ok {
 			panic("This is a programming error.")
 		}
-		routing.Setup(routers, cfg, nil, r, keyRing, &fedClient, &userapi, &cfg.MSCs, nil, nil, caching.DisableMetrics)
+		routing.Setup(routers, cfg, nil, r, keyRing, &fedClient, &userapi, &cfg.MSCs, nil, caching.DisableMetrics)
 
 		handler := fedMux.Get(routing.QueryProfileRouteName).GetHandler().ServeHTTP
 		_, sk, _ := ed25519.GenerateKey(nil)

federationapi/routing/query_test.go

@@ -69,7 +69,7 @@ func TestHandleQueryDirectory(t *testing.T) {
 		if !ok {
 			panic("This is a programming error.")
 		}
-		routing.Setup(routers, cfg, nil, r, keyRing, &fedClient, &userapi, &cfg.MSCs, nil, nil, caching.DisableMetrics)
+		routing.Setup(routers, cfg, nil, r, keyRing, &fedClient, &userapi, &cfg.MSCs, nil, caching.DisableMetrics)
 
 		handler := fedMux.Get(routing.QueryDirectoryRouteName).GetHandler().ServeHTTP
 		_, sk, _ := ed25519.GenerateKey(nil)

federationapi/routing/routing.go

@@ -24,7 +24,6 @@ import (
 	"github.com/getsentry/sentry-go"
 	"github.com/gorilla/mux"
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
-	federationAPI "github.com/matrix-org/dendrite/federationapi/api"
 	fedInternal "github.com/matrix-org/dendrite/federationapi/internal"
 	"github.com/matrix-org/dendrite/federationapi/producers"
 	"github.com/matrix-org/dendrite/internal"
@@ -64,7 +63,6 @@ func Setup(
 	federation fclient.FederationClient,
 	userAPI userapi.FederationUserAPI,
 	mscCfg *config.MSCs,
-	servers federationAPI.ServersInRoomProvider,
 	producer *producers.SyncAPIProducer, enableMetrics bool,
 ) {
 	fedMux := routers.Federation
@@ -141,7 +139,7 @@ func Setup(
 		func(httpReq *http.Request, request *fclient.FederationRequest, vars map[string]string) util.JSONResponse {
 			return Send(
 				httpReq, request, gomatrixserverlib.TransactionID(vars["txnID"]),
-				cfg, rsAPI, userAPI, keys, federation, mu, servers, producer,
+				cfg, rsAPI, userAPI, keys, federation, mu, producer,
 			)
 		},
 	)).Methods(http.MethodPut, http.MethodOptions).Name(SendRouteName)

federationapi/routing/send.go

@@ -26,7 +26,6 @@ import (
 	"github.com/matrix-org/util"
 
 	"github.com/matrix-org/dendrite/clientapi/jsonerror"
-	federationAPI "github.com/matrix-org/dendrite/federationapi/api"
 	"github.com/matrix-org/dendrite/federationapi/producers"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -64,7 +63,6 @@ func Send(
 	keys gomatrixserverlib.JSONVerifier,
 	federation fclient.FederationClient,
 	mu *internal.MutexByRoom,
-	servers federationAPI.ServersInRoomProvider,
 	producer *producers.SyncAPIProducer,
 ) util.JSONResponse {
 	// First we should check if this origin has already submitted this

federationapi/routing/send_test.go

@@ -66,7 +66,7 @@ func TestHandleSend(t *testing.T) {
 		if !ok {
 			panic("This is a programming error.")
 		}
-		routing.Setup(routers, cfg, nil, r, keyRing, nil, nil, &cfg.MSCs, nil, nil, caching.DisableMetrics)
+		routing.Setup(routers, cfg, nil, r, keyRing, nil, nil, &cfg.MSCs, nil, caching.DisableMetrics)
 
 		handler := fedMux.Get(routing.SendRouteName).GetHandler().ServeHTTP
 		_, sk, _ := ed25519.GenerateKey(nil)

federationapi/routing/threepid.go

@@ -237,7 +237,7 @@ func createInviteFrom3PIDInvite(
 	cfg *config.FederationAPI,
 	inv invite, federation fclient.FederationClient,
 	userAPI userapi.FederationUserAPI,
-) (*gomatrixserverlib.Event, error) {
+) (gomatrixserverlib.PDU, error) {
 	_, server, err := gomatrixserverlib.SplitID('@', inv.MXID)
 	if err != nil {
 		return nil, err
@@ -293,7 +293,7 @@ func buildMembershipEvent(
 	ctx context.Context,
 	builder *gomatrixserverlib.EventBuilder, rsAPI api.FederationRoomserverAPI,
 	cfg *config.FederationAPI,
-) (*gomatrixserverlib.Event, error) {
+) (gomatrixserverlib.PDU, error) {
 	eventsNeeded, err := gomatrixserverlib.StateNeededForEventBuilder(builder)
 	if err != nil {
 		return nil, err

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/matrix-org/dugong v0.0.0-20210921133753-66e6b1c67e2e
 	github.com/matrix-org/go-sqlite3-js v0.0.0-20220419092513-28aa791a1c91
 	github.com/matrix-org/gomatrix v0.0.0-20220926102614-ceba4d9f7530
-	github.com/matrix-org/gomatrixserverlib v0.0.0-20230502133856-ad26780a085c
+	github.com/matrix-org/gomatrixserverlib v0.0.0-20230503081352-9e29bff996eb
 	github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a
 	github.com/matrix-org/util v0.0.0-20221111132719-399730281e66
 	github.com/mattn/go-sqlite3 v1.14.16

go.sum

@@ -331,6 +331,8 @@ github.com/matrix-org/gomatrixserverlib v0.0.0-20230502101247-782aebf83205 h1:fo
 github.com/matrix-org/gomatrixserverlib v0.0.0-20230502101247-782aebf83205/go.mod h1:7HTbSZe+CIdmeqVyFMekwD5dFU8khWQyngKATvd12FU=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20230502133856-ad26780a085c h1:5xXMu/08j8tWfiVUvD4yfs6mepz07BgC4kL2i0oGJX4=
 github.com/matrix-org/gomatrixserverlib v0.0.0-20230502133856-ad26780a085c/go.mod h1:7HTbSZe+CIdmeqVyFMekwD5dFU8khWQyngKATvd12FU=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20230503081352-9e29bff996eb h1:qg9iR39ctvB7A4hBcddjxmHQO/t3y4mpQnpmEc3xvNI=
+github.com/matrix-org/gomatrixserverlib v0.0.0-20230503081352-9e29bff996eb/go.mod h1:7HTbSZe+CIdmeqVyFMekwD5dFU8khWQyngKATvd12FU=
 github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a h1:awrPDf9LEFySxTLKYBMCiObelNx/cBuv/wzllvCCH3A=
 github.com/matrix-org/pinecone v0.11.1-0.20230210171230-8c3b24f2649a/go.mod h1:HchJX9oKMXaT2xYFs0Ha/6Zs06mxLU8k6F1ODnrGkeQ=
 github.com/matrix-org/util v0.0.0-20221111132719-399730281e66 h1:6z4KxomXSIGWqhHcfzExgkH3Z3UkIXry4ibJS4Aqz2Y=

internal/caching/cache_roomevents.go

@@ -2,22 +2,21 @@ package caching
 
 import (
 	"github.com/matrix-org/dendrite/roomserver/types"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // RoomServerEventsCache contains the subset of functions needed for
 // a roomserver event cache.
 type RoomServerEventsCache interface {
-	GetRoomServerEvent(eventNID types.EventNID) (*gomatrixserverlib.Event, bool)
-	StoreRoomServerEvent(eventNID types.EventNID, event *gomatrixserverlib.Event)
+	GetRoomServerEvent(eventNID types.EventNID) (*types.HeaderedEvent, bool)
+	StoreRoomServerEvent(eventNID types.EventNID, event *types.HeaderedEvent)
 	InvalidateRoomServerEvent(eventNID types.EventNID)
 }
 
-func (c Caches) GetRoomServerEvent(eventNID types.EventNID) (*gomatrixserverlib.Event, bool) {
+func (c Caches) GetRoomServerEvent(eventNID types.EventNID) (*types.HeaderedEvent, bool) {
 	return c.RoomServerEvents.Get(int64(eventNID))
 }
 
-func (c Caches) StoreRoomServerEvent(eventNID types.EventNID, event *gomatrixserverlib.Event) {
+func (c Caches) StoreRoomServerEvent(eventNID types.EventNID, event *types.HeaderedEvent) {
 	c.RoomServerEvents.Set(int64(eventNID), event)
 }
 

internal/caching/caches.go

@@ -28,7 +28,7 @@ type Caches struct {
 	ServerKeys              Cache[string, gomatrixserverlib.PublicKeyLookupResult] // server name -> server keys
 	RoomServerRoomNIDs      Cache[string, types.RoomNID]                           // room ID -> room NID
 	RoomServerRoomIDs       Cache[types.RoomNID, string]                           // room NID -> room ID
-	RoomServerEvents        Cache[int64, *gomatrixserverlib.Event]                 // event NID -> event
+	RoomServerEvents        Cache[int64, *types.HeaderedEvent]                     // event NID -> event
 	RoomServerStateKeys     Cache[types.EventStateKeyNID, string]                  // eventStateKey NID -> event state key
 	RoomServerStateKeyNIDs  Cache[string, types.EventStateKeyNID]                  // event state key -> eventStateKey NID
 	RoomServerEventTypeNIDs Cache[string, types.EventTypeNID]                      // eventType -> eventType NID

internal/caching/impl_ristretto.go

@@ -103,8 +103,8 @@ func NewRistrettoCache(maxCost config.DataUnit, maxAge time.Duration, enableProm
 			Prefix: roomIDsCache,
 			MaxAge: maxAge,
 		},
-		RoomServerEvents: &RistrettoCostedCachePartition[int64, *gomatrixserverlib.Event]{ // event NID -> event
-			&RistrettoCachePartition[int64, *gomatrixserverlib.Event]{
+		RoomServerEvents: &RistrettoCostedCachePartition[int64, *types.HeaderedEvent]{ // event NID -> event
+			&RistrettoCachePartition[int64, *types.HeaderedEvent]{
 				cache:   cache,
 				Prefix:  roomEventsCache,
 				MaxAge:  maxAge,

roomserver/internal/helpers/helpers.go

@@ -100,7 +100,7 @@ func IsServerCurrentlyInRoom(ctx context.Context, db storage.Database, serverNam
 func IsInvitePending(
 	ctx context.Context, db storage.Database,
 	roomID, userID string,
-) (bool, string, string, *gomatrixserverlib.Event, error) {
+) (bool, string, string, gomatrixserverlib.PDU, error) {
 	// Look up the room NID for the supplied room ID.
 	info, err := db.RoomInfo(ctx, roomID)
 	if err != nil {

roomserver/internal/input/input_missing.go

@@ -856,7 +856,7 @@ func (t *missingStateReq) lookupEvent(ctx context.Context, roomVersion gomatrixs
 			return events[0].PDU, nil
 		}
 	}
-	var event *gomatrixserverlib.Event
+	var event gomatrixserverlib.PDU
 	found := false
 	for _, serverName := range t.servers {
 		reqctx, cancel := context.WithTimeout(ctx, time.Second*30)

roomserver/internal/perform/perform_upgrade.go

@@ -473,7 +473,7 @@ func (r *Upgrader) sendInitialEvents(ctx context.Context, evTime time.Time, user
 		if i > 0 {
 			builder.PrevEvents = []gomatrixserverlib.EventReference{builtEvents[i-1].EventReference()}
 		}
-		var event *gomatrixserverlib.Event
+		var event gomatrixserverlib.PDU
 		event, err = builder.AddAuthEventsAndBuild(userDomain, &authEvents, evTime, newVersion, r.Cfg.Matrix.KeyID, r.Cfg.Matrix.PrivateKey)
 		if err != nil {
 			return fmt.Errorf("failed to build new %q event: %w", builder.Type, err)

roomserver/internal/query/query_test.go

@@ -26,12 +26,12 @@ import (
 
 // used to implement RoomserverInternalAPIEventDB to test getAuthChain
 type getEventDB struct {
-	eventMap map[string]*gomatrixserverlib.Event
+	eventMap map[string]gomatrixserverlib.PDU
 }
 
 func createEventDB() *getEventDB {
 	return &getEventDB{
-		eventMap: make(map[string]*gomatrixserverlib.Event),
+		eventMap: make(map[string]gomatrixserverlib.PDU),
 	}
 }
 

roomserver/storage/shared/storage.go

@@ -543,7 +543,7 @@ func (d *EventDatabase) events(
 	}
 
 	sort.Sort(inputEventNIDs)
-	events := make(map[types.EventNID]*gomatrixserverlib.Event, len(inputEventNIDs))
+	events := make(map[types.EventNID]gomatrixserverlib.PDU, len(inputEventNIDs))
 	eventNIDs := make([]types.EventNID, 0, len(inputEventNIDs))
 	for _, nid := range inputEventNIDs {
 		if event, ok := d.Cache.GetRoomServerEvent(nid); ok && event != nil {
@@ -593,7 +593,7 @@ func (d *EventDatabase) events(
 			return nil, err
 		}
 		if event := events[eventJSON.EventNID]; event != nil {
-			d.Cache.StoreRoomServerEvent(eventJSON.EventNID, event)
+			d.Cache.StoreRoomServerEvent(eventJSON.EventNID, &types.HeaderedEvent{PDU: event})
 		}
 	}
 	results := make([]types.Event, 0, len(inputEventNIDs))

setup/monolith.go

@@ -76,7 +76,7 @@ func (m *Monolith) AddAllPublicRoutes(
 		m.ExtPublicRoomsProvider, enableMetrics,
 	)
 	federationapi.AddPublicRoutes(
-		processCtx, routers, cfg, natsInstance, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, nil, enableMetrics,
+		processCtx, routers, cfg, natsInstance, m.UserAPI, m.FedClient, m.KeyRing, m.RoomserverAPI, m.FederationAPI, enableMetrics,
 	)
 	mediaapi.AddPublicRoutes(routers.Media, cm, cfg, m.UserAPI, m.Client)
 	syncapi.AddPublicRoutes(processCtx, routers, cfg, cm, natsInstance, m.UserAPI, m.RoomserverAPI, caches, enableMetrics)

syncapi/notifier/notifier.go

@@ -79,7 +79,7 @@ func (n *Notifier) SetCurrentPosition(currPos types.StreamingToken) {
 // OnNewEvent is called when a new event is received from the room server. Must only be
 // called from a single goroutine, to avoid races between updates which could set the
 // current sync position incorrectly.
-// Chooses which user sync streams to update by a provided *gomatrixserverlib.Event
+// Chooses which user sync streams to update by a provided gomatrixserverlib.PDU
 // (based on the users in the event's room),
 // a roomID directly, or a list of user IDs, prioritised by parameter ordering.
 // posUpdate contains the latest position(s) for one or more types of events.