Merge branch 'master' into nats

federationapi/consumers/eduserver.go

@@ -0,0 +1,251 @@
+// 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 consumers
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/matrix-org/dendrite/eduserver/api"
+	"github.com/matrix-org/dendrite/federationapi/queue"
+	"github.com/matrix-org/dendrite/federationapi/storage"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/dendrite/setup/process"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/matrix-org/util"
+	"github.com/nats-io/nats.go"
+	log "github.com/sirupsen/logrus"
+)
+
+// OutputEDUConsumer consumes events that originate in EDU server.
+type OutputEDUConsumer struct {
+	jetstream         nats.JetStreamContext
+	db                storage.Database
+	queues            *queue.OutgoingQueues
+	ServerName        gomatrixserverlib.ServerName
+	typingTopic       string
+	sendToDeviceTopic string
+	receiptTopic      string
+}
+
+// NewOutputEDUConsumer creates a new OutputEDUConsumer. Call Start() to begin consuming from EDU servers.
+func NewOutputEDUConsumer(
+	process *process.ProcessContext,
+	cfg *config.FederationAPI,
+	js nats.JetStreamContext,
+	queues *queue.OutgoingQueues,
+	store storage.Database,
+) *OutputEDUConsumer {
+	return &OutputEDUConsumer{
+		jetstream:         js,
+		queues:            queues,
+		db:                store,
+		ServerName:        cfg.Matrix.ServerName,
+		typingTopic:       cfg.Matrix.JetStream.TopicFor(jetstream.OutputTypingEvent),
+		sendToDeviceTopic: cfg.Matrix.JetStream.TopicFor(jetstream.OutputSendToDeviceEvent),
+		receiptTopic:      cfg.Matrix.JetStream.TopicFor(jetstream.OutputReceiptEvent),
+	}
+}
+
+// Start consuming from EDU servers
+func (t *OutputEDUConsumer) Start() error {
+	if _, err := t.jetstream.Subscribe(t.typingTopic, t.onTypingEvent); err != nil {
+		return err
+	}
+	if _, err := t.jetstream.Subscribe(t.sendToDeviceTopic, t.onSendToDeviceEvent); err != nil {
+		return err
+	}
+	if _, err := t.jetstream.Subscribe(t.receiptTopic, t.onReceiptEvent); err != nil {
+		return err
+	}
+	return nil
+}
+
+// onSendToDeviceEvent is called in response to a message received on the
+// send-to-device events topic from the EDU server.
+func (t *OutputEDUConsumer) onSendToDeviceEvent(msg *nats.Msg) {
+	// Extract the send-to-device event from msg.
+	var ote api.OutputSendToDeviceEvent
+	if err := json.Unmarshal(msg.Data, &ote); err != nil {
+		log.WithError(err).Errorf("eduserver output log: message parse failed (expected send-to-device)")
+		_ = msg.Ack()
+		return
+	}
+
+	// only send send-to-device events which originated from us
+	_, originServerName, err := gomatrixserverlib.SplitID('@', ote.Sender)
+	if err != nil {
+		log.WithError(err).WithField("user_id", ote.Sender).Error("Failed to extract domain from send-to-device sender")
+		_ = msg.Ack()
+		return
+	}
+	if originServerName != t.ServerName {
+		log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
+		_ = msg.Ack()
+		return
+	}
+
+	_, destServerName, err := gomatrixserverlib.SplitID('@', ote.UserID)
+	if err != nil {
+		log.WithError(err).WithField("user_id", ote.UserID).Error("Failed to extract domain from send-to-device destination")
+		_ = msg.Ack()
+		return
+	}
+
+	// Pack the EDU and marshal it
+	edu := &gomatrixserverlib.EDU{
+		Type:   gomatrixserverlib.MDirectToDevice,
+		Origin: string(t.ServerName),
+	}
+	tdm := gomatrixserverlib.ToDeviceMessage{
+		Sender:    ote.Sender,
+		Type:      ote.Type,
+		MessageID: util.RandomString(32),
+		Messages: map[string]map[string]json.RawMessage{
+			ote.UserID: {
+				ote.DeviceID: ote.Content,
+			},
+		},
+	}
+	if edu.Content, err = json.Marshal(tdm); err != nil {
+		log.WithError(err).Error("failed to marshal EDU JSON")
+		_ = msg.Ack()
+		return
+	}
+
+	log.Infof("Sending send-to-device message into %q destination queue", destServerName)
+	if err := t.queues.SendEDU(edu, t.ServerName, []gomatrixserverlib.ServerName{destServerName}); err != nil {
+		log.WithError(err).Error("failed to send EDU")
+	}
+
+	_ = msg.Ack()
+}
+
+// onTypingEvent is called in response to a message received on the typing
+// events topic from the EDU server.
+func (t *OutputEDUConsumer) onTypingEvent(msg *nats.Msg) {
+	// Extract the typing event from msg.
+	var ote api.OutputTypingEvent
+	if err := json.Unmarshal(msg.Data, &ote); err != nil {
+		// Skip this msg but continue processing messages.
+		log.WithError(err).Errorf("eduserver output log: message parse failed (expected typing)")
+		_ = msg.Ack()
+		return
+	}
+
+	// only send typing events which originated from us
+	_, typingServerName, err := gomatrixserverlib.SplitID('@', ote.Event.UserID)
+	if err != nil {
+		log.WithError(err).WithField("user_id", ote.Event.UserID).Error("Failed to extract domain from typing sender")
+		_ = msg.Ack()
+		return
+	}
+	if typingServerName != t.ServerName {
+		return
+	}
+
+	joined, err := t.db.GetJoinedHosts(context.TODO(), ote.Event.RoomID)
+	if err != nil {
+		log.WithError(err).WithField("room_id", ote.Event.RoomID).Error("failed to get joined hosts for room")
+		return
+	}
+
+	names := make([]gomatrixserverlib.ServerName, len(joined))
+	for i := range joined {
+		names[i] = joined[i].ServerName
+	}
+
+	edu := &gomatrixserverlib.EDU{Type: ote.Event.Type}
+	if edu.Content, err = json.Marshal(map[string]interface{}{
+		"room_id": ote.Event.RoomID,
+		"user_id": ote.Event.UserID,
+		"typing":  ote.Event.Typing,
+	}); err != nil {
+		log.WithError(err).Error("failed to marshal EDU JSON")
+		_ = msg.Ack()
+		return
+	}
+
+	if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
+		log.WithError(err).Error("failed to send EDU")
+	}
+
+	_ = msg.Ack()
+}
+
+// onReceiptEvent is called in response to a message received on the receipt
+// events topic from the EDU server.
+func (t *OutputEDUConsumer) onReceiptEvent(msg *nats.Msg) {
+	// Extract the typing event from msg.
+	var receipt api.OutputReceiptEvent
+	if err := json.Unmarshal(msg.Data, &receipt); err != nil {
+		// Skip this msg but continue processing messages.
+		log.WithError(err).Errorf("eduserver output log: message parse failed (expected receipt)")
+		_ = msg.Ack()
+		return
+	}
+
+	// only send receipt events which originated from us
+	_, receiptServerName, err := gomatrixserverlib.SplitID('@', receipt.UserID)
+	if err != nil {
+		log.WithError(err).WithField("user_id", receipt.UserID).Error("failed to extract domain from receipt sender")
+		_ = msg.Ack()
+		return
+	}
+	if receiptServerName != t.ServerName {
+		_ = msg.Ack()
+		return // don't log, very spammy as it logs for each remote receipt
+	}
+
+	joined, err := t.db.GetJoinedHosts(context.TODO(), receipt.RoomID)
+	if err != nil {
+		log.WithError(err).WithField("room_id", receipt.RoomID).Error("failed to get joined hosts for room")
+		return
+	}
+
+	names := make([]gomatrixserverlib.ServerName, len(joined))
+	for i := range joined {
+		names[i] = joined[i].ServerName
+	}
+
+	content := map[string]api.FederationReceiptMRead{}
+	content[receipt.RoomID] = api.FederationReceiptMRead{
+		User: map[string]api.FederationReceiptData{
+			receipt.UserID: {
+				Data: api.ReceiptTS{
+					TS: receipt.Timestamp,
+				},
+				EventIDs: []string{receipt.EventID},
+			},
+		},
+	}
+
+	edu := &gomatrixserverlib.EDU{
+		Type:   gomatrixserverlib.MReceipt,
+		Origin: string(t.ServerName),
+	}
+	if edu.Content, err = json.Marshal(content); err != nil {
+		log.WithError(err).Error("failed to marshal EDU JSON")
+		_ = msg.Ack()
+		return
+	}
+
+	if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
+		log.WithError(err).Error("failed to send EDU")
+	}
+
+	_ = msg.Ack()
+}

