Loopback event from invite response (#982)

* Working invite v2 support

* Fix copyright notice

* Update gomatrixserverlib

* Add fallthrough

* Add missing continue

* Update sytest-whitelist, gomatrixserverlib

* Update gomatrixserverlib to test matrix-org/gomatrixserverlib#181

* Update gomatrixserverlib

federationsender/federationsender.go

@@ -20,6 +20,7 @@ import (
 	"github.com/matrix-org/dendrite/common/basecomponent"
 	"github.com/matrix-org/dendrite/federationsender/api"
 	"github.com/matrix-org/dendrite/federationsender/consumers"
+	"github.com/matrix-org/dendrite/federationsender/producers"
 	"github.com/matrix-org/dendrite/federationsender/query"
 	"github.com/matrix-org/dendrite/federationsender/queue"
 	"github.com/matrix-org/dendrite/federationsender/storage"
@@ -34,13 +35,16 @@ func SetupFederationSenderComponent(
 	base *basecomponent.BaseDendrite,
 	federation *gomatrixserverlib.FederationClient,
 	rsQueryAPI roomserverAPI.RoomserverQueryAPI,
+	rsInputAPI roomserverAPI.RoomserverInputAPI,
 ) api.FederationSenderQueryAPI {
 	federationSenderDB, err := storage.NewDatabase(string(base.Cfg.Database.FederationSender))
 	if err != nil {
 		logrus.WithError(err).Panic("failed to connect to federation sender db")
 	}
 
-	queues := queue.NewOutgoingQueues(base.Cfg.Matrix.ServerName, federation)
+	roomserverProducer := producers.NewRoomserverProducer(rsInputAPI, base.Cfg.Matrix.ServerName)
+
+	queues := queue.NewOutgoingQueues(base.Cfg.Matrix.ServerName, federation, roomserverProducer)
 
 	rsConsumer := consumers.NewOutputRoomEventConsumer(
 		base.Cfg, base.KafkaConsumer, queues,

federationsender/producers/roomserver.go

@@ -0,0 +1,66 @@
+// Copyright 2020 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 producers
+
+import (
+	"context"
+
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+// RoomserverProducer produces events for the roomserver to consume.
+type RoomserverProducer struct {
+	InputAPI   api.RoomserverInputAPI
+	serverName gomatrixserverlib.ServerName
+}
+
+// NewRoomserverProducer creates a new RoomserverProducer
+func NewRoomserverProducer(
+	inputAPI api.RoomserverInputAPI, serverName gomatrixserverlib.ServerName,
+) *RoomserverProducer {
+	return &RoomserverProducer{
+		InputAPI:   inputAPI,
+		serverName: serverName,
+	}
+}
+
+// SendInviteResponse drops an invite response back into the roomserver so that users
+// already in the room will be notified of the new invite. The invite response is signed
+// by the remote side.
+func (c *RoomserverProducer) SendInviteResponse(
+	ctx context.Context, res gomatrixserverlib.RespInviteV2, roomVersion gomatrixserverlib.RoomVersion,
+) (string, error) {
+	ev := res.Event.Headered(roomVersion)
+	ire := api.InputRoomEvent{
+		Kind:          api.KindNew,
+		Event:         ev,
+		AuthEventIDs:  ev.AuthEventIDs(),
+		SendAsServer:  string(c.serverName),
+		TransactionID: nil,
+	}
+	return c.SendInputRoomEvents(ctx, []api.InputRoomEvent{ire})
+}
+
+// SendInputRoomEvents writes the given input room events to the roomserver input API.
+func (c *RoomserverProducer) SendInputRoomEvents(
+	ctx context.Context, ires []api.InputRoomEvent,
+) (eventID string, err error) {
+	request := api.InputRoomEventsRequest{InputRoomEvents: ires}
+	var response api.InputRoomEventsResponse
+	err = c.InputAPI.InputRoomEvents(ctx, &request, &response)
+	eventID = response.EventID
+	return
+}

federationsender/queue/destinationqueue.go

@@ -21,6 +21,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/matrix-org/dendrite/federationsender/producers"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
 	log "github.com/sirupsen/logrus"
@@ -32,6 +33,7 @@ import (
 // ensures that only one request is in flight to a given destination
 // at a time.
 type destinationQueue struct {
+	rsProducer  *producers.RoomserverProducer
 	client      *gomatrixserverlib.FederationClient
 	origin      gomatrixserverlib.ServerName
 	destination gomatrixserverlib.ServerName
@@ -165,18 +167,38 @@ func (oq *destinationQueue) nextInvites() bool {
 	}
 
 	for _, inviteReq := range oq.pendingInvites {
-		ev := inviteReq.Event()
+		ev, roomVersion := inviteReq.Event(), inviteReq.RoomVersion()
 
-		if _, err := oq.client.SendInviteV2(
+		log.WithFields(log.Fields{
+			"event_id":     ev.EventID(),
+			"room_version": roomVersion,
+			"destination":  oq.destination,
+		}).Info("sending invite")
+
+		inviteRes, err := oq.client.SendInviteV2(
 			context.TODO(),
 			oq.destination,
 			*inviteReq,
-		); err != nil {
+		)
+		if err != nil {
 			log.WithFields(log.Fields{
 				"event_id":    ev.EventID(),
 				"state_key":   ev.StateKey(),
 				"destination": oq.destination,
 			}).WithError(err).Error("failed to send invite")
+			continue
+		}
+
+		if _, err = oq.rsProducer.SendInviteResponse(
+			context.TODO(),
+			inviteRes,
+			roomVersion,
+		); err != nil {
+			log.WithFields(log.Fields{
+				"event_id":    ev.EventID(),
+				"state_key":   ev.StateKey(),
+				"destination": oq.destination,
+			}).WithError(err).Error("failed to return signed invite to roomserver")
 		}
 	}
 

federationsender/queue/queue.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/matrix-org/dendrite/federationsender/producers"
 	"github.com/matrix-org/gomatrixserverlib"
 	log "github.com/sirupsen/logrus"
 )
@@ -25,19 +26,25 @@ import (
 // OutgoingQueues is a collection of queues for sending transactions to other
 // matrix servers
 type OutgoingQueues struct {
-	origin gomatrixserverlib.ServerName
-	client *gomatrixserverlib.FederationClient
+	rsProducer *producers.RoomserverProducer
+	origin     gomatrixserverlib.ServerName
+	client     *gomatrixserverlib.FederationClient
 	// The queuesMutex protects queues
 	queuesMutex sync.Mutex
 	queues      map[gomatrixserverlib.ServerName]*destinationQueue
 }
 
 // NewOutgoingQueues makes a new OutgoingQueues
-func NewOutgoingQueues(origin gomatrixserverlib.ServerName, client *gomatrixserverlib.FederationClient) *OutgoingQueues {
+func NewOutgoingQueues(
+	origin gomatrixserverlib.ServerName,
+	client *gomatrixserverlib.FederationClient,
+	rsProducer *producers.RoomserverProducer,
+) *OutgoingQueues {
 	return &OutgoingQueues{
-		origin: origin,
-		client: client,
-		queues: map[gomatrixserverlib.ServerName]*destinationQueue{},
+		rsProducer: rsProducer,
+		origin:     origin,
+		client:     client,
+		queues:     map[gomatrixserverlib.ServerName]*destinationQueue{},
 	}
 }
 
@@ -67,6 +74,7 @@ func (oqs *OutgoingQueues) SendEvent(
 		oq := oqs.queues[destination]
 		if oq == nil {
 			oq = &destinationQueue{
+				rsProducer:  oqs.rsProducer,
 				origin:      oqs.origin,
 				destination: destination,
 				client:      oqs.client,
@@ -111,6 +119,7 @@ func (oqs *OutgoingQueues) SendInvite(
 	oq := oqs.queues[destination]
 	if oq == nil {
 		oq = &destinationQueue{
+			rsProducer:  oqs.rsProducer,
 			origin:      oqs.origin,
 			destination: destination,
 			client:      oqs.client,
@@ -151,6 +160,7 @@ func (oqs *OutgoingQueues) SendEDU(
 		oq := oqs.queues[destination]
 		if oq == nil {
 			oq = &destinationQueue{
+				rsProducer:  oqs.rsProducer,
 				origin:      oqs.origin,
 				destination: destination,
 				client:      oqs.client,