Merge branch 'matrix-org:main' into main

CHANGES.md

@@ -1,5 +1,13 @@
 # Changelog
 
+## Dendrite 0.8.4 (2022-05-10)
+
+### Fixes
+
+* Fixes a regression introduced in the previous version where appservices, push and phone-home statistics would not work over plain HTTP
+* Adds missing indexes to the sync API output events table, which should significantly improve `/sync` performance and reduce database CPU usage
+* Building Dendrite with the `bimg` thumbnailer should now work again (contributed by [database64128](https://github.com/database64128))
+
 ## Dendrite 0.8.3 (2022-05-09)
 
 ### Features

appservice/appservice.go

@@ -16,6 +16,8 @@ package appservice
 
 import (
 	"context"
+	"crypto/tls"
+	"net/http"
 	"sync"
 	"time"
 
@@ -33,7 +35,6 @@ import (
 	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/config"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
-	"github.com/matrix-org/gomatrixserverlib"
 )
 
 // AddInternalRoutes registers HTTP handlers for internal API calls
@@ -45,15 +46,19 @@ func AddInternalRoutes(router *mux.Router, queryAPI appserviceAPI.AppServiceInte
 // can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
 func NewInternalAPI(
 	base *base.BaseDendrite,
-	userAPI userapi.AppserviceUserAPI,
+	userAPI userapi.UserInternalAPI,
-	rsAPI roomserverAPI.AppserviceRoomserverAPI,
+	rsAPI roomserverAPI.RoomserverInternalAPI,
 ) appserviceAPI.AppServiceInternalAPI {
-	client := gomatrixserverlib.NewClient(
+	client := &http.Client{
-		gomatrixserverlib.WithTimeout(time.Second*30),
+		Timeout: time.Second * 30,
-		gomatrixserverlib.WithKeepAlives(false),
+		Transport: &http.Transport{
-		gomatrixserverlib.WithSkipVerify(base.Cfg.AppServiceAPI.DisableTLSValidation),
+			DisableKeepAlives: true,
-	)
+			TLSClientConfig: &tls.Config{
-
+				InsecureSkipVerify: base.Cfg.AppServiceAPI.DisableTLSValidation,
+			},
+			Proxy: http.ProxyFromEnvironment,
+		},
+	}
 	js, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
 
 	// Create a connection to the appservice postgres DB

appservice/query/query.go

@@ -23,7 +23,6 @@ import (
 
 	"github.com/matrix-org/dendrite/appservice/api"
 	"github.com/matrix-org/dendrite/setup/config"
-	"github.com/matrix-org/gomatrixserverlib"
 	opentracing "github.com/opentracing/opentracing-go"
 	log "github.com/sirupsen/logrus"
 )
@@ -33,7 +32,7 @@ const userIDExistsPath = "/users/"
 
 // AppServiceQueryAPI is an implementation of api.AppServiceQueryAPI
 type AppServiceQueryAPI struct {
-	HTTPClient *gomatrixserverlib.Client
+	HTTPClient *http.Client
 	Cfg        *config.Dendrite
 }
 
@@ -65,8 +64,9 @@ func (a *AppServiceQueryAPI) RoomAliasExists(
 			if err != nil {
 				return err
 			}
+			req = req.WithContext(ctx)
 
-			resp, err := a.HTTPClient.DoHTTPRequest(ctx, req)
+			resp, err := a.HTTPClient.Do(req)
 			if resp != nil {
 				defer func() {
 					err = resp.Body.Close()
@@ -130,7 +130,7 @@ func (a *AppServiceQueryAPI) UserIDExists(
 			if err != nil {
 				return err
 			}
-			resp, err := a.HTTPClient.DoHTTPRequest(ctx, req)
+			resp, err := a.HTTPClient.Do(req.WithContext(ctx))
 			if resp != nil {
 				defer func() {
 					err = resp.Body.Close()

appservice/workers/transaction_scheduler.go

@@ -42,7 +42,7 @@ var (
 // size), then send that off to the AS's /transactions/{txnID} endpoint. It also
 // handles exponentially backing off in case the AS isn't currently available.
 func SetupTransactionWorkers(
-	client *gomatrixserverlib.Client,
+	client *http.Client,
 	appserviceDB storage.Database,
 	workerStates []types.ApplicationServiceWorkerState,
 ) error {
@@ -58,7 +58,7 @@ func SetupTransactionWorkers(
 
 // worker is a goroutine that sends any queued events to the application service
 // it is given.
-func worker(client *gomatrixserverlib.Client, db storage.Database, ws types.ApplicationServiceWorkerState) {
+func worker(client *http.Client, db storage.Database, ws types.ApplicationServiceWorkerState) {
 	log.WithFields(log.Fields{
 		"appservice": ws.AppService.ID,
 	}).Info("Starting application service")
@@ -200,7 +200,7 @@ func createTransaction(
 // send sends events to an application service. Returns an error if an OK was not
 // received back from the application service or the request timed out.
 func send(
-	client *gomatrixserverlib.Client,
+	client *http.Client,
 	appservice config.ApplicationService,
 	txnID int,
 	transaction []byte,
@@ -213,7 +213,7 @@ func send(
 		return err
 	}
 	req.Header.Set("Content-Type", "application/json")
-	resp, err := client.DoHTTPRequest(context.TODO(), req)
+	resp, err := client.Do(req)
 	if err != nil {
 		return err
 	}

clientapi/routing/membership.go

@@ -188,6 +188,12 @@ func SendUnban(
 	if err != nil {
 		return util.ErrorResponse(err)
 	}
+	if !queryRes.RoomExists {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("room does not exist"),
+		}
+	}
 	// unban is only valid if the user is currently banned
 	if queryRes.Membership != "ban" {
 		return util.JSONResponse{
@@ -471,6 +477,12 @@ func SendForget(
 		logger.WithError(err).Error("QueryMembershipForUser: could not query membership for user")
 		return jsonerror.InternalServerError()
 	}
+	if !membershipRes.RoomExists {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("room does not exist"),
+		}
+	}
 	if membershipRes.IsInRoom {
 		return util.JSONResponse{
 			Code: http.StatusBadRequest,

clientapi/routing/state.go

@@ -56,6 +56,12 @@ func OnIncomingStateRequest(ctx context.Context, device *userapi.Device, rsAPI a
 		util.GetLogger(ctx).WithError(err).Error("queryAPI.QueryLatestEventsAndState failed")
 		return jsonerror.InternalServerError()
 	}
+	if !stateRes.RoomExists {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("room does not exist"),
+		}
+	}
 
 	// Look at the room state and see if we have a history visibility event
 	// that marks the room as world-readable. If we don't then we assume that

clientapi/threepid/invites.go

@@ -231,7 +231,7 @@ func queryIDServerStoreInvite(
 		profile = &authtypes.Profile{}
 	}
 
-	client := gomatrixserverlib.NewClient()
+	client := http.Client{}
 
 	data := url.Values{}
 	data.Add("medium", body.Medium)
@@ -253,7 +253,7 @@ func queryIDServerStoreInvite(
 	}
 
 	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
-	resp, err := client.DoHTTPRequest(ctx, req)
+	resp, err := client.Do(req.WithContext(ctx))
 	if err != nil {
 		return nil, err
 	}

internal/pushgateway/client.go

@@ -3,28 +3,32 @@ package pushgateway
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"net/http"
 	"time"
 
-	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/opentracing/opentracing-go"
 )
 
 type httpClient struct {
-	hc *gomatrixserverlib.Client
+	hc *http.Client
 }
 
 // NewHTTPClient creates a new Push Gateway client.
 func NewHTTPClient(disableTLSValidation bool) Client {
-	return &httpClient{
+	hc := &http.Client{
-		hc: gomatrixserverlib.NewClient(
+		Timeout: 30 * time.Second,
-			gomatrixserverlib.WithTimeout(time.Second*30),
+		Transport: &http.Transport{
-			gomatrixserverlib.WithKeepAlives(false),
+			DisableKeepAlives: true,
-			gomatrixserverlib.WithSkipVerify(disableTLSValidation),
+			TLSClientConfig: &tls.Config{
-		),
+				InsecureSkipVerify: disableTLSValidation,
+			},
+			Proxy: http.ProxyFromEnvironment,
+		},
 	}
+	return &httpClient{hc: hc}
 }
 
 func (h *httpClient) Notify(ctx context.Context, url string, req *NotifyRequest, resp *NotifyResponse) error {
@@ -41,7 +45,7 @@ func (h *httpClient) Notify(ctx context.Context, url string, req *NotifyRequest,
 	}
 	hreq.Header.Set("Content-Type", "application/json")
 
-	hresp, err := h.hc.DoHTTPRequest(ctx, hreq)
+	hresp, err := h.hc.Do(hreq)
 	if err != nil {
 		return err
 	}

internal/version.go

@@ -17,7 +17,7 @@ var build string
 const (
 	VersionMajor = 0
 	VersionMinor = 8
-	VersionPatch = 3
+	VersionPatch = 4
 	VersionTag   = "" // example: "rc1"
 )
 

mediaapi/thumbnailer/thumbnailer_bimg.go

@@ -37,7 +37,7 @@ func GenerateThumbnails(
 	mediaMetadata *types.MediaMetadata,
 	activeThumbnailGeneration *types.ActiveThumbnailGeneration,
 	maxThumbnailGenerators int,
-	db *storage.Database,
+	db storage.Database,
 	logger *log.Entry,
 ) (busy bool, errorReturn error) {
 	buffer, err := bimg.Read(string(src))
@@ -49,7 +49,7 @@ func GenerateThumbnails(
 	for _, config := range configs {
 		// Note: createThumbnail does locking based on activeThumbnailGeneration
 		busy, err = createThumbnail(
-			ctx, src, img, config, mediaMetadata, activeThumbnailGeneration,
+			ctx, src, img, types.ThumbnailSize(config), mediaMetadata, activeThumbnailGeneration,
 			maxThumbnailGenerators, db, logger,
 		)
 		if err != nil {
@@ -71,7 +71,7 @@ func GenerateThumbnail(
 	mediaMetadata *types.MediaMetadata,
 	activeThumbnailGeneration *types.ActiveThumbnailGeneration,
 	maxThumbnailGenerators int,
-	db *storage.Database,
+	db storage.Database,
 	logger *log.Entry,
 ) (busy bool, errorReturn error) {
 	buffer, err := bimg.Read(string(src))
@@ -109,7 +109,7 @@ func createThumbnail(
 	mediaMetadata *types.MediaMetadata,
 	activeThumbnailGeneration *types.ActiveThumbnailGeneration,
 	maxThumbnailGenerators int,
-	db *storage.Database,
+	db storage.Database,
 	logger *log.Entry,
 ) (busy bool, errorReturn error) {
 	logger = logger.WithFields(log.Fields{

roomserver/api/query.go

@@ -122,6 +122,7 @@ type QueryMembershipForUserResponse struct {
 	Membership string `json:"membership"`
 	// True if the user asked to forget this room.
 	IsRoomForgotten bool `json:"is_room_forgotten"`
+	RoomExists      bool `json:"room_exists"`
 }
 
 // QueryMembershipsForRoomRequest is a request to QueryMembershipsForRoom

roomserver/internal/query/query.go

@@ -169,8 +169,10 @@ func (r *Queryer) QueryMembershipForUser(
 		return err
 	}
 	if info == nil {
-		return fmt.Errorf("QueryMembershipForUser: unknown room %s", request.RoomID)
+		response.RoomExists = false
+		return nil
 	}
+	response.RoomExists = true
 
 	membershipEventNID, stillInRoom, isRoomforgotten, err := r.DB.GetMembership(ctx, info.RoomNID, request.UserID)
 	if err != nil {

roomserver/storage/postgres/invite_table.go

@@ -81,12 +81,12 @@ type inviteStatements struct {
 	updateInviteRetiredStmt             *sql.Stmt
 }
 
-func createInvitesTable(db *sql.DB) error {
+func CreateInvitesTable(db *sql.DB) error {
 	_, err := db.Exec(inviteSchema)
 	return err
 }
 
-func prepareInvitesTable(db *sql.DB) (tables.Invites, error) {
+func PrepareInvitesTable(db *sql.DB) (tables.Invites, error) {
 	s := &inviteStatements{}
 
 	return s, sqlutil.StatementList{
@@ -127,8 +127,8 @@ func (s *inviteStatements) UpdateInviteRetired(
 	defer internal.CloseAndLogIfError(ctx, rows, "updateInviteRetired: rows.close() failed")
 
 	var eventIDs []string
+	var inviteEventID string
 	for rows.Next() {
-		var inviteEventID string
 		if err = rows.Scan(&inviteEventID); err != nil {
 			return nil, err
 		}
@@ -152,9 +152,9 @@ func (s *inviteStatements) SelectInviteActiveForUserInRoom(
 	defer internal.CloseAndLogIfError(ctx, rows, "selectInviteActiveForUserInRoom: rows.close() failed")
 	var result []types.EventStateKeyNID
 	var eventIDs []string
+	var inviteEventID string
+	var senderUserNID int64
 	for rows.Next() {
-		var inviteEventID string
-		var senderUserNID int64
 		if err := rows.Scan(&inviteEventID, &senderUserNID); err != nil {
 			return nil, nil, err
 		}

roomserver/storage/postgres/membership_table.go

@@ -160,12 +160,12 @@ type membershipStatements struct {
 	selectServerInRoomStmt                          *sql.Stmt
 }
 
-func createMembershipTable(db *sql.DB) error {
+func CreateMembershipTable(db *sql.DB) error {
 	_, err := db.Exec(membershipSchema)
 	return err
 }
 
-func prepareMembershipTable(db *sql.DB) (tables.Membership, error) {
+func PrepareMembershipTable(db *sql.DB) (tables.Membership, error) {
 	s := &membershipStatements{}
 
 	return s, sqlutil.StatementList{
@@ -234,8 +234,8 @@ func (s *membershipStatements) SelectMembershipsFromRoom(
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "selectMembershipsFromRoom: rows.close() failed")
 
+	var eNID types.EventNID
 	for rows.Next() {
-		var eNID types.EventNID
 		if err = rows.Scan(&eNID); err != nil {
 			return
 		}
@@ -262,8 +262,8 @@ func (s *membershipStatements) SelectMembershipsFromRoomAndMembership(
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "selectMembershipsFromRoomAndMembership: rows.close() failed")
 
+	var eNID types.EventNID
 	for rows.Next() {
-		var eNID types.EventNID
 		if err = rows.Scan(&eNID); err != nil {
 			return
 		}
@@ -298,8 +298,8 @@ func (s *membershipStatements) SelectRoomsWithMembership(
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "SelectRoomsWithMembership: rows.close() failed")
 	var roomNIDs []types.RoomNID
+	var roomNID types.RoomNID
 	for rows.Next() {
-		var roomNID types.RoomNID
 		if err := rows.Scan(&roomNID); err != nil {
 			return nil, err
 		}
@@ -320,9 +320,9 @@ func (s *membershipStatements) SelectJoinedUsersSetForRooms(
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "selectJoinedUsersSetForRooms: rows.close() failed")
 	result := make(map[types.EventStateKeyNID]int)
+	var userID types.EventStateKeyNID
+	var count int
 	for rows.Next() {
-		var userID types.EventStateKeyNID
-		var count int
 		if err := rows.Scan(&userID, &count); err != nil {
 			return nil, err
 		}
@@ -342,12 +342,12 @@ func (s *membershipStatements) SelectKnownUsers(
 	}
 	result := []string{}
 	defer internal.CloseAndLogIfError(ctx, rows, "SelectKnownUsers: rows.close() failed")
+	var resUserID string
 	for rows.Next() {
-		var userID string
+		if err := rows.Scan(&resUserID); err != nil {
-		if err := rows.Scan(&userID); err != nil {
 			return nil, err
 		}
-		result = append(result, userID)
+		result = append(result, resUserID)
 	}
 	return result, rows.Err()
 }

roomserver/storage/postgres/previous_events_table.go

@@ -64,12 +64,12 @@ type previousEventStatements struct {
 	selectPreviousEventExistsStmt *sql.Stmt
 }
 
-func createPrevEventsTable(db *sql.DB) error {
+func CreatePrevEventsTable(db *sql.DB) error {
 	_, err := db.Exec(previousEventSchema)
 	return err
 }
 
-func preparePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
+func PreparePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
 	s := &previousEventStatements{}
 
 	return s, sqlutil.StatementList{

roomserver/storage/postgres/published_table.go

@@ -49,12 +49,12 @@ type publishedStatements struct {
 	selectPublishedStmt    *sql.Stmt
 }
 
-func createPublishedTable(db *sql.DB) error {
+func CreatePublishedTable(db *sql.DB) error {
 	_, err := db.Exec(publishedSchema)
 	return err
 }
 
-func preparePublishedTable(db *sql.DB) (tables.Published, error) {
+func PreparePublishedTable(db *sql.DB) (tables.Published, error) {
 	s := &publishedStatements{}
 
 	return s, sqlutil.StatementList{
@@ -94,8 +94,8 @@ func (s *publishedStatements) SelectAllPublishedRooms(
 	defer internal.CloseAndLogIfError(ctx, rows, "selectAllPublishedStmt: rows.close() failed")
 
 	var roomIDs []string
+	var roomID string
 	for rows.Next() {
-		var roomID string
 		if err = rows.Scan(&roomID); err != nil {
 			return nil, err
 		}

roomserver/storage/postgres/redactions_table.go

@@ -59,12 +59,12 @@ type redactionStatements struct {
 	markRedactionValidatedStmt                  *sql.Stmt
 }
 
-func createRedactionsTable(db *sql.DB) error {
+func CreateRedactionsTable(db *sql.DB) error {
 	_, err := db.Exec(redactionsSchema)
 	return err
 }
 
-func prepareRedactionsTable(db *sql.DB) (tables.Redactions, error) {
+func PrepareRedactionsTable(db *sql.DB) (tables.Redactions, error) {
 	s := &redactionStatements{}
 
 	return s, sqlutil.StatementList{

roomserver/storage/postgres/storage.go

@@ -89,22 +89,22 @@ func (d *Database) create(db *sql.DB) error {
 	if err := createStateSnapshotTable(db); err != nil {
 		return err
 	}
-	if err := createPrevEventsTable(db); err != nil {
+	if err := CreatePrevEventsTable(db); err != nil {
 		return err
 	}
 	if err := createRoomAliasesTable(db); err != nil {
 		return err
 	}
-	if err := createInvitesTable(db); err != nil {
+	if err := CreateInvitesTable(db); err != nil {
 		return err
 	}
-	if err := createMembershipTable(db); err != nil {
+	if err := CreateMembershipTable(db); err != nil {
 		return err
 	}
-	if err := createPublishedTable(db); err != nil {
+	if err := CreatePublishedTable(db); err != nil {
 		return err
 	}
-	if err := createRedactionsTable(db); err != nil {
+	if err := CreateRedactionsTable(db); err != nil {
 		return err
 	}
 
@@ -140,7 +140,7 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
 	if err != nil {
 		return err
 	}
-	prevEvents, err := preparePrevEventsTable(db)
+	prevEvents, err := PreparePrevEventsTable(db)
 	if err != nil {
 		return err
 	}
@@ -148,19 +148,19 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
 	if err != nil {
 		return err
 	}
-	invites, err := prepareInvitesTable(db)
+	invites, err := PrepareInvitesTable(db)
 	if err != nil {
 		return err
 	}
-	membership, err := prepareMembershipTable(db)
+	membership, err := PrepareMembershipTable(db)
 	if err != nil {
 		return err
 	}
-	published, err := preparePublishedTable(db)
+	published, err := PreparePublishedTable(db)
 	if err != nil {
 		return err
 	}
-	redactions, err := prepareRedactionsTable(db)
+	redactions, err := PrepareRedactionsTable(db)
 	if err != nil {
 		return err
 	}

roomserver/storage/sqlite3/invite_table.go

@@ -69,12 +69,12 @@ type inviteStatements struct {
 	selectInvitesAboutToRetireStmt      *sql.Stmt
 }
 
-func createInvitesTable(db *sql.DB) error {
+func CreateInvitesTable(db *sql.DB) error {
 	_, err := db.Exec(inviteSchema)
 	return err
 }
 
-func prepareInvitesTable(db *sql.DB) (tables.Invites, error) {
+func PrepareInvitesTable(db *sql.DB) (tables.Invites, error) {
 	s := &inviteStatements{
 		db: db,
 	}
@@ -119,8 +119,8 @@ func (s *inviteStatements) UpdateInviteRetired(
 		return
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "UpdateInviteRetired: rows.close() failed")
+	var inviteEventID string
 	for rows.Next() {
-		var inviteEventID string
 		if err = rows.Scan(&inviteEventID); err != nil {
 			return
 		}
@@ -147,9 +147,9 @@ func (s *inviteStatements) SelectInviteActiveForUserInRoom(
 	defer internal.CloseAndLogIfError(ctx, rows, "selectInviteActiveForUserInRoom: rows.close() failed")
 	var result []types.EventStateKeyNID
 	var eventIDs []string
+	var eventID string
+	var senderUserNID int64
 	for rows.Next() {
-		var eventID string
-		var senderUserNID int64
 		if err := rows.Scan(&eventID, &senderUserNID); err != nil {
 			return nil, nil, err
 		}

roomserver/storage/sqlite3/membership_table.go

@@ -136,12 +136,12 @@ type membershipStatements struct {
 	selectServerInRoomStmt                          *sql.Stmt
 }
 
-func createMembershipTable(db *sql.DB) error {
+func CreateMembershipTable(db *sql.DB) error {
 	_, err := db.Exec(membershipSchema)
 	return err
 }
 
-func prepareMembershipTable(db *sql.DB) (tables.Membership, error) {
+func PrepareMembershipTable(db *sql.DB) (tables.Membership, error) {
 	s := &membershipStatements{
 		db: db,
 	}
@@ -212,8 +212,8 @@ func (s *membershipStatements) SelectMembershipsFromRoom(
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "selectMembershipsFromRoom: rows.close() failed")
 
+	var eNID types.EventNID
 	for rows.Next() {
-		var eNID types.EventNID
 		if err = rows.Scan(&eNID); err != nil {
 			return
 		}
@@ -239,8 +239,8 @@ func (s *membershipStatements) SelectMembershipsFromRoomAndMembership(
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "selectMembershipsFromRoomAndMembership: rows.close() failed")
 
+	var eNID types.EventNID
 	for rows.Next() {
-		var eNID types.EventNID
 		if err = rows.Scan(&eNID); err != nil {
 			return
 		}
@@ -275,8 +275,8 @@ func (s *membershipStatements) SelectRoomsWithMembership(
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "SelectRoomsWithMembership: rows.close() failed")
 	var roomNIDs []types.RoomNID
+	var roomNID types.RoomNID
 	for rows.Next() {
-		var roomNID types.RoomNID
 		if err := rows.Scan(&roomNID); err != nil {
 			return nil, err
 		}
@@ -307,9 +307,9 @@ func (s *membershipStatements) SelectJoinedUsersSetForRooms(ctx context.Context,
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "selectJoinedUsersSetForRooms: rows.close() failed")
 	result := make(map[types.EventStateKeyNID]int)
+	var userID types.EventStateKeyNID
+	var count int
 	for rows.Next() {
-		var userID types.EventStateKeyNID
-		var count int
 		if err := rows.Scan(&userID, &count); err != nil {
 			return nil, err
 		}
@@ -326,12 +326,12 @@ func (s *membershipStatements) SelectKnownUsers(ctx context.Context, txn *sql.Tx
 	}
 	result := []string{}
 	defer internal.CloseAndLogIfError(ctx, rows, "SelectKnownUsers: rows.close() failed")
+	var resUserID string
 	for rows.Next() {
-		var userID string
+		if err := rows.Scan(&resUserID); err != nil {
-		if err := rows.Scan(&userID); err != nil {
 			return nil, err
 		}
-		result = append(result, userID)
+		result = append(result, resUserID)
 	}
 	return result, rows.Err()
 }

roomserver/storage/sqlite3/previous_events_table.go

@@ -70,12 +70,12 @@ type previousEventStatements struct {
 	selectPreviousEventExistsStmt *sql.Stmt
 }
 
-func createPrevEventsTable(db *sql.DB) error {
+func CreatePrevEventsTable(db *sql.DB) error {
 	_, err := db.Exec(previousEventSchema)
 	return err
 }
 
-func preparePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
+func PreparePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
 	s := &previousEventStatements{
 		db: db,
 	}

roomserver/storage/sqlite3/published_table.go

@@ -49,12 +49,12 @@ type publishedStatements struct {
 	selectPublishedStmt    *sql.Stmt
 }
 
-func createPublishedTable(db *sql.DB) error {
+func CreatePublishedTable(db *sql.DB) error {
 	_, err := db.Exec(publishedSchema)
 	return err
 }
 
-func preparePublishedTable(db *sql.DB) (tables.Published, error) {
+func PreparePublishedTable(db *sql.DB) (tables.Published, error) {
 	s := &publishedStatements{
 		db: db,
 	}
@@ -96,8 +96,8 @@ func (s *publishedStatements) SelectAllPublishedRooms(
 	defer internal.CloseAndLogIfError(ctx, rows, "selectAllPublishedStmt: rows.close() failed")
 
 	var roomIDs []string
+	var roomID string
 	for rows.Next() {
-		var roomID string
 		if err = rows.Scan(&roomID); err != nil {
 			return nil, err
 		}

roomserver/storage/sqlite3/redactions_table.go

@@ -48,7 +48,7 @@ const selectRedactionInfoByEventBeingRedactedSQL = "" +
 	" WHERE redacts_event_id = $1"
 
 const markRedactionValidatedSQL = "" +
-	" UPDATE roomserver_redactions SET validated = $2 WHERE redaction_event_id = $1"
+	" UPDATE roomserver_redactions SET validated = $1 WHERE redaction_event_id = $2"
 
 type redactionStatements struct {
 	db                                          *sql.DB
@@ -58,12 +58,12 @@ type redactionStatements struct {
 	markRedactionValidatedStmt                  *sql.Stmt
 }
 
-func createRedactionsTable(db *sql.DB) error {
+func CreateRedactionsTable(db *sql.DB) error {
 	_, err := db.Exec(redactionsSchema)
 	return err
 }
 
-func prepareRedactionsTable(db *sql.DB) (tables.Redactions, error) {
+func PrepareRedactionsTable(db *sql.DB) (tables.Redactions, error) {
 	s := &redactionStatements{
 		db: db,
 	}
@@ -118,6 +118,6 @@ func (s *redactionStatements) MarkRedactionValidated(
 	ctx context.Context, txn *sql.Tx, redactionEventID string, validated bool,
 ) error {
 	stmt := sqlutil.TxStmt(txn, s.markRedactionValidatedStmt)
-	_, err := stmt.ExecContext(ctx, redactionEventID, validated)
+	_, err := stmt.ExecContext(ctx, validated, redactionEventID)
 	return err
 }

roomserver/storage/sqlite3/storage.go

@@ -98,22 +98,22 @@ func (d *Database) create(db *sql.DB) error {
 	if err := createStateSnapshotTable(db); err != nil {
 		return err
 	}
-	if err := createPrevEventsTable(db); err != nil {
+	if err := CreatePrevEventsTable(db); err != nil {
 		return err
 	}
 	if err := createRoomAliasesTable(db); err != nil {
 		return err
 	}
-	if err := createInvitesTable(db); err != nil {
+	if err := CreateInvitesTable(db); err != nil {
 		return err
 	}
-	if err := createMembershipTable(db); err != nil {
+	if err := CreateMembershipTable(db); err != nil {
 		return err
 	}
-	if err := createPublishedTable(db); err != nil {
+	if err := CreatePublishedTable(db); err != nil {
 		return err
 	}
-	if err := createRedactionsTable(db); err != nil {
+	if err := CreateRedactionsTable(db); err != nil {
 		return err
 	}
 
@@ -149,7 +149,7 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
 	if err != nil {
 		return err
 	}
-	prevEvents, err := preparePrevEventsTable(db)
+	prevEvents, err := PreparePrevEventsTable(db)
 	if err != nil {
 		return err
 	}
@@ -157,19 +157,19 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
 	if err != nil {
 		return err
 	}
-	invites, err := prepareInvitesTable(db)
+	invites, err := PrepareInvitesTable(db)
 	if err != nil {
 		return err
 	}
-	membership, err := prepareMembershipTable(db)
+	membership, err := PrepareMembershipTable(db)
 	if err != nil {
 		return err
 	}
-	published, err := preparePublishedTable(db)
+	published, err := PreparePublishedTable(db)
 	if err != nil {
 		return err
 	}
-	redactions, err := prepareRedactionsTable(db)
+	redactions, err := PrepareRedactionsTable(db)
 	if err != nil {
 		return err
 	}

roomserver/storage/tables/invite_table_test.go

@@ -0,0 +1,92 @@
+package tables_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/roomserver/storage/postgres"
+	"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
+	"github.com/matrix-org/dendrite/roomserver/storage/tables"
+	"github.com/matrix-org/dendrite/roomserver/types"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/util"
+	"github.com/stretchr/testify/assert"
+)
+
+func mustCreateInviteTable(t *testing.T, dbType test.DBType) (tables.Invites, func()) {
+	t.Helper()
+	connStr, close := test.PrepareDBConnectionString(t, dbType)
+	db, err := sqlutil.Open(&config.DatabaseOptions{
+		ConnectionString: config.DataSource(connStr),
+	}, sqlutil.NewExclusiveWriter())
+	assert.NoError(t, err)
+	var tab tables.Invites
+	switch dbType {
+	case test.DBTypePostgres:
+		err = postgres.CreateInvitesTable(db)
+		assert.NoError(t, err)
+		tab, err = postgres.PrepareInvitesTable(db)
+	case test.DBTypeSQLite:
+		err = sqlite3.CreateInvitesTable(db)
+		assert.NoError(t, err)
+		tab, err = sqlite3.PrepareInvitesTable(db)
+	}
+	assert.NoError(t, err)
+
+	return tab, close
+}
+
+func TestInviteTable(t *testing.T) {
+	ctx := context.Background()
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		tab, close := mustCreateInviteTable(t, dbType)
+		defer close()
+		eventID1 := util.RandomString(16)
+		roomNID := types.RoomNID(1)
+		targetUserNID, senderUserNID := types.EventStateKeyNID(1), types.EventStateKeyNID(2)
+		newInvite, err := tab.InsertInviteEvent(ctx, nil, eventID1, roomNID, targetUserNID, senderUserNID, []byte(""))
+		assert.NoError(t, err)
+		assert.True(t, newInvite)
+
+		// Try adding the same invite again
+		newInvite, err = tab.InsertInviteEvent(ctx, nil, eventID1, roomNID, targetUserNID, senderUserNID, []byte(""))
+		assert.NoError(t, err)
+		assert.False(t, newInvite)
+
+		// Add another invite for this room
+		eventID2 := util.RandomString(16)
+		newInvite, err = tab.InsertInviteEvent(ctx, nil, eventID2, roomNID, targetUserNID, senderUserNID, []byte(""))
+		assert.NoError(t, err)
+		assert.True(t, newInvite)
+
+		// Add another invite for a different user
+		eventID := util.RandomString(16)
+		newInvite, err = tab.InsertInviteEvent(ctx, nil, eventID, types.RoomNID(3), targetUserNID, senderUserNID, []byte(""))
+		assert.NoError(t, err)
+		assert.True(t, newInvite)
+
+		stateKeyNIDs, eventIDs, err := tab.SelectInviteActiveForUserInRoom(ctx, nil, targetUserNID, roomNID)
+		assert.NoError(t, err)
+		assert.Equal(t, []string{eventID1, eventID2}, eventIDs)
+		assert.Equal(t, []types.EventStateKeyNID{2, 2}, stateKeyNIDs)
+
+		// retire the invite
+		retiredEventIDs, err := tab.UpdateInviteRetired(ctx, nil, roomNID, targetUserNID)
+		assert.NoError(t, err)
+		assert.Equal(t, []string{eventID1, eventID2}, retiredEventIDs)
+
+		// This should now be empty
+		stateKeyNIDs, eventIDs, err = tab.SelectInviteActiveForUserInRoom(ctx, nil, targetUserNID, roomNID)
+		assert.NoError(t, err)
+		assert.Empty(t, eventIDs)
+		assert.Empty(t, stateKeyNIDs)
+
+		// Non-existent targetUserNID
+		stateKeyNIDs, eventIDs, err = tab.SelectInviteActiveForUserInRoom(ctx, nil, types.EventStateKeyNID(10), roomNID)
+		assert.NoError(t, err)
+		assert.Empty(t, stateKeyNIDs)
+		assert.Empty(t, eventIDs)
+	})
+}

roomserver/storage/tables/membership_table_test.go

@@ -0,0 +1,130 @@
+package tables_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/roomserver/storage/postgres"
+	"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
+	"github.com/matrix-org/dendrite/roomserver/storage/tables"
+	"github.com/matrix-org/dendrite/roomserver/types"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/stretchr/testify/assert"
+)
+
+func mustCreateMembershipTable(t *testing.T, dbType test.DBType) (tab tables.Membership, stateKeyTab tables.EventStateKeys, close func()) {
+	t.Helper()
+	connStr, close := test.PrepareDBConnectionString(t, dbType)
+	db, err := sqlutil.Open(&config.DatabaseOptions{
+		ConnectionString: config.DataSource(connStr),
+	}, sqlutil.NewExclusiveWriter())
+	assert.NoError(t, err)
+	switch dbType {
+	case test.DBTypePostgres:
+		err = postgres.CreateEventStateKeysTable(db)
+		assert.NoError(t, err)
+		err = postgres.CreateMembershipTable(db)
+		assert.NoError(t, err)
+		tab, err = postgres.PrepareMembershipTable(db)
+		assert.NoError(t, err)
+		stateKeyTab, err = postgres.PrepareEventStateKeysTable(db)
+	case test.DBTypeSQLite:
+		err = sqlite3.CreateEventStateKeysTable(db)
+		assert.NoError(t, err)
+		err = sqlite3.CreateMembershipTable(db)
+		assert.NoError(t, err)
+		tab, err = sqlite3.PrepareMembershipTable(db)
+		assert.NoError(t, err)
+		stateKeyTab, err = sqlite3.PrepareEventStateKeysTable(db)
+	}
+	assert.NoError(t, err)
+
+	return tab, stateKeyTab, close
+}
+
+func TestMembershipTable(t *testing.T) {
+	ctx := context.Background()
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		tab, stateKeyTab, close := mustCreateMembershipTable(t, dbType)
+		defer close()
+		_ = close
+
+		userNIDs := make([]types.EventStateKeyNID, 0, 10)
+		for i := 0; i < 10; i++ {
+			stateKeyNID, err := stateKeyTab.InsertEventStateKeyNID(ctx, nil, fmt.Sprintf("@dummy%d:localhost", i))
+			assert.NoError(t, err)
+			userNIDs = append(userNIDs, stateKeyNID)
+			// This inserts a left user to the room
+			err = tab.InsertMembership(ctx, nil, 1, stateKeyNID, true)
+			assert.NoError(t, err)
+		}
+
+		// ... so this should be false
+		inRoom, err := tab.SelectLocalServerInRoom(ctx, nil, 1)
+		assert.NoError(t, err)
+		assert.False(t, inRoom)
+
+		changed, err := tab.UpdateMembership(ctx, nil, 1, userNIDs[0], userNIDs[0], tables.MembershipStateJoin, 1, false)
+		assert.NoError(t, err)
+		assert.True(t, changed)
+
+		// ... should now be true
+		inRoom, err = tab.SelectLocalServerInRoom(ctx, nil, 1)
+		assert.NoError(t, err)
+		assert.True(t, inRoom)
+
+		userJoinedToRooms, err := tab.SelectJoinedUsersSetForRooms(ctx, nil, []types.RoomNID{1}, userNIDs)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(userJoinedToRooms))
+
+		// Get all left/banned users
+		eventNIDs, err := tab.SelectMembershipsFromRoomAndMembership(ctx, nil, 1, tables.MembershipStateLeaveOrBan, true)
+		assert.NoError(t, err)
+		assert.Equal(t, 9, len(eventNIDs))
+
+		_, membershipState, forgotten, err := tab.SelectMembershipFromRoomAndTarget(ctx, nil, 1, userNIDs[5])
+		assert.NoError(t, err)
+		assert.False(t, forgotten)
+		assert.Equal(t, tables.MembershipStateLeaveOrBan, membershipState)
+
+		// Get all members, regardless of state
+		members, err := tab.SelectMembershipsFromRoom(ctx, nil, 1, true)
+		assert.NoError(t, err)
+		assert.Equal(t, 10, len(members))
+
+		// Get correct user
+		roomNIDs, err := tab.SelectRoomsWithMembership(ctx, nil, userNIDs[1], tables.MembershipStateLeaveOrBan)
+		assert.NoError(t, err)
+		assert.Equal(t, []types.RoomNID{1}, roomNIDs)
+
+		// User is not joined to room
+		roomNIDs, err = tab.SelectRoomsWithMembership(ctx, nil, userNIDs[5], tables.MembershipStateJoin)
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(roomNIDs))
+
+		// Forget room
+		err = tab.UpdateForgetMembership(ctx, nil, 1, userNIDs[0], true)
+		assert.NoError(t, err)
+
+		// should now return true
+		_, _, forgotten, err = tab.SelectMembershipFromRoomAndTarget(ctx, nil, 1, userNIDs[0])
+		assert.NoError(t, err)
+		assert.True(t, forgotten)
+
+		serverInRoom, err := tab.SelectServerInRoom(ctx, nil, 1, "localhost")
+		assert.NoError(t, err)
+		assert.True(t, serverInRoom)
+
+		serverInRoom, err = tab.SelectServerInRoom(ctx, nil, 1, "notJoined")
+		assert.NoError(t, err)
+		assert.False(t, serverInRoom)
+
+		// get all users we know about; should be only one user, since no other user joined the room
+		knownUsers, err := tab.SelectKnownUsers(ctx, nil, userNIDs[0], "localhost", 2)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(knownUsers))
+	})
+}

roomserver/storage/tables/previous_events_table_test.go

@@ -0,0 +1,61 @@
+package tables_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/roomserver/storage/postgres"
+	"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
+	"github.com/matrix-org/dendrite/roomserver/storage/tables"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/util"
+	"github.com/stretchr/testify/assert"
+)
+
+func mustCreatePreviousEventsTable(t *testing.T, dbType test.DBType) (tab tables.PreviousEvents, close func()) {
+	t.Helper()
+	connStr, close := test.PrepareDBConnectionString(t, dbType)
+	db, err := sqlutil.Open(&config.DatabaseOptions{
+		ConnectionString: config.DataSource(connStr),
+	}, sqlutil.NewExclusiveWriter())
+	assert.NoError(t, err)
+	switch dbType {
+	case test.DBTypePostgres:
+		err = postgres.CreatePrevEventsTable(db)
+		assert.NoError(t, err)
+		tab, err = postgres.PreparePrevEventsTable(db)
+	case test.DBTypeSQLite:
+		err = sqlite3.CreatePrevEventsTable(db)
+		assert.NoError(t, err)
+		tab, err = sqlite3.PreparePrevEventsTable(db)
+	}
+	assert.NoError(t, err)
+
+	return tab, close
+}
+
+func TestPreviousEventsTable(t *testing.T) {
+	ctx := context.Background()
+	alice := test.NewUser()
+	room := test.NewRoom(t, alice)
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		tab, close := mustCreatePreviousEventsTable(t, dbType)
+		defer close()
+
+		for _, x := range room.Events() {
+			for _, prevEvent := range x.PrevEvents() {
+				err := tab.InsertPreviousEvent(ctx, nil, prevEvent.EventID, prevEvent.EventSHA256, 1)
+				assert.NoError(t, err)
+
+				err = tab.SelectPreviousEventExists(ctx, nil, prevEvent.EventID, prevEvent.EventSHA256)
+				assert.NoError(t, err)
+			}
+		}
+
+		// RandomString with a correct EventSHA256 should fail and return sql.ErrNoRows
+		err := tab.SelectPreviousEventExists(ctx, nil, util.RandomString(16), room.Events()[0].EventReference().EventSHA256)
+		assert.Error(t, err)
+	})
+}

roomserver/storage/tables/published_table_test.go

@@ -0,0 +1,79 @@
+package tables_test
+
+import (
+	"context"
+	"sort"
+	"testing"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/roomserver/storage/postgres"
+	"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
+	"github.com/matrix-org/dendrite/roomserver/storage/tables"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/stretchr/testify/assert"
+)
+
+func mustCreatePublishedTable(t *testing.T, dbType test.DBType) (tab tables.Published, close func()) {
+	t.Helper()
+	connStr, close := test.PrepareDBConnectionString(t, dbType)
+	db, err := sqlutil.Open(&config.DatabaseOptions{
+		ConnectionString: config.DataSource(connStr),
+	}, sqlutil.NewExclusiveWriter())
+	assert.NoError(t, err)
+	switch dbType {
+	case test.DBTypePostgres:
+		err = postgres.CreatePublishedTable(db)
+		assert.NoError(t, err)
+		tab, err = postgres.PreparePublishedTable(db)
+	case test.DBTypeSQLite:
+		err = sqlite3.CreatePublishedTable(db)
+		assert.NoError(t, err)
+		tab, err = sqlite3.PreparePublishedTable(db)
+	}
+	assert.NoError(t, err)
+
+	return tab, close
+}
+
+func TestPublishedTable(t *testing.T) {
+	ctx := context.Background()
+	alice := test.NewUser()
+
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		tab, close := mustCreatePublishedTable(t, dbType)
+		defer close()
+
+		// Publish some rooms
+		publishedRooms := []string{}
+		for i := 0; i < 10; i++ {
+			room := test.NewRoom(t, alice)
+			published := i%2 == 0
+			err := tab.UpsertRoomPublished(ctx, nil, room.ID, published)
+			assert.NoError(t, err)
+			if published {
+				publishedRooms = append(publishedRooms, room.ID)
+			}
+			publishedRes, err := tab.SelectPublishedFromRoomID(ctx, nil, room.ID)
+			assert.NoError(t, err)
+			assert.Equal(t, published, publishedRes)
+		}
+		sort.Strings(publishedRooms)
+
+		// check that we get the expected published rooms
+		roomIDs, err := tab.SelectAllPublishedRooms(ctx, nil, true)
+		assert.NoError(t, err)
+		assert.Equal(t, publishedRooms, roomIDs)
+
+		// test an actual upsert
+		room := test.NewRoom(t, alice)
+		err = tab.UpsertRoomPublished(ctx, nil, room.ID, true)
+		assert.NoError(t, err)
+		err = tab.UpsertRoomPublished(ctx, nil, room.ID, false)
+		assert.NoError(t, err)
+		// should now be false, due to the upsert
+		publishedRes, err := tab.SelectPublishedFromRoomID(ctx, nil, room.ID)
+		assert.NoError(t, err)
+		assert.False(t, publishedRes)
+	})
+}

roomserver/storage/tables/redactions_table_test.go

@@ -0,0 +1,89 @@
+package tables_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/roomserver/storage/postgres"
+	"github.com/matrix-org/dendrite/roomserver/storage/sqlite3"
+	"github.com/matrix-org/dendrite/roomserver/storage/tables"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/util"
+	"github.com/stretchr/testify/assert"
+)
+
+func mustCreateRedactionsTable(t *testing.T, dbType test.DBType) (tab tables.Redactions, close func()) {
+	t.Helper()
+	connStr, close := test.PrepareDBConnectionString(t, dbType)
+	db, err := sqlutil.Open(&config.DatabaseOptions{
+		ConnectionString: config.DataSource(connStr),
+	}, sqlutil.NewExclusiveWriter())
+	assert.NoError(t, err)
+	switch dbType {
+	case test.DBTypePostgres:
+		err = postgres.CreateRedactionsTable(db)
+		assert.NoError(t, err)
+		tab, err = postgres.PrepareRedactionsTable(db)
+	case test.DBTypeSQLite:
+		err = sqlite3.CreateRedactionsTable(db)
+		assert.NoError(t, err)
+		tab, err = sqlite3.PrepareRedactionsTable(db)
+	}
+	assert.NoError(t, err)
+
+	return tab, close
+}
+
+func TestRedactionsTable(t *testing.T) {
+	ctx := context.Background()
+
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		tab, close := mustCreateRedactionsTable(t, dbType)
+		defer close()
+
+		// insert and verify some redactions
+		for i := 0; i < 10; i++ {
+			redactionEventID, redactsEventID := util.RandomString(16), util.RandomString(16)
+			wantRedactionInfo := tables.RedactionInfo{
+				Validated:        false,
+				RedactsEventID:   redactsEventID,
+				RedactionEventID: redactionEventID,
+			}
+			err := tab.InsertRedaction(ctx, nil, wantRedactionInfo)
+			assert.NoError(t, err)
+
+			// verify the redactions are inserted as expected
+			redactionInfo, err := tab.SelectRedactionInfoByRedactionEventID(ctx, nil, redactionEventID)
+			assert.NoError(t, err)
+			assert.Equal(t, &wantRedactionInfo, redactionInfo)
+
+			redactionInfo, err = tab.SelectRedactionInfoByEventBeingRedacted(ctx, nil, redactsEventID)
+			assert.NoError(t, err)
+			assert.Equal(t, &wantRedactionInfo, redactionInfo)
+
+			// redact event
+			err = tab.MarkRedactionValidated(ctx, nil, redactionEventID, true)
+			assert.NoError(t, err)
+
+			wantRedactionInfo.Validated = true
+			redactionInfo, err = tab.SelectRedactionInfoByRedactionEventID(ctx, nil, redactionEventID)
+			assert.NoError(t, err)
+			assert.Equal(t, &wantRedactionInfo, redactionInfo)
+		}
+
+		// Should not fail, it just updates 0 rows
+		err := tab.MarkRedactionValidated(ctx, nil, "iDontExist", true)
+		assert.NoError(t, err)
+
+		// Should also not fail, but return a nil redactionInfo
+		redactionInfo, err := tab.SelectRedactionInfoByRedactionEventID(ctx, nil, "iDontExist")
+		assert.NoError(t, err)
+		assert.Nil(t, redactionInfo)
+
+		redactionInfo, err = tab.SelectRedactionInfoByEventBeingRedacted(ctx, nil, "iDontExist")
+		assert.NoError(t, err)
+		assert.Nil(t, redactionInfo)
+	})
+}

syncapi/routing/context.go

@@ -73,6 +73,12 @@ func Context(
 		logrus.WithError(err).Error("unable to query membership")
 		return jsonerror.InternalServerError()
 	}
+	if !membershipRes.RoomExists {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("room does not exist"),
+		}
+	}
 
 	stateFilter := gomatrixserverlib.StateFilter{
 		Limit:                   100,

syncapi/routing/messages.go

@@ -68,10 +68,16 @@ func OnIncomingMessagesRequest(
 	var err error
 
 	// check if the user has already forgotten about this room
-	isForgotten, err := checkIsRoomForgotten(req.Context(), roomID, device.UserID, rsAPI)
+	isForgotten, roomExists, err := checkIsRoomForgotten(req.Context(), roomID, device.UserID, rsAPI)
 	if err != nil {
 		return jsonerror.InternalServerError()
 	}
+	if !roomExists {
+		return util.JSONResponse{
+			Code: http.StatusForbidden,
+			JSON: jsonerror.Forbidden("room does not exist"),
+		}
+	}
 
 	if isForgotten {
 		return util.JSONResponse{
@@ -244,17 +250,17 @@ func OnIncomingMessagesRequest(
 	}
 }
 
-func checkIsRoomForgotten(ctx context.Context, roomID, userID string, rsAPI api.SyncRoomserverAPI) (bool, error) {
+func checkIsRoomForgotten(ctx context.Context, roomID, userID string, rsAPI api.SyncRoomserverAPI) (forgotten bool, exists bool, err error) {
 	req := api.QueryMembershipForUserRequest{
 		RoomID: roomID,
 		UserID: userID,
 	}
 	resp := api.QueryMembershipForUserResponse{}
 	if err := rsAPI.QueryMembershipForUser(ctx, &req, &resp); err != nil {
-		return false, err
+		return false, false, err
 	}
 
-	return resp.IsRoomForgotten, nil
+	return resp.IsRoomForgotten, resp.RoomExists, nil
 }
 
 // retrieveEvents retrieves events from the local database for a request on

syncapi/storage/postgres/output_room_events_table.go

@@ -69,6 +69,11 @@ CREATE TABLE IF NOT EXISTS syncapi_output_room_events (
   -- were emitted.
   exclude_from_sync BOOL DEFAULT FALSE
 );
+
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_type_idx ON syncapi_output_room_events (type);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_sender_idx ON syncapi_output_room_events (sender);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_room_id_idx ON syncapi_output_room_events (room_id);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_exclude_from_sync_idx ON syncapi_output_room_events (exclude_from_sync);
 `
 
 const insertEventSQL = "" +

syncapi/storage/sqlite3/output_room_events_table.go

@@ -49,6 +49,11 @@ CREATE TABLE IF NOT EXISTS syncapi_output_room_events (
   transaction_id TEXT,
   exclude_from_sync BOOL NOT NULL DEFAULT FALSE
 );
+
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_type_idx ON syncapi_output_room_events (type);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_sender_idx ON syncapi_output_room_events (sender);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_room_id_idx ON syncapi_output_room_events (room_id);
+CREATE INDEX IF NOT EXISTS syncapi_output_room_events_exclude_from_sync_idx ON syncapi_output_room_events (exclude_from_sync);
 `
 
 const insertEventSQL = "" +

syncapi/syncapi_test.go

@@ -11,10 +11,12 @@ import (
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	rsapi "github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/base"
 	"github.com/matrix-org/dendrite/setup/jetstream"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/dendrite/test"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
+	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/nats-io/nats.go"
 )
 
@@ -39,6 +41,14 @@ func (s *syncRoomserverAPI) QueryLatestEventsAndState(ctx context.Context, req *
 	return nil // TODO: return state
 }
 
+func (s *syncRoomserverAPI) QuerySharedUsers(ctx context.Context, req *rsapi.QuerySharedUsersRequest, res *rsapi.QuerySharedUsersResponse) error {
+	res.UserIDsToCount = make(map[string]int)
+	return nil
+}
+func (s *syncRoomserverAPI) QueryBulkStateContent(ctx context.Context, req *rsapi.QueryBulkStateContentRequest, res *rsapi.QueryBulkStateContentResponse) error {
+	return nil
+}
+
 type syncUserAPI struct {
 	userapi.SyncUserAPI
 	accounts []userapi.Device
@@ -60,16 +70,22 @@ func (s *syncUserAPI) PerformLastSeenUpdate(ctx context.Context, req *userapi.Pe
 }
 
 type syncKeyAPI struct {
-	keyapi.KeyInternalAPI
+	keyapi.SyncKeyAPI
 }
 
-func TestSyncAPI(t *testing.T) {
+func (s *syncKeyAPI) QueryKeyChanges(ctx context.Context, req *keyapi.QueryKeyChangesRequest, res *keyapi.QueryKeyChangesResponse) {
+}
+func (s *syncKeyAPI) QueryOneTimeKeys(ctx context.Context, req *keyapi.QueryOneTimeKeysRequest, res *keyapi.QueryOneTimeKeysResponse) {
+
+}
+
+func TestSyncAPIAccessTokens(t *testing.T) {
 	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
-		testSync(t, dbType)
+		testSyncAccessTokens(t, dbType)
 	})
 }
 
-func testSync(t *testing.T, dbType test.DBType) {
+func testSyncAccessTokens(t *testing.T, dbType test.DBType) {
 	user := test.NewUser()
 	room := test.NewRoom(t, user)
 	alice := userapi.Device{
@@ -85,20 +101,7 @@ func testSync(t *testing.T, dbType test.DBType) {
 
 	jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
 	defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
-	var msgs []*nats.Msg
+	msgs := toNATSMsgs(t, base, room.Events())
-	for _, ev := range room.Events() {
-		var addsStateIDs []string
-		if ev.StateKey() != nil {
-			addsStateIDs = append(addsStateIDs, ev.EventID())
-		}
-		msgs = append(msgs, test.NewOutputEventMsg(t, base, room.ID, api.OutputEvent{
-			Type: rsapi.OutputTypeNewRoomEvent,
-			NewRoomEvent: &rsapi.OutputNewRoomEvent{
-				Event:             ev,
-				AddsStateEventIDs: addsStateIDs,
-			},
-		}))
-	}
 	AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, &syncKeyAPI{})
 	test.MustPublishMsgs(t, jsctx, msgs...)
 
@@ -160,3 +163,112 @@ func testSync(t *testing.T, dbType test.DBType) {
 		}
 	}
 }
+
+// Tests what happens when we create a room and then /sync before all events from /createRoom have
+// been sent to the syncapi
+func TestSyncAPICreateRoomSyncEarly(t *testing.T) {
+	test.WithAllDatabases(t, func(t *testing.T, dbType test.DBType) {
+		testSyncAPICreateRoomSyncEarly(t, dbType)
+	})
+}
+
+func testSyncAPICreateRoomSyncEarly(t *testing.T, dbType test.DBType) {
+	user := test.NewUser()
+	room := test.NewRoom(t, user)
+	alice := userapi.Device{
+		ID:          "ALICEID",
+		UserID:      user.ID,
+		AccessToken: "ALICE_BEARER_TOKEN",
+		DisplayName: "Alice",
+		AccountType: userapi.AccountTypeUser,
+	}
+
+	base, close := test.CreateBaseDendrite(t, dbType)
+	defer close()
+
+	jsctx, _ := base.NATS.Prepare(base.ProcessContext, &base.Cfg.Global.JetStream)
+	defer jetstream.DeleteAllStreams(jsctx, &base.Cfg.Global.JetStream)
+	// order is:
+	// m.room.create
+	// m.room.member
+	// m.room.power_levels
+	// m.room.join_rules
+	// m.room.history_visibility
+	msgs := toNATSMsgs(t, base, room.Events())
+	sinceTokens := make([]string, len(msgs))
+	AddPublicRoutes(base, &syncUserAPI{accounts: []userapi.Device{alice}}, &syncRoomserverAPI{rooms: []*test.Room{room}}, &syncKeyAPI{})
+	for i, msg := range msgs {
+		test.MustPublishMsgs(t, jsctx, msg)
+		time.Sleep(50 * time.Millisecond)
+		w := httptest.NewRecorder()
+		base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+			"access_token": alice.AccessToken,
+			"timeout":      "0",
+		})))
+		if w.Code != 200 {
+			t.Errorf("got HTTP %d want 200", w.Code)
+			continue
+		}
+		var res types.Response
+		if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
+			t.Errorf("failed to decode response body: %s", err)
+		}
+		sinceTokens[i] = res.NextBatch.String()
+		if i == 0 { // create event does not produce a room section
+			if len(res.Rooms.Join) != 0 {
+				t.Fatalf("i=%v got %d joined rooms, want 0", i, len(res.Rooms.Join))
+			}
+		} else { // we should have that room somewhere
+			if len(res.Rooms.Join) != 1 {
+				t.Fatalf("i=%v got %d joined rooms, want 1", i, len(res.Rooms.Join))
+			}
+		}
+	}
+
+	// sync with no token "" and with the penultimate token and this should neatly return room events in the timeline block
+	sinceTokens = append([]string{""}, sinceTokens[:len(sinceTokens)-1]...)
+
+	t.Logf("waited for events to be consumed; syncing with %v", sinceTokens)
+	for i, since := range sinceTokens {
+		w := httptest.NewRecorder()
+		base.PublicClientAPIMux.ServeHTTP(w, test.NewRequest(t, "GET", "/_matrix/client/v3/sync", test.WithQueryParams(map[string]string{
+			"access_token": alice.AccessToken,
+			"timeout":      "0",
+			"since":        since,
+		})))
+		if w.Code != 200 {
+			t.Errorf("since=%s got HTTP %d want 200", since, w.Code)
+		}
+		var res types.Response
+		if err := json.NewDecoder(w.Body).Decode(&res); err != nil {
+			t.Errorf("failed to decode response body: %s", err)
+		}
+		if len(res.Rooms.Join) != 1 {
+			t.Fatalf("since=%s got %d joined rooms, want 1", since, len(res.Rooms.Join))
+		}
+		t.Logf("since=%s res state:%+v res timeline:%+v", since, res.Rooms.Join[room.ID].State.Events, res.Rooms.Join[room.ID].Timeline.Events)
+		gotEventIDs := make([]string, len(res.Rooms.Join[room.ID].Timeline.Events))
+		for j, ev := range res.Rooms.Join[room.ID].Timeline.Events {
+			gotEventIDs[j] = ev.EventID
+		}
+		test.AssertEventIDsEqual(t, gotEventIDs, room.Events()[i:])
+	}
+}
+
+func toNATSMsgs(t *testing.T, base *base.BaseDendrite, input []*gomatrixserverlib.HeaderedEvent) []*nats.Msg {
+	result := make([]*nats.Msg, len(input))
+	for i, ev := range input {
+		var addsStateIDs []string
+		if ev.StateKey() != nil {
+			addsStateIDs = append(addsStateIDs, ev.EventID())
+		}
+		result[i] = test.NewOutputEventMsg(t, base, ev.RoomID(), api.OutputEvent{
+			Type: rsapi.OutputTypeNewRoomEvent,
+			NewRoomEvent: &rsapi.OutputNewRoomEvent{
+				Event:             ev,
+				AddsStateEventIDs: addsStateIDs,
+			},
+		})
+	}
+	return result
+}

test/event.go

@@ -64,7 +64,8 @@ func Reversed(in []*gomatrixserverlib.HeaderedEvent) []*gomatrixserverlib.Header
 func AssertEventIDsEqual(t *testing.T, gotEventIDs []string, wants []*gomatrixserverlib.HeaderedEvent) {
 	t.Helper()
 	if len(gotEventIDs) != len(wants) {
-		t.Fatalf("length mismatch: got %d events, want %d", len(gotEventIDs), len(wants))
+		t.Errorf("length mismatch: got %d events, want %d", len(gotEventIDs), len(wants))
+		return
 	}
 	for i := range wants {
 		w := wants[i].EventID()

userapi/util/phonehomestats.go

@@ -39,7 +39,7 @@ type phoneHomeStats struct {
 	cfg        *config.Dendrite
 	db         storage.Statistics
 	isMonolith bool
-	client     *gomatrixserverlib.Client
+	client     *http.Client
 }
 
 type timestampToRUUsage struct {
@@ -55,9 +55,10 @@ func StartPhoneHomeCollector(startTime time.Time, cfg *config.Dendrite, statsDB
 		cfg:        cfg,
 		db:         statsDB,
 		isMonolith: cfg.IsMonolith,
-		client: gomatrixserverlib.NewClient(
+		client: &http.Client{
-			gomatrixserverlib.WithTimeout(time.Second * 30),
+			Timeout:   time.Second * 30,
-		),
+			Transport: http.DefaultTransport,
+		},
 	}
 
 	// start initial run after 5min
@@ -151,7 +152,8 @@ func (p *phoneHomeStats) collect() {
 	}
 	request.Header.Set("User-Agent", "Dendrite/"+internal.VersionString())
 
-	if _, err = p.client.DoHTTPRequest(ctx, request); err != nil {
+	_, err = p.client.Do(request)
+	if err != nil {
 		logrus.WithError(err).Error("unable to send anonymous stats")
 		return
 	}