federationapi/consumers/keychange.go

@@ -0,0 +1,206 @@
+// 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 consumers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/Shopify/sarama"
+	eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
+	"github.com/matrix-org/dendrite/federationapi/queue"
+	"github.com/matrix-org/dendrite/federationapi/storage"
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/keyserver/api"
+	roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/dendrite/setup/process"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/sirupsen/logrus"
+)
+
+// KeyChangeConsumer consumes events that originate in key server.
+type KeyChangeConsumer struct {
+	consumer   *internal.ContinualConsumer
+	db         storage.Database
+	queues     *queue.OutgoingQueues
+	serverName gomatrixserverlib.ServerName
+	rsAPI      roomserverAPI.RoomserverInternalAPI
+}
+
+// NewKeyChangeConsumer creates a new KeyChangeConsumer. Call Start() to begin consuming from key servers.
+func NewKeyChangeConsumer(
+	process *process.ProcessContext,
+	cfg *config.KeyServer,
+	kafkaConsumer sarama.Consumer,
+	queues *queue.OutgoingQueues,
+	store storage.Database,
+	rsAPI roomserverAPI.RoomserverInternalAPI,
+) *KeyChangeConsumer {
+	c := &KeyChangeConsumer{
+		consumer: &internal.ContinualConsumer{
+			Process:        process,
+			ComponentName:  "federationsender/keychange",
+			Topic:          string(cfg.Matrix.JetStream.TopicFor(jetstream.OutputKeyChangeEvent)),
+			Consumer:       kafkaConsumer,
+			PartitionStore: store,
+		},
+		queues:     queues,
+		db:         store,
+		serverName: cfg.Matrix.ServerName,
+		rsAPI:      rsAPI,
+	}
+	c.consumer.ProcessMessage = c.onMessage
+
+	return c
+}
+
+// Start consuming from key servers
+func (t *KeyChangeConsumer) Start() error {
+	if err := t.consumer.Start(); err != nil {
+		return fmt.Errorf("t.consumer.Start: %w", err)
+	}
+	return nil
+}
+
+// onMessage is called in response to a message received on the
+// key change events topic from the key server.
+func (t *KeyChangeConsumer) onMessage(msg *sarama.ConsumerMessage) error {
+	var m api.DeviceMessage
+	if err := json.Unmarshal(msg.Value, &m); err != nil {
+		logrus.WithError(err).Errorf("failed to read device message from key change topic")
+		return nil
+	}
+	if m.DeviceKeys == nil && m.OutputCrossSigningKeyUpdate == nil {
+		// This probably shouldn't happen but stops us from panicking if we come
+		// across an update that doesn't satisfy either types.
+		return nil
+	}
+	switch m.Type {
+	case api.TypeCrossSigningUpdate:
+		return t.onCrossSigningMessage(m)
+	case api.TypeDeviceKeyUpdate:
+		fallthrough
+	default:
+		return t.onDeviceKeyMessage(m)
+	}
+}
+
+func (t *KeyChangeConsumer) onDeviceKeyMessage(m api.DeviceMessage) error {
+	if m.DeviceKeys == nil {
+		return nil
+	}
+	logger := logrus.WithField("user_id", m.UserID)
+
+	// only send key change events which originated from us
+	_, originServerName, err := gomatrixserverlib.SplitID('@', m.UserID)
+	if err != nil {
+		logger.WithError(err).Error("Failed to extract domain from key change event")
+		return nil
+	}
+	if originServerName != t.serverName {
+		return nil
+	}
+
+	var queryRes roomserverAPI.QueryRoomsForUserResponse
+	err = t.rsAPI.QueryRoomsForUser(context.Background(), &roomserverAPI.QueryRoomsForUserRequest{
+		UserID:         m.UserID,
+		WantMembership: "join",
+	}, &queryRes)
+	if err != nil {
+		logger.WithError(err).Error("failed to calculate joined rooms for user")
+		return nil
+	}
+	// send this key change to all servers who share rooms with this user.
+	destinations, err := t.db.GetJoinedHostsForRooms(context.Background(), queryRes.RoomIDs)
+	if err != nil {
+		logger.WithError(err).Error("failed to calculate joined hosts for rooms user is in")
+		return nil
+	}
+
+	// Pack the EDU and marshal it
+	edu := &gomatrixserverlib.EDU{
+		Type:   gomatrixserverlib.MDeviceListUpdate,
+		Origin: string(t.serverName),
+	}
+	event := gomatrixserverlib.DeviceListUpdateEvent{
+		UserID:            m.UserID,
+		DeviceID:          m.DeviceID,
+		DeviceDisplayName: m.DisplayName,
+		StreamID:          m.StreamID,
+		PrevID:            prevID(m.StreamID),
+		Deleted:           len(m.KeyJSON) == 0,
+		Keys:              m.KeyJSON,
+	}
+	if edu.Content, err = json.Marshal(event); err != nil {
+		return err
+	}
+
+	logrus.Infof("Sending device list update message to %q", destinations)
+	return t.queues.SendEDU(edu, t.serverName, destinations)
+}
+
+func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) error {
+	output := m.CrossSigningKeyUpdate
+	_, host, err := gomatrixserverlib.SplitID('@', output.UserID)
+	if err != nil {
+		logrus.WithError(err).Errorf("fedsender key change consumer: user ID parse failure")
+		return nil
+	}
+	if host != gomatrixserverlib.ServerName(t.serverName) {
+		// Ignore any messages that didn't originate locally, otherwise we'll
+		// end up parroting information we received from other servers.
+		return nil
+	}
+	logger := logrus.WithField("user_id", output.UserID)
+
+	var queryRes roomserverAPI.QueryRoomsForUserResponse
+	err = t.rsAPI.QueryRoomsForUser(context.Background(), &roomserverAPI.QueryRoomsForUserRequest{
+		UserID:         output.UserID,
+		WantMembership: "join",
+	}, &queryRes)
+	if err != nil {
+		logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined rooms for user")
+		return nil
+	}
+	// send this key change to all servers who share rooms with this user.
+	destinations, err := t.db.GetJoinedHostsForRooms(context.Background(), queryRes.RoomIDs)
+	if err != nil {
+		logger.WithError(err).Error("fedsender key change consumer: failed to calculate joined hosts for rooms user is in")
+		return nil
+	}
+
+	// Pack the EDU and marshal it
+	edu := &gomatrixserverlib.EDU{
+		Type:   eduserverAPI.MSigningKeyUpdate,
+		Origin: string(t.serverName),
+	}
+	if edu.Content, err = json.Marshal(output); err != nil {
+		logger.WithError(err).Error("fedsender key change consumer: failed to marshal output, dropping")
+		return nil
+	}
+
+	logger.Infof("Sending cross-signing update message to %q", destinations)
+	return t.queues.SendEDU(edu, t.serverName, destinations)
+}
+
+func prevID(streamID int) []int {
+	if streamID <= 1 {
+		return nil
+	}
+	return []int{streamID - 1}
+}

