Initial Store & Forward Implementation (#2917)

This adds store & forward relays into dendrite for p2p.
A few things have changed:
- new relay api serves new http endpoints for s&f federation
- updated outbound federation queueing which will attempt to forward
using s&f if appropriate
- database entries to track s&f relays for other nodes

5 files changed, 703 insertions, 0 deletions

diff --git a/relayapi/routing/relaytxn.go b/relayapi/routing/relaytxn.go
new file mode 100644
index 00000000..1b11b0ec
--- /dev/null
+++ b/relayapi/routing/relaytxn.go
@@ -0,0 +1,74 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/relayapi/api"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+	"github.com/sirupsen/logrus"
+)
+
+type RelayTransactionResponse struct {
+	Transaction   gomatrixserverlib.Transaction `json:"transaction"`
+	EntryID       int64                         `json:"entry_id,omitempty"`
+	EntriesQueued bool                          `json:"entries_queued"`
+}
+
+// GetTransactionFromRelay implements /_matrix/federation/v1/relay_txn/{userID}
+// This endpoint can be extracted into a separate relay server service.
+func GetTransactionFromRelay(
+	httpReq *http.Request,
+	fedReq *gomatrixserverlib.FederationRequest,
+	relayAPI api.RelayInternalAPI,
+	userID gomatrixserverlib.UserID,
+) util.JSONResponse {
+	logrus.Infof("Handling relay_txn for %s", userID.Raw())
+
+	previousEntry := gomatrixserverlib.RelayEntry{}
+	if err := json.Unmarshal(fedReq.Content(), &previousEntry); err != nil {
+		return util.JSONResponse{
+			Code: http.StatusInternalServerError,
+			JSON: jsonerror.BadJSON("invalid json provided"),
+		}
+	}
+	if previousEntry.EntryID < 0 {
+		return util.JSONResponse{
+			Code: http.StatusInternalServerError,
+			JSON: jsonerror.BadJSON("Invalid entry id provided. Must be >= 0."),
+		}
+	}
+	logrus.Infof("Previous entry provided: %v", previousEntry.EntryID)
+
+	response, err := relayAPI.QueryTransactions(httpReq.Context(), userID, previousEntry)
+	if err != nil {
+		return util.JSONResponse{
+			Code: http.StatusInternalServerError,
+		}
+	}
+
+	return util.JSONResponse{
+		Code: http.StatusOK,
+		JSON: RelayTransactionResponse{
+			Transaction:   response.Transaction,
+			EntryID:       response.EntryID,
+			EntriesQueued: response.EntriesQueued,
+		},
+	}
+}
diff --git a/relayapi/routing/relaytxn_test.go b/relayapi/routing/relaytxn_test.go
new file mode 100644
index 00000000..a47fdb19
--- /dev/null
+++ b/relayapi/routing/relaytxn_test.go
@@ -0,0 +1,220 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing_test
+
+import (
+	"context"
+	"net/http"
+	"testing"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/relayapi/internal"
+	"github.com/matrix-org/dendrite/relayapi/routing"
+	"github.com/matrix-org/dendrite/relayapi/storage/shared"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/stretchr/testify/assert"
+)
+
+func createQuery(
+	userID gomatrixserverlib.UserID,
+	prevEntry gomatrixserverlib.RelayEntry,
+) gomatrixserverlib.FederationRequest {
+	var federationPathPrefixV1 = "/_matrix/federation/v1"
+	path := federationPathPrefixV1 + "/relay_txn/" + userID.Raw()
+	request := gomatrixserverlib.NewFederationRequest("GET", userID.Domain(), "relay", path)
+	request.SetContent(prevEntry)
+
+	return request
+}
+
+func TestGetEmptyDatabaseReturnsNothing(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	transaction := createTransaction()
+
+	_, err = db.StoreTransaction(context.Background(), transaction)
+	assert.NoError(t, err, "Failed to store transaction")
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	request := createQuery(*userID, gomatrixserverlib.RelayEntry{})
+	response := routing.GetTransactionFromRelay(httpReq, &request, relayAPI, *userID)
+	assert.Equal(t, http.StatusOK, response.Code)
+
+	jsonResponse := response.JSON.(routing.RelayTransactionResponse)
+	assert.Equal(t, false, jsonResponse.EntriesQueued)
+	assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Transaction)
+
+	count, err := db.GetTransactionCount(context.Background(), *userID)
+	assert.NoError(t, err)
+	assert.Zero(t, count)
+}
+
+func TestGetInvalidPrevEntryFails(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	transaction := createTransaction()
+
+	_, err = db.StoreTransaction(context.Background(), transaction)
+	assert.NoError(t, err, "Failed to store transaction")
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	request := createQuery(*userID, gomatrixserverlib.RelayEntry{EntryID: -1})
+	response := routing.GetTransactionFromRelay(httpReq, &request, relayAPI, *userID)
+	assert.Equal(t, http.StatusInternalServerError, response.Code)
+}
+
+func TestGetReturnsSavedTransaction(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	transaction := createTransaction()
+	receipt, err := db.StoreTransaction(context.Background(), transaction)
+	assert.NoError(t, err, "Failed to store transaction")
+
+	err = db.AssociateTransactionWithDestinations(
+		context.Background(),
+		map[gomatrixserverlib.UserID]struct{}{
+			*userID: {},
+		},
+		transaction.TransactionID,
+		receipt)
+	assert.NoError(t, err, "Failed to associate transaction with user")
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	request := createQuery(*userID, gomatrixserverlib.RelayEntry{})
+	response := routing.GetTransactionFromRelay(httpReq, &request, relayAPI, *userID)
+	assert.Equal(t, http.StatusOK, response.Code)
+
+	jsonResponse := response.JSON.(routing.RelayTransactionResponse)
+	assert.True(t, jsonResponse.EntriesQueued)
+	assert.Equal(t, transaction, jsonResponse.Transaction)
+
+	// And once more to clear the queue
+	request = createQuery(*userID, gomatrixserverlib.RelayEntry{EntryID: jsonResponse.EntryID})
+	response = routing.GetTransactionFromRelay(httpReq, &request, relayAPI, *userID)
+	assert.Equal(t, http.StatusOK, response.Code)
+
+	jsonResponse = response.JSON.(routing.RelayTransactionResponse)
+	assert.False(t, jsonResponse.EntriesQueued)
+	assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Transaction)
+
+	count, err := db.GetTransactionCount(context.Background(), *userID)
+	assert.NoError(t, err)
+	assert.Zero(t, count)
+}
+
+func TestGetReturnsMultipleSavedTransactions(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	transaction := createTransaction()
+	receipt, err := db.StoreTransaction(context.Background(), transaction)
+	assert.NoError(t, err, "Failed to store transaction")
+
+	err = db.AssociateTransactionWithDestinations(
+		context.Background(),
+		map[gomatrixserverlib.UserID]struct{}{
+			*userID: {},
+		},
+		transaction.TransactionID,
+		receipt)
+	assert.NoError(t, err, "Failed to associate transaction with user")
+
+	transaction2 := createTransaction()
+	receipt2, err := db.StoreTransaction(context.Background(), transaction2)
+	assert.NoError(t, err, "Failed to store transaction")
+
+	err = db.AssociateTransactionWithDestinations(
+		context.Background(),
+		map[gomatrixserverlib.UserID]struct{}{
+			*userID: {},
+		},
+		transaction2.TransactionID,
+		receipt2)
+	assert.NoError(t, err, "Failed to associate transaction with user")
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	request := createQuery(*userID, gomatrixserverlib.RelayEntry{})
+	response := routing.GetTransactionFromRelay(httpReq, &request, relayAPI, *userID)
+	assert.Equal(t, http.StatusOK, response.Code)
+
+	jsonResponse := response.JSON.(routing.RelayTransactionResponse)
+	assert.True(t, jsonResponse.EntriesQueued)
+	assert.Equal(t, transaction, jsonResponse.Transaction)
+
+	request = createQuery(*userID, gomatrixserverlib.RelayEntry{EntryID: jsonResponse.EntryID})
+	response = routing.GetTransactionFromRelay(httpReq, &request, relayAPI, *userID)
+	assert.Equal(t, http.StatusOK, response.Code)
+
+	jsonResponse = response.JSON.(routing.RelayTransactionResponse)
+	assert.True(t, jsonResponse.EntriesQueued)
+	assert.Equal(t, transaction2, jsonResponse.Transaction)
+
+	// And once more to clear the queue
+	request = createQuery(*userID, gomatrixserverlib.RelayEntry{EntryID: jsonResponse.EntryID})
+	response = routing.GetTransactionFromRelay(httpReq, &request, relayAPI, *userID)
+	assert.Equal(t, http.StatusOK, response.Code)
+
+	jsonResponse = response.JSON.(routing.RelayTransactionResponse)
+	assert.False(t, jsonResponse.EntriesQueued)
+	assert.Equal(t, gomatrixserverlib.Transaction{}, jsonResponse.Transaction)
+
+	count, err := db.GetTransactionCount(context.Background(), *userID)
+	assert.NoError(t, err)
+	assert.Zero(t, count)
+}
diff --git a/relayapi/routing/routing.go b/relayapi/routing/routing.go
new file mode 100644
index 00000000..6df0cdc5
--- /dev/null
+++ b/relayapi/routing/routing.go
@@ -0,0 +1,123 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/getsentry/sentry-go"
+	"github.com/gorilla/mux"
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/internal/httputil"
+	relayInternal "github.com/matrix-org/dendrite/relayapi/internal"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+)
+
+// Setup registers HTTP handlers with the given ServeMux.
+// The provided publicAPIMux MUST have `UseEncodedPath()` enabled or else routes will incorrectly
+// path unescape twice (once from the router, once from MakeRelayAPI). We need to have this enabled
+// so we can decode paths like foo/bar%2Fbaz as [foo, bar/baz] - by default it will decode to [foo, bar, baz]
+//
+// Due to Setup being used to call many other functions, a gocyclo nolint is
+// applied:
+// nolint: gocyclo
+func Setup(
+	fedMux *mux.Router,
+	cfg *config.FederationAPI,
+	relayAPI *relayInternal.RelayInternalAPI,
+	keys gomatrixserverlib.JSONVerifier,
+) {
+	v1fedmux := fedMux.PathPrefix("/v1").Subrouter()
+
+	v1fedmux.Handle("/send_relay/{txnID}/{userID}", MakeRelayAPI(
+		"send_relay_transaction", "", cfg.Matrix.IsLocalServerName, keys,
+		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
+			userID, err := gomatrixserverlib.NewUserID(vars["userID"], false)
+			if err != nil {
+				return util.JSONResponse{
+					Code: http.StatusBadRequest,
+					JSON: jsonerror.InvalidUsername("Username was invalid"),
+				}
+			}
+			return SendTransactionToRelay(
+				httpReq, request, relayAPI, gomatrixserverlib.TransactionID(vars["txnID"]),
+				*userID,
+			)
+		},
+	)).Methods(http.MethodPut, http.MethodOptions)
+
+	v1fedmux.Handle("/relay_txn/{userID}", MakeRelayAPI(
+		"get_relay_transaction", "", cfg.Matrix.IsLocalServerName, keys,
+		func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
+			userID, err := gomatrixserverlib.NewUserID(vars["userID"], false)
+			if err != nil {
+				return util.JSONResponse{
+					Code: http.StatusBadRequest,
+					JSON: jsonerror.InvalidUsername("Username was invalid"),
+				}
+			}
+			return GetTransactionFromRelay(httpReq, request, relayAPI, *userID)
+		},
+	)).Methods(http.MethodGet, http.MethodOptions)
+}
+
+// MakeRelayAPI makes an http.Handler that checks matrix relay authentication.
+func MakeRelayAPI(
+	metricsName string, serverName gomatrixserverlib.ServerName,
+	isLocalServerName func(gomatrixserverlib.ServerName) bool,
+	keyRing gomatrixserverlib.JSONVerifier,
+	f func(*http.Request, *gomatrixserverlib.FederationRequest, map[string]string) util.JSONResponse,
+) http.Handler {
+	h := func(req *http.Request) util.JSONResponse {
+		fedReq, errResp := gomatrixserverlib.VerifyHTTPRequest(
+			req, time.Now(), serverName, isLocalServerName, keyRing,
+		)
+		if fedReq == nil {
+			return errResp
+		}
+		// add the user to Sentry, if enabled
+		hub := sentry.GetHubFromContext(req.Context())
+		if hub != nil {
+			hub.Scope().SetTag("origin", string(fedReq.Origin()))
+			hub.Scope().SetTag("uri", fedReq.RequestURI())
+		}
+		defer func() {
+			if r := recover(); r != nil {
+				if hub != nil {
+					hub.CaptureException(fmt.Errorf("%s panicked", req.URL.Path))
+				}
+				// re-panic to return the 500
+				panic(r)
+			}
+		}()
+		vars, err := httputil.URLDecodeMapValues(mux.Vars(req))
+		if err != nil {
+			return util.MatrixErrorResponse(400, "M_UNRECOGNISED", "badly encoded query params")
+		}
+
+		jsonRes := f(req, fedReq, vars)
+		// do not log 4xx as errors as they are client fails, not server fails
+		if hub != nil && jsonRes.Code >= 500 {
+			hub.Scope().SetExtra("response", jsonRes)
+			hub.CaptureException(fmt.Errorf("%s returned HTTP %d", req.URL.Path, jsonRes.Code))
+		}
+		return jsonRes
+	}
+	return httputil.MakeExternalAPI(metricsName, h)
+}
diff --git a/relayapi/routing/sendrelay.go b/relayapi/routing/sendrelay.go
new file mode 100644
index 00000000..a7027f29
--- /dev/null
+++ b/relayapi/routing/sendrelay.go
@@ -0,0 +1,77 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/matrix-org/dendrite/clientapi/jsonerror"
+	"github.com/matrix-org/dendrite/relayapi/api"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+	"github.com/sirupsen/logrus"
+)
+
+// SendTransactionToRelay implements PUT /_matrix/federation/v1/relay_txn/{txnID}/{userID}
+// This endpoint can be extracted into a separate relay server service.
+func SendTransactionToRelay(
+	httpReq *http.Request,
+	fedReq *gomatrixserverlib.FederationRequest,
+	relayAPI api.RelayInternalAPI,
+	txnID gomatrixserverlib.TransactionID,
+	userID gomatrixserverlib.UserID,
+) util.JSONResponse {
+	var txnEvents struct {
+		PDUs []json.RawMessage       `json:"pdus"`
+		EDUs []gomatrixserverlib.EDU `json:"edus"`
+	}
+
+	if err := json.Unmarshal(fedReq.Content(), &txnEvents); err != nil {
+		logrus.Info("The request body could not be decoded into valid JSON." + err.Error())
+		return util.JSONResponse{
+			Code: http.StatusBadRequest,
+			JSON: jsonerror.NotJSON("The request body could not be decoded into valid JSON." + err.Error()),
+		}
+	}
+
+	// Transactions are limited in size; they can have at most 50 PDUs and 100 EDUs.
+	// https://matrix.org/docs/spec/server_server/latest#transactions
+	if len(txnEvents.PDUs) > 50 || len(txnEvents.EDUs) > 100 {
+		return util.JSONResponse{
+			Code: http.StatusBadRequest,
+			JSON: jsonerror.BadJSON("max 50 pdus / 100 edus"),
+		}
+	}
+
+	t := gomatrixserverlib.Transaction{}
+	t.PDUs = txnEvents.PDUs
+	t.EDUs = txnEvents.EDUs
+	t.Origin = fedReq.Origin()
+	t.TransactionID = txnID
+	t.Destination = userID.Domain()
+
+	util.GetLogger(httpReq.Context()).Warnf("Received transaction %q from %q containing %d PDUs, %d EDUs", txnID, fedReq.Origin(), len(t.PDUs), len(t.EDUs))
+
+	err := relayAPI.PerformStoreTransaction(httpReq.Context(), t, userID)
+	if err != nil {
+		return util.JSONResponse{
+			Code: http.StatusInternalServerError,
+			JSON: jsonerror.BadJSON("could not store the transaction for forwarding"),
+		}
+	}
+
+	return util.JSONResponse{Code: 200}
+}
diff --git a/relayapi/routing/sendrelay_test.go b/relayapi/routing/sendrelay_test.go
new file mode 100644
index 00000000..d9ed7500
--- /dev/null
+++ b/relayapi/routing/sendrelay_test.go
@@ -0,0 +1,209 @@
+// Copyright 2022 The Matrix.org Foundation C.I.C.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package routing_test
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"testing"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/relayapi/internal"
+	"github.com/matrix-org/dendrite/relayapi/routing"
+	"github.com/matrix-org/dendrite/relayapi/storage/shared"
+	"github.com/matrix-org/dendrite/test"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	testOrigin = gomatrixserverlib.ServerName("kaer.morhen")
+)
+
+func createTransaction() gomatrixserverlib.Transaction {
+	txn := gomatrixserverlib.Transaction{}
+	txn.PDUs = []json.RawMessage{
+		[]byte(`{"auth_events":[["$0ok8ynDp7kjc95e3:kaer.morhen",{"sha256":"sWCi6Ckp9rDimQON+MrUlNRkyfZ2tjbPbWfg2NMB18Q"}],["$LEwEu0kxrtu5fOiS:kaer.morhen",{"sha256":"1aKajq6DWHru1R1HJjvdWMEavkJJHGaTmPvfuERUXaA"}]],"content":{"body":"Test Message"},"depth":5,"event_id":"$gl2T9l3qm0kUbiIJ:kaer.morhen","hashes":{"sha256":"Qx3nRMHLDPSL5hBAzuX84FiSSP0K0Kju2iFoBWH4Za8"},"origin":"kaer.morhen","origin_server_ts":0,"prev_events":[["$UKNe10XzYzG0TeA9:kaer.morhen",{"sha256":"KtSRyMjt0ZSjsv2koixTRCxIRCGoOp6QrKscsW97XRo"}]],"room_id":"!roomid:kaer.morhen","sender":"@userid:kaer.morhen","signatures":{"kaer.morhen":{"ed25519:auto":"sqDgv3EG7ml5VREzmT9aZeBpS4gAPNIaIeJOwqjDhY0GPU/BcpX5wY4R7hYLrNe5cChgV+eFy/GWm1Zfg5FfDg"}},"type":"m.room.message"}`),
+	}
+	txn.Origin = testOrigin
+	return txn
+}
+
+func createFederationRequest(
+	userID gomatrixserverlib.UserID,
+	txnID gomatrixserverlib.TransactionID,
+	origin gomatrixserverlib.ServerName,
+	destination gomatrixserverlib.ServerName,
+	content interface{},
+) gomatrixserverlib.FederationRequest {
+	var federationPathPrefixV1 = "/_matrix/federation/v1"
+	path := federationPathPrefixV1 + "/send_relay/" + string(txnID) + "/" + userID.Raw()
+	request := gomatrixserverlib.NewFederationRequest("PUT", origin, destination, path)
+	request.SetContent(content)
+
+	return request
+}
+
+func TestForwardEmptyReturnsOk(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	txn := createTransaction()
+	request := createFederationRequest(*userID, txn.TransactionID, txn.Origin, txn.Destination, txn)
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
+
+	assert.Equal(t, 200, response.Code)
+}
+
+func TestForwardBadJSONReturnsError(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	type BadData struct {
+		Field bool `json:"pdus"`
+	}
+	content := BadData{
+		Field: false,
+	}
+	txn := createTransaction()
+	request := createFederationRequest(*userID, txn.TransactionID, txn.Origin, txn.Destination, content)
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
+
+	assert.NotEqual(t, 200, response.Code)
+}
+
+func TestForwardTooManyPDUsReturnsError(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	type BadData struct {
+		Field []json.RawMessage `json:"pdus"`
+	}
+	content := BadData{
+		Field: []json.RawMessage{},
+	}
+	for i := 0; i < 51; i++ {
+		content.Field = append(content.Field, []byte{})
+	}
+	assert.Greater(t, len(content.Field), 50)
+
+	txn := createTransaction()
+	request := createFederationRequest(*userID, txn.TransactionID, txn.Origin, txn.Destination, content)
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
+
+	assert.NotEqual(t, 200, response.Code)
+}
+
+func TestForwardTooManyEDUsReturnsError(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	type BadData struct {
+		Field []gomatrixserverlib.EDU `json:"edus"`
+	}
+	content := BadData{
+		Field: []gomatrixserverlib.EDU{},
+	}
+	for i := 0; i < 101; i++ {
+		content.Field = append(content.Field, gomatrixserverlib.EDU{Type: gomatrixserverlib.MTyping})
+	}
+	assert.Greater(t, len(content.Field), 100)
+
+	txn := createTransaction()
+	request := createFederationRequest(*userID, txn.TransactionID, txn.Origin, txn.Destination, content)
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	response := routing.SendTransactionToRelay(httpReq, &request, relayAPI, "1", *userID)
+
+	assert.NotEqual(t, 200, response.Code)
+}
+
+func TestUniqueTransactionStoredInDatabase(t *testing.T) {
+	testDB := test.NewInMemoryRelayDatabase()
+	db := shared.Database{
+		Writer:         sqlutil.NewDummyWriter(),
+		RelayQueue:     testDB,
+		RelayQueueJSON: testDB,
+	}
+	httpReq := &http.Request{}
+	userID, err := gomatrixserverlib.NewUserID("@local:domain", false)
+	assert.NoError(t, err, "Invalid userID")
+
+	txn := createTransaction()
+	request := createFederationRequest(*userID, txn.TransactionID, txn.Origin, txn.Destination, txn)
+
+	relayAPI := internal.NewRelayInternalAPI(
+		&db, nil, nil, nil, nil, false, "",
+	)
+
+	response := routing.SendTransactionToRelay(
+		httpReq, &request, relayAPI, txn.TransactionID, *userID)
+	transaction, _, err := db.GetTransaction(context.Background(), *userID)
+	assert.NoError(t, err, "Failed retrieving transaction")
+
+	transactionCount, err := db.GetTransactionCount(context.Background(), *userID)
+	assert.NoError(t, err, "Failed retrieving transaction count")
+
+	assert.Equal(t, 200, response.Code)
+	assert.Equal(t, int64(1), transactionCount)
+	assert.Equal(t, txn.TransactionID, transaction.TransactionID)
+}