federationapi/consumers/roomserver.go

@@ -0,0 +1,407 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// 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 consumers
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/matrix-org/dendrite/federationapi/queue"
+	"github.com/matrix-org/dendrite/federationapi/storage"
+	"github.com/matrix-org/dendrite/federationapi/types"
+	"github.com/matrix-org/dendrite/roomserver/api"
+	"github.com/matrix-org/dendrite/setup/config"
+	"github.com/matrix-org/dendrite/setup/jetstream"
+	"github.com/matrix-org/dendrite/setup/process"
+	"github.com/matrix-org/gomatrixserverlib"
+	"github.com/nats-io/nats.go"
+	log "github.com/sirupsen/logrus"
+)
+
+// OutputRoomEventConsumer consumes events that originated in the room server.
+type OutputRoomEventConsumer struct {
+	cfg       *config.FederationAPI
+	rsAPI     api.RoomserverInternalAPI
+	jetstream nats.JetStreamContext
+	db        storage.Database
+	queues    *queue.OutgoingQueues
+	topic     string
+}
+
+// NewOutputRoomEventConsumer creates a new OutputRoomEventConsumer. Call Start() to begin consuming from room servers.
+func NewOutputRoomEventConsumer(
+	process *process.ProcessContext,
+	cfg *config.FederationAPI,
+	js nats.JetStreamContext,
+	queues *queue.OutgoingQueues,
+	store storage.Database,
+	rsAPI api.RoomserverInternalAPI,
+) *OutputRoomEventConsumer {
+	return &OutputRoomEventConsumer{
+		cfg:       cfg,
+		jetstream: js,
+		db:        store,
+		queues:    queues,
+		rsAPI:     rsAPI,
+		topic:     cfg.Matrix.JetStream.TopicFor(jetstream.OutputRoomEvent),
+	}
+}
+
+// Start consuming from room servers
+func (s *OutputRoomEventConsumer) Start() error {
+	_, err := s.jetstream.Subscribe(s.topic, s.onMessage)
+	return err
+}
+
+// onMessage is called when the federation server receives a new event from the room server output log.
+// It is unsafe to call this with messages for the same room in multiple gorountines
+// because updates it will likely fail with a types.EventIDMismatchError when it
+// realises that it cannot update the room state using the deltas.
+func (s *OutputRoomEventConsumer) onMessage(msg *nats.Msg) {
+	// Parse out the event JSON
+	var output api.OutputEvent
+	if err := json.Unmarshal(msg.Data, &output); err != nil {
+		// If the message was invalid, log it and move on to the next message in the stream
+		log.WithError(err).Errorf("roomserver output log: message parse failure")
+		_ = msg.Ack()
+		return
+	}
+
+	switch output.Type {
+	case api.OutputTypeNewRoomEvent:
+		ev := output.NewRoomEvent.Event
+
+		if output.NewRoomEvent.RewritesState {
+			if err := s.db.PurgeRoomState(context.TODO(), ev.RoomID()); err != nil {
+				log.WithError(err).Errorf("roomserver output log: purge room state failure")
+				return
+			}
+		}
+
+		if err := s.processMessage(*output.NewRoomEvent); err != nil {
+			switch err.(type) {
+			case *queue.ErrorFederationDisabled:
+				log.WithField("error", output.Type).Info(
+					err.Error(),
+				)
+				_ = msg.Ack()
+			default:
+				// panic rather than continue with an inconsistent database
+				log.WithFields(log.Fields{
+					"event_id":   ev.EventID(),
+					"event":      string(ev.JSON()),
+					"add":        output.NewRoomEvent.AddsStateEventIDs,
+					"del":        output.NewRoomEvent.RemovesStateEventIDs,
+					log.ErrorKey: err,
+				}).Panicf("roomserver output log: write room event failure")
+			}
+			return
+		}
+
+		_ = msg.Ack()
+
+	case api.OutputTypeNewInboundPeek:
+		if err := s.processInboundPeek(*output.NewInboundPeek); err != nil {
+			log.WithFields(log.Fields{
+				"event":      output.NewInboundPeek,
+				log.ErrorKey: err,
+			}).Panicf("roomserver output log: remote peek event failure")
+			return
+		}
+		_ = msg.Ack()
+
+	default:
+		log.WithField("type", output.Type).Debug(
+			"roomserver output log: ignoring unknown output type",
+		)
+		_ = msg.Ack()
+		return
+	}
+}
+
+// processInboundPeek starts tracking a new federated inbound peek (replacing the existing one if any)
+// causing the federationapi to start sending messages to the peeking server
+func (s *OutputRoomEventConsumer) processInboundPeek(orp api.OutputNewInboundPeek) error {
+
+	// FIXME: there's a race here - we should start /sending new peeked events
+	// atomically after the orp.LatestEventID to ensure there are no gaps between
+	// the peek beginning and the send stream beginning.
+	//
+	// We probably need to track orp.LatestEventID on the inbound peek, but it's
+	// unclear how we then use that to prevent the race when we start the send
+	// stream.
+	//
+	// This is making the tests flakey.
+
+	return s.db.AddInboundPeek(context.TODO(), orp.ServerName, orp.RoomID, orp.PeekID, orp.RenewalInterval)
+}
+
+// processMessage updates the list of currently joined hosts in the room
+// and then sends the event to the hosts that were joined before the event.
+func (s *OutputRoomEventConsumer) processMessage(ore api.OutputNewRoomEvent) error {
+	addsJoinedHosts, err := joinedHostsFromEvents(gomatrixserverlib.UnwrapEventHeaders(ore.AddsState()))
+	if err != nil {
+		return err
+	}
+	// Update our copy of the current state.
+	// We keep a copy of the current state because the state at each event is
+	// expressed as a delta against the current state.
+	// TODO(#290): handle EventIDMismatchError and recover the current state by
+	// talking to the roomserver
+	oldJoinedHosts, err := s.db.UpdateRoom(
+		context.TODO(),
+		ore.Event.RoomID(),
+		ore.LastSentEventID,
+		ore.Event.EventID(),
+		addsJoinedHosts,
+		ore.RemovesStateEventIDs,
+	)
+	if err != nil {
+		return err
+	}
+
+	if oldJoinedHosts == nil {
+		// This means that there is nothing to update as this is a duplicate
+		// message.
+		// This can happen if dendrite crashed between reading the message and
+		// persisting the stream position.
+		return nil
+	}
+
+	if ore.SendAsServer == api.DoNotSendToOtherServers {
+		// Ignore event that we don't need to send anywhere.
+		return nil
+	}
+
+	// Work out which hosts were joined at the event itself.
+	joinedHostsAtEvent, err := s.joinedHostsAtEvent(ore, oldJoinedHosts)
+	if err != nil {
+		return err
+	}
+
+	// TODO: do housekeeping to evict unrenewed peeking hosts
+
+	// TODO: implement query to let the fedapi check whether a given peek is live or not
+
+	// Send the event.
+	return s.queues.SendEvent(
+		ore.Event, gomatrixserverlib.ServerName(ore.SendAsServer), joinedHostsAtEvent,
+	)
+}
+
+// joinedHostsAtEvent works out a list of matrix servers that were joined to
+// the room at the event (including peeking ones)
+// It is important to use the state at the event for sending messages because:
+//   1) We shouldn't send messages to servers that weren't in the room.
+//   2) If a server is kicked from the rooms it should still be told about the
+//      kick event,
+// Usually the list can be calculated locally, but sometimes it will need fetch
+// events from the room server.
+// Returns an error if there was a problem talking to the room server.
+func (s *OutputRoomEventConsumer) joinedHostsAtEvent(
+	ore api.OutputNewRoomEvent, oldJoinedHosts []types.JoinedHost,
+) ([]gomatrixserverlib.ServerName, error) {
+	// Combine the delta into a single delta so that the adds and removes can
+	// cancel each other out. This should reduce the number of times we need
+	// to fetch a state event from the room server.
+	combinedAdds, combinedRemoves := combineDeltas(
+		ore.AddsStateEventIDs, ore.RemovesStateEventIDs,
+		ore.StateBeforeAddsEventIDs, ore.StateBeforeRemovesEventIDs,
+	)
+	combinedAddsEvents, err := s.lookupStateEvents(combinedAdds, ore.Event.Event)
+	if err != nil {
+		return nil, err
+	}
+
+	combinedAddsJoinedHosts, err := joinedHostsFromEvents(combinedAddsEvents)
+	if err != nil {
+		return nil, err
+	}
+
+	removed := map[string]bool{}
+	for _, eventID := range combinedRemoves {
+		removed[eventID] = true
+	}
+
+	joined := map[gomatrixserverlib.ServerName]bool{}
+	for _, joinedHost := range oldJoinedHosts {
+		if removed[joinedHost.MemberEventID] {
+			// This m.room.member event is part of the current state of the
+			// room, but not part of the state at the event we are processing
+			// Therefore we can't use it to tell whether the server was in
+			// the room at the event.
+			continue
+		}
+		joined[joinedHost.ServerName] = true
+	}
+
+	for _, joinedHost := range combinedAddsJoinedHosts {
+		// This m.room.member event was part of the state of the room at the
+		// event, but isn't part of the current state of the room now.
+		joined[joinedHost.ServerName] = true
+	}
+
+	// handle peeking hosts
+	inboundPeeks, err := s.db.GetInboundPeeks(context.TODO(), ore.Event.Event.RoomID())
+	if err != nil {
+		return nil, err
+	}
+	for _, inboundPeek := range inboundPeeks {
+		joined[inboundPeek.ServerName] = true
+	}
+
+	var result []gomatrixserverlib.ServerName
+	for serverName, include := range joined {
+		if include {
+			result = append(result, serverName)
+		}
+	}
+	return result, nil
+}
+
+// joinedHostsFromEvents turns a list of state events into a list of joined hosts.
+// This errors if one of the events was invalid.
+// It should be impossible for an invalid event to get this far in the pipeline.
+func joinedHostsFromEvents(evs []*gomatrixserverlib.Event) ([]types.JoinedHost, error) {
+	var joinedHosts []types.JoinedHost
+	for _, ev := range evs {
+		if ev.Type() != "m.room.member" || ev.StateKey() == nil {
+			continue
+		}
+		membership, err := ev.Membership()
+		if err != nil {
+			return nil, err
+		}
+		if membership != gomatrixserverlib.Join {
+			continue
+		}
+		_, serverName, err := gomatrixserverlib.SplitID('@', *ev.StateKey())
+		if err != nil {
+			return nil, err
+		}
+		joinedHosts = append(joinedHosts, types.JoinedHost{
+			MemberEventID: ev.EventID(), ServerName: serverName,
+		})
+	}
+	return joinedHosts, nil
+}
+
+// combineDeltas combines two deltas into a single delta.
+// Assumes that the order of operations is add(1), remove(1), add(2), remove(2).
+// Removes duplicate entries and redundant operations from each delta.
+func combineDeltas(adds1, removes1, adds2, removes2 []string) (adds, removes []string) {
+	addSet := map[string]bool{}
+	removeSet := map[string]bool{}
+
+	// combine processes each unique value in a list.
+	// If the value is in the removeFrom set then it is removed from that set.
+	// Otherwise it is added to the addTo set.
+	combine := func(values []string, removeFrom, addTo map[string]bool) {
+		processed := map[string]bool{}
+		for _, value := range values {
+			if processed[value] {
+				continue
+			}
+			processed[value] = true
+			if removeFrom[value] {
+				delete(removeFrom, value)
+			} else {
+				addTo[value] = true
+			}
+		}
+	}
+
+	combine(adds1, nil, addSet)
+	combine(removes1, addSet, removeSet)
+	combine(adds2, removeSet, addSet)
+	combine(removes2, addSet, removeSet)
+
+	for value := range addSet {
+		adds = append(adds, value)
+	}
+	for value := range removeSet {
+		removes = append(removes, value)
+	}
+	return
+}
+
+// lookupStateEvents looks up the state events that are added by a new event.
+func (s *OutputRoomEventConsumer) lookupStateEvents(
+	addsStateEventIDs []string, event *gomatrixserverlib.Event,
+) ([]*gomatrixserverlib.Event, error) {
+	// Fast path if there aren't any new state events.
+	if len(addsStateEventIDs) == 0 {
+		return nil, nil
+	}
+
+	// Fast path if the only state event added is the event itself.
+	if len(addsStateEventIDs) == 1 && addsStateEventIDs[0] == event.EventID() {
+		return []*gomatrixserverlib.Event{event}, nil
+	}
+
+	missing := addsStateEventIDs
+	var result []*gomatrixserverlib.Event
+
+	// Check if event itself is being added.
+	for _, eventID := range missing {
+		if eventID == event.EventID() {
+			result = append(result, event)
+			break
+		}
+	}
+	missing = missingEventsFrom(result, addsStateEventIDs)
+
+	if len(missing) == 0 {
+		return result, nil
+	}
+
+	// At this point the missing events are neither the event itself nor are
+	// they present in our local database. Our only option is to fetch them
+	// from the roomserver using the query API.
+	eventReq := api.QueryEventsByIDRequest{EventIDs: missing}
+	var eventResp api.QueryEventsByIDResponse
+	if err := s.rsAPI.QueryEventsByID(context.TODO(), &eventReq, &eventResp); err != nil {
+		return nil, err
+	}
+
+	for _, headeredEvent := range eventResp.Events {
+		result = append(result, headeredEvent.Event)
+	}
+
+	missing = missingEventsFrom(result, addsStateEventIDs)
+
+	if len(missing) != 0 {
+		return nil, fmt.Errorf(
+			"missing %d state events IDs at event %q", len(missing), event.EventID(),
+		)
+	}
+
+	return result, nil
+}
+
+func missingEventsFrom(events []*gomatrixserverlib.Event, required []string) []string {
+	have := map[string]bool{}
+	for _, event := range events {
+		have[event.EventID()] = true
+	}
+	var missing []string
+	for _, eventID := range required {
+		if !have[eventID] {
+			missing = append(missing, eventID)
+		}
+	}
+	return missing
+}

federationapi/consumers/roomserver_test.go

@@ -0,0 +1,53 @@
+// Copyright 2017 Vector Creations Ltd
+//
+// 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 consumers
+
+import (
+	"testing"
+)
+
+func TestCombineNoOp(t *testing.T) {
+	inputAdd1 := []string{"a", "b", "c"}
+	inputDel1 := []string{"a", "b", "d"}
+	inputAdd2 := []string{"a", "d", "e"}
+	inputDel2 := []string{"a", "c", "e", "e"}
+
+	gotAdd, gotDel := combineDeltas(inputAdd1, inputDel1, inputAdd2, inputDel2)
+
+	if len(gotAdd) != 0 {
+		t.Errorf("wanted combined adds to be an empty list, got %#v", gotAdd)
+	}
+
+	if len(gotDel) != 0 {
+		t.Errorf("wanted combined removes to be an empty list, got %#v", gotDel)
+	}
+}
+
+func TestCombineDedup(t *testing.T) {
+	inputAdd1 := []string{"a", "a"}
+	inputDel1 := []string{"b", "b"}
+	inputAdd2 := []string{"a", "a"}
+	inputDel2 := []string{"b", "b"}
+
+	gotAdd, gotDel := combineDeltas(inputAdd1, inputDel1, inputAdd2, inputDel2)
+
+	if len(gotAdd) != 1 || gotAdd[0] != "a" {
+		t.Errorf("wanted combined adds to be %#v, got %#v", []string{"a"}, gotAdd)
+	}
+
+	if len(gotDel) != 1 || gotDel[0] != "b" {
+		t.Errorf("wanted combined removes to be %#v, got %#v", []string{"b"}, gotDel)
+	}
+}