aboutsummaryrefslogtreecommitdiff
path: root/federationapi
diff options
context:
space:
mode:
Diffstat (limited to 'federationapi')
-rw-r--r--federationapi/consumers/eduserver.go257
-rw-r--r--federationapi/consumers/keychange.go4
-rw-r--r--federationapi/consumers/receipts.go141
-rw-r--r--federationapi/consumers/sendtodevice.go125
-rw-r--r--federationapi/consumers/typing.go119
-rw-r--r--federationapi/federationapi.go41
-rw-r--r--federationapi/federationapi_test.go2
-rw-r--r--federationapi/producers/syncapi.go144
-rw-r--r--federationapi/routing/routing.go6
-rw-r--r--federationapi/routing/send.go39
-rw-r--r--federationapi/routing/send_test.go40
-rw-r--r--federationapi/types/types.go15
12 files changed, 597 insertions, 336 deletions
diff --git a/federationapi/consumers/eduserver.go b/federationapi/consumers/eduserver.go
deleted file mode 100644
index e14e60f4..00000000
--- a/federationapi/consumers/eduserver.go
+++ /dev/null
@@ -1,257 +0,0 @@
-// 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 {
- ctx context.Context
- jetstream nats.JetStreamContext
- durable string
- 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{
- ctx: process.Context(),
- jetstream: js,
- queues: queues,
- db: store,
- ServerName: cfg.Matrix.ServerName,
- durable: cfg.Matrix.JetStream.Durable("FederationAPIEDUServerConsumer"),
- typingTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
- sendToDeviceTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
- receiptTopic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
- }
-}
-
-// Start consuming from EDU servers
-func (t *OutputEDUConsumer) Start() error {
- if err := jetstream.JetStreamConsumer(
- t.ctx, t.jetstream, t.typingTopic, t.durable, t.onTypingEvent,
- nats.DeliverAll(), nats.ManualAck(),
- ); err != nil {
- return err
- }
- if err := jetstream.JetStreamConsumer(
- t.ctx, t.jetstream, t.sendToDeviceTopic, t.durable, t.onSendToDeviceEvent,
- nats.DeliverAll(), nats.ManualAck(),
- ); err != nil {
- return err
- }
- if err := jetstream.JetStreamConsumer(
- t.ctx, t.jetstream, t.receiptTopic, t.durable, t.onReceiptEvent,
- nats.DeliverAll(), nats.ManualAck(),
- ); 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(ctx context.Context, msg *nats.Msg) bool {
- // 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)")
- return true
- }
-
- // 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")
- return true
- }
- if originServerName != t.ServerName {
- log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
- return true
- }
-
- _, 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")
- return true
- }
-
- // 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")
- return true
- }
-
- log.Debugf("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")
- return false
- }
-
- return true
-}
-
-// onTypingEvent is called in response to a message received on the typing
-// events topic from the EDU server.
-func (t *OutputEDUConsumer) onTypingEvent(ctx context.Context, msg *nats.Msg) bool {
- // 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 true
- }
-
- // 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 true
- }
- if typingServerName != t.ServerName {
- return true
- }
-
- joined, err := t.db.GetJoinedHosts(ctx, ote.Event.RoomID)
- if err != nil {
- log.WithError(err).WithField("room_id", ote.Event.RoomID).Error("failed to get joined hosts for room")
- return false
- }
-
- 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")
- return true
- }
-
- if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
- log.WithError(err).Error("failed to send EDU")
- return false
- }
-
- return true
-}
-
-// onReceiptEvent is called in response to a message received on the receipt
-// events topic from the EDU server.
-func (t *OutputEDUConsumer) onReceiptEvent(ctx context.Context, msg *nats.Msg) bool {
- // 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)")
- return true
- }
-
- // 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")
- return true
- }
- if receiptServerName != t.ServerName {
- return true
- }
-
- joined, err := t.db.GetJoinedHosts(ctx, receipt.RoomID)
- if err != nil {
- log.WithError(err).WithField("room_id", receipt.RoomID).Error("failed to get joined hosts for room")
- return false
- }
-
- 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")
- return true
- }
-
- if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
- log.WithError(err).Error("failed to send EDU")
- return false
- }
-
- return true
-}
diff --git a/federationapi/consumers/keychange.go b/federationapi/consumers/keychange.go
index 94e45435..0ece18e9 100644
--- a/federationapi/consumers/keychange.go
+++ b/federationapi/consumers/keychange.go
@@ -18,9 +18,9 @@ import (
"context"
"encoding/json"
- 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/federationapi/types"
"github.com/matrix-org/dendrite/keyserver/api"
roomserverAPI "github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/setup/config"
@@ -190,7 +190,7 @@ func (t *KeyChangeConsumer) onCrossSigningMessage(m api.DeviceMessage) bool {
// Pack the EDU and marshal it
edu := &gomatrixserverlib.EDU{
- Type: eduserverAPI.MSigningKeyUpdate,
+ Type: types.MSigningKeyUpdate,
Origin: string(t.serverName),
}
if edu.Content, err = json.Marshal(output); err != nil {
diff --git a/federationapi/consumers/receipts.go b/federationapi/consumers/receipts.go
new file mode 100644
index 00000000..9300451e
--- /dev/null
+++ b/federationapi/consumers/receipts.go
@@ -0,0 +1,141 @@
+// 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 consumers
+
+import (
+ "context"
+ "encoding/json"
+ "strconv"
+
+ "github.com/getsentry/sentry-go"
+ "github.com/matrix-org/dendrite/federationapi/queue"
+ "github.com/matrix-org/dendrite/federationapi/storage"
+ fedTypes "github.com/matrix-org/dendrite/federationapi/types"
+ "github.com/matrix-org/dendrite/setup/config"
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
+ syncTypes "github.com/matrix-org/dendrite/syncapi/types"
+ "github.com/matrix-org/gomatrixserverlib"
+ "github.com/nats-io/nats.go"
+ log "github.com/sirupsen/logrus"
+)
+
+// OutputReceiptConsumer consumes events that originate in the clientapi.
+type OutputReceiptConsumer struct {
+ ctx context.Context
+ jetstream nats.JetStreamContext
+ durable string
+ db storage.Database
+ queues *queue.OutgoingQueues
+ ServerName gomatrixserverlib.ServerName
+ topic string
+}
+
+// NewOutputReceiptConsumer creates a new OutputReceiptConsumer. Call Start() to begin consuming typing events.
+func NewOutputReceiptConsumer(
+ process *process.ProcessContext,
+ cfg *config.FederationAPI,
+ js nats.JetStreamContext,
+ queues *queue.OutgoingQueues,
+ store storage.Database,
+) *OutputReceiptConsumer {
+ return &OutputReceiptConsumer{
+ ctx: process.Context(),
+ jetstream: js,
+ queues: queues,
+ db: store,
+ ServerName: cfg.Matrix.ServerName,
+ durable: cfg.Matrix.JetStream.Durable("FederationAPIReceiptConsumer"),
+ topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
+ }
+}
+
+// Start consuming from the clientapi
+func (t *OutputReceiptConsumer) Start() error {
+ return jetstream.JetStreamConsumer(
+ t.ctx, t.jetstream, t.topic, t.durable, t.onMessage,
+ nats.DeliverAll(), nats.ManualAck(), nats.HeadersOnly(),
+ )
+}
+
+// onMessage is called in response to a message received on the receipt
+// events topic from the client api.
+func (t *OutputReceiptConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
+ receipt := syncTypes.OutputReceiptEvent{
+ UserID: msg.Header.Get(jetstream.UserID),
+ RoomID: msg.Header.Get(jetstream.RoomID),
+ EventID: msg.Header.Get(jetstream.EventID),
+ Type: msg.Header.Get("type"),
+ }
+
+ // 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")
+ return true
+ }
+ if receiptServerName != t.ServerName {
+ return true
+ }
+
+ timestamp, err := strconv.Atoi(msg.Header.Get("timestamp"))
+ if err != nil {
+ // If the message was invalid, log it and move on to the next message in the stream
+ log.WithError(err).Errorf("EDU output log: message parse failure")
+ sentry.CaptureException(err)
+ return true
+ }
+
+ receipt.Timestamp = gomatrixserverlib.Timestamp(timestamp)
+
+ joined, err := t.db.GetJoinedHosts(ctx, receipt.RoomID)
+ if err != nil {
+ log.WithError(err).WithField("room_id", receipt.RoomID).Error("failed to get joined hosts for room")
+ return false
+ }
+
+ names := make([]gomatrixserverlib.ServerName, len(joined))
+ for i := range joined {
+ names[i] = joined[i].ServerName
+ }
+
+ content := map[string]fedTypes.FederationReceiptMRead{}
+ content[receipt.RoomID] = fedTypes.FederationReceiptMRead{
+ User: map[string]fedTypes.FederationReceiptData{
+ receipt.UserID: {
+ Data: fedTypes.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")
+ return true
+ }
+
+ if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
+ log.WithError(err).Error("failed to send EDU")
+ return false
+ }
+
+ return true
+}
diff --git a/federationapi/consumers/sendtodevice.go b/federationapi/consumers/sendtodevice.go
new file mode 100644
index 00000000..84c9f620
--- /dev/null
+++ b/federationapi/consumers/sendtodevice.go
@@ -0,0 +1,125 @@
+// 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 consumers
+
+import (
+ "context"
+ "encoding/json"
+
+ "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"
+ syncTypes "github.com/matrix-org/dendrite/syncapi/types"
+ "github.com/matrix-org/gomatrixserverlib"
+ "github.com/matrix-org/util"
+ "github.com/nats-io/nats.go"
+ log "github.com/sirupsen/logrus"
+)
+
+// OutputSendToDeviceConsumer consumes events that originate in the clientapi.
+type OutputSendToDeviceConsumer struct {
+ ctx context.Context
+ jetstream nats.JetStreamContext
+ durable string
+ db storage.Database
+ queues *queue.OutgoingQueues
+ ServerName gomatrixserverlib.ServerName
+ topic string
+}
+
+// NewOutputSendToDeviceConsumer creates a new OutputSendToDeviceConsumer. Call Start() to begin consuming send-to-device events.
+func NewOutputSendToDeviceConsumer(
+ process *process.ProcessContext,
+ cfg *config.FederationAPI,
+ js nats.JetStreamContext,
+ queues *queue.OutgoingQueues,
+ store storage.Database,
+) *OutputSendToDeviceConsumer {
+ return &OutputSendToDeviceConsumer{
+ ctx: process.Context(),
+ jetstream: js,
+ queues: queues,
+ db: store,
+ ServerName: cfg.Matrix.ServerName,
+ durable: cfg.Matrix.JetStream.Durable("FederationAPIESendToDeviceConsumer"),
+ topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
+ }
+}
+
+// Start consuming from the client api
+func (t *OutputSendToDeviceConsumer) Start() error {
+ return jetstream.JetStreamConsumer(
+ t.ctx, t.jetstream, t.topic, t.durable, t.onMessage,
+ nats.DeliverAll(), nats.ManualAck(),
+ )
+}
+
+// onMessage is called in response to a message received on the
+// send-to-device events topic from the client api.
+func (t *OutputSendToDeviceConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
+ // only send send-to-device events which originated from us
+ sender := msg.Header.Get("sender")
+ _, originServerName, err := gomatrixserverlib.SplitID('@', sender)
+ if err != nil {
+ log.WithError(err).WithField("user_id", sender).Error("Failed to extract domain from send-to-device sender")
+ return true
+ }
+ if originServerName != t.ServerName {
+ log.WithField("other_server", originServerName).Info("Suppressing send-to-device: originated elsewhere")
+ return true
+ }
+ // Extract the send-to-device event from msg.
+ var ote syncTypes.OutputSendToDeviceEvent
+ if err = json.Unmarshal(msg.Data, &ote); err != nil {
+ log.WithError(err).Errorf("output log: message parse failed (expected send-to-device)")
+ return true
+ }
+
+ _, 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")
+ return true
+ }
+
+ // 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")
+ return true
+ }
+
+ log.Debugf("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")
+ return false
+ }
+
+ return true
+}
diff --git a/federationapi/consumers/typing.go b/federationapi/consumers/typing.go
new file mode 100644
index 00000000..428e1a86
--- /dev/null
+++ b/federationapi/consumers/typing.go
@@ -0,0 +1,119 @@
+// 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 consumers
+
+import (
+ "context"
+ "encoding/json"
+ "strconv"
+
+ "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/nats-io/nats.go"
+ log "github.com/sirupsen/logrus"
+)
+
+// OutputTypingConsumer consumes events that originate in the clientapi.
+type OutputTypingConsumer struct {
+ ctx context.Context
+ jetstream nats.JetStreamContext
+ durable string
+ db storage.Database
+ queues *queue.OutgoingQueues
+ ServerName gomatrixserverlib.ServerName
+ topic string
+}
+
+// NewOutputTypingConsumer creates a new OutputTypingConsumer. Call Start() to begin consuming typing events.
+func NewOutputTypingConsumer(
+ process *process.ProcessContext,
+ cfg *config.FederationAPI,
+ js nats.JetStreamContext,
+ queues *queue.OutgoingQueues,
+ store storage.Database,
+) *OutputTypingConsumer {
+ return &OutputTypingConsumer{
+ ctx: process.Context(),
+ jetstream: js,
+ queues: queues,
+ db: store,
+ ServerName: cfg.Matrix.ServerName,
+ durable: cfg.Matrix.JetStream.Durable("FederationAPITypingConsumer"),
+ topic: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
+ }
+}
+
+// Start consuming from the clientapi
+func (t *OutputTypingConsumer) Start() error {
+ return jetstream.JetStreamConsumer(
+ t.ctx, t.jetstream, t.topic, t.durable, t.onMessage,
+ nats.DeliverAll(), nats.ManualAck(), nats.HeadersOnly(),
+ )
+}
+
+// onMessage is called in response to a message received on the typing
+// events topic from the client api.
+func (t *OutputTypingConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
+ // Extract the typing event from msg.
+ roomID := msg.Header.Get(jetstream.RoomID)
+ userID := msg.Header.Get(jetstream.UserID)
+ typing, err := strconv.ParseBool(msg.Header.Get("typing"))
+ if err != nil {
+ log.WithError(err).Errorf("EDU output log: typing parse failure")
+ return true
+ }
+
+ // only send typing events which originated from us
+ _, typingServerName, err := gomatrixserverlib.SplitID('@', userID)
+ if err != nil {
+ log.WithError(err).WithField("user_id", userID).Error("Failed to extract domain from typing sender")
+ _ = msg.Ack()
+ return true
+ }
+ if typingServerName != t.ServerName {
+ return true
+ }
+
+ joined, err := t.db.GetJoinedHosts(ctx, roomID)
+ if err != nil {
+ log.WithError(err).WithField("room_id", roomID).Error("failed to get joined hosts for room")
+ return false
+ }
+
+ names := make([]gomatrixserverlib.ServerName, len(joined))
+ for i := range joined {
+ names[i] = joined[i].ServerName
+ }
+
+ edu := &gomatrixserverlib.EDU{Type: "m.typing"}
+ if edu.Content, err = json.Marshal(map[string]interface{}{
+ "room_id": roomID,
+ "user_id": userID,
+ "typing": typing,
+ }); err != nil {
+ log.WithError(err).Error("failed to marshal EDU JSON")
+ return true
+ }
+ if err := t.queues.SendEDU(edu, t.ServerName, names); err != nil {
+ log.WithError(err).Error("failed to send EDU")
+ return false
+ }
+
+ return true
+}
diff --git a/federationapi/federationapi.go b/federationapi/federationapi.go
index b7f93ecb..8a0ce8e3 100644
--- a/federationapi/federationapi.go
+++ b/federationapi/federationapi.go
@@ -16,12 +16,12 @@ package federationapi
import (
"github.com/gorilla/mux"
- eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/federationapi/api"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
"github.com/matrix-org/dendrite/federationapi/consumers"
"github.com/matrix-org/dendrite/federationapi/internal"
"github.com/matrix-org/dendrite/federationapi/inthttp"
+ "github.com/matrix-org/dendrite/federationapi/producers"
"github.com/matrix-org/dendrite/federationapi/queue"
"github.com/matrix-org/dendrite/federationapi/statistics"
"github.com/matrix-org/dendrite/federationapi/storage"
@@ -31,6 +31,7 @@ import (
"github.com/matrix-org/dendrite/setup/base"
"github.com/matrix-org/dendrite/setup/config"
"github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/setup/process"
userapi "github.com/matrix-org/dendrite/userapi/api"
"github.com/sirupsen/logrus"
@@ -46,6 +47,7 @@ func AddInternalRoutes(router *mux.Router, intAPI api.FederationInternalAPI) {
// AddPublicRoutes sets up and registers HTTP handlers on the base API muxes for the FederationAPI component.
func AddPublicRoutes(
+ process *process.ProcessContext,
fedRouter, keyRouter, wellKnownRouter *mux.Router,
cfg *config.FederationAPI,
userAPI userapi.UserInternalAPI,
@@ -53,16 +55,26 @@ func AddPublicRoutes(
keyRing gomatrixserverlib.JSONVerifier,
rsAPI roomserverAPI.RoomserverInternalAPI,
federationAPI federationAPI.FederationInternalAPI,
- eduAPI eduserverAPI.EDUServerInputAPI,
keyAPI keyserverAPI.KeyInternalAPI,
mscCfg *config.MSCs,
servers federationAPI.ServersInRoomProvider,
) {
+
+ js, _ := jetstream.Prepare(process, &cfg.Matrix.JetStream)
+ producer := &producers.SyncAPIProducer{
+ JetStream: js,
+ TopicReceiptEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
+ TopicSendToDeviceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
+ TopicTypingEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
+ ServerName: cfg.Matrix.ServerName,
+ UserAPI: userAPI,
+ }
+
routing.Setup(
fedRouter, keyRouter, wellKnownRouter, cfg, rsAPI,
- eduAPI, federationAPI, keyRing,
+ federationAPI, keyRing,
federation, userAPI, keyAPI, mscCfg,
- servers,
+ servers, producer,
)
}
@@ -112,17 +124,28 @@ func NewInternalAPI(
if err = rsConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start room server consumer")
}
-
- tsConsumer := consumers.NewOutputEDUConsumer(
+ tsConsumer := consumers.NewOutputSendToDeviceConsumer(
+ base.ProcessContext, cfg, js, queues, federationDB,
+ )
+ if err = tsConsumer.Start(); err != nil {
+ logrus.WithError(err).Panic("failed to start send-to-device consumer")
+ }
+ receiptConsumer := consumers.NewOutputReceiptConsumer(
+ base.ProcessContext, cfg, js, queues, federationDB,
+ )
+ if err = receiptConsumer.Start(); err != nil {
+ logrus.WithError(err).Panic("failed to start receipt consumer")
+ }
+ typingConsumer := consumers.NewOutputTypingConsumer(
base.ProcessContext, cfg, js, queues, federationDB,
)
- if err := tsConsumer.Start(); err != nil {
- logrus.WithError(err).Panic("failed to start typing server consumer")
+ if err = typingConsumer.Start(); err != nil {
+ logrus.WithError(err).Panic("failed to start typing consumer")
}
keyConsumer := consumers.NewKeyChangeConsumer(
base.ProcessContext, &base.Cfg.KeyServer, js, queues, federationDB, rsAPI,
)
- if err := keyConsumer.Start(); err != nil {
+ if err = keyConsumer.Start(); err != nil {
logrus.WithError(err).Panic("failed to start key server consumer")
}
diff --git a/federationapi/federationapi_test.go b/federationapi/federationapi_test.go
index c660f12e..833359c1 100644
--- a/federationapi/federationapi_test.go
+++ b/federationapi/federationapi_test.go
@@ -30,7 +30,7 @@ func TestRoomsV3URLEscapeDoNot404(t *testing.T) {
fsAPI := base.FederationAPIHTTPClient()
// TODO: This is pretty fragile, as if anything calls anything on these nils this test will break.
// Unfortunately, it makes little sense to instantiate these dependencies when we just want to test routing.
- federationapi.AddPublicRoutes(base.PublicFederationAPIMux, base.PublicKeyAPIMux, base.PublicWellKnownAPIMux, &cfg.FederationAPI, nil, nil, keyRing, nil, fsAPI, nil, nil, &cfg.MSCs, nil)
+ federationapi.AddPublicRoutes(base.ProcessContext, base.PublicFederationAPIMux, base.PublicKeyAPIMux, base.PublicWellKnownAPIMux, &cfg.FederationAPI, nil, nil, keyRing, nil, fsAPI, nil, &cfg.MSCs, nil)
baseURL, cancel := test.ListenAndServe(t, base.PublicFederationAPIMux, true)
defer cancel()
serverName := gomatrixserverlib.ServerName(strings.TrimPrefix(baseURL, "https://"))
diff --git a/federationapi/producers/syncapi.go b/federationapi/producers/syncapi.go
new file mode 100644
index 00000000..24acb126
--- /dev/null
+++ b/federationapi/producers/syncapi.go
@@ -0,0 +1,144 @@
+// 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 producers
+
+import (
+ "context"
+ "encoding/json"
+ "strconv"
+
+ "github.com/matrix-org/dendrite/setup/jetstream"
+ "github.com/matrix-org/dendrite/syncapi/types"
+ userapi "github.com/matrix-org/dendrite/userapi/api"
+ "github.com/matrix-org/gomatrixserverlib"
+ "github.com/nats-io/nats.go"
+ log "github.com/sirupsen/logrus"
+)
+
+// SyncAPIProducer produces events for the sync API server to consume
+type SyncAPIProducer struct {
+ TopicReceiptEvent string
+ TopicSendToDeviceEvent string
+ TopicTypingEvent string
+ JetStream nats.JetStreamContext
+ ServerName gomatrixserverlib.ServerName
+ UserAPI userapi.UserInternalAPI
+}
+
+func (p *SyncAPIProducer) SendReceipt(
+ ctx context.Context,
+ userID, roomID, eventID, receiptType string, timestamp gomatrixserverlib.Timestamp,
+) error {
+ m := &nats.Msg{
+ Subject: p.TopicReceiptEvent,
+ Header: nats.Header{},
+ }
+ m.Header.Set(jetstream.UserID, userID)
+ m.Header.Set(jetstream.RoomID, roomID)
+ m.Header.Set(jetstream.EventID, eventID)
+ m.Header.Set("type", receiptType)
+ m.Header.Set("timestamp", strconv.Itoa(int(timestamp)))
+
+ log.WithFields(log.Fields{}).Tracef("Producing to topic '%s'", p.TopicReceiptEvent)
+ _, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
+ return err
+}
+
+func (p *SyncAPIProducer) SendToDevice(
+ ctx context.Context, sender, userID, deviceID, eventType string,
+ message interface{},
+) error {
+ devices := []string{}
+ _, domain, err := gomatrixserverlib.SplitID('@', userID)
+ if err != nil {
+ return err
+ }
+
+ // If the event is targeted locally then we want to expand the wildcard
+ // out into individual device IDs so that we can send them to each respective
+ // device. If the event isn't targeted locally then we can't expand the
+ // wildcard as we don't know about the remote devices, so instead we leave it
+ // as-is, so that the federation sender can send it on with the wildcard intact.
+ if domain == p.ServerName && deviceID == "*" {
+ var res userapi.QueryDevicesResponse
+ err = p.UserAPI.QueryDevices(context.TODO(), &userapi.QueryDevicesRequest{
+ UserID: userID,
+ }, &res)
+ if err != nil {
+ return err
+ }
+ for _, dev := range res.Devices {
+ devices = append(devices, dev.ID)
+ }
+ } else {
+ devices = append(devices, deviceID)
+ }
+
+ js, err := json.Marshal(message)
+ if err != nil {
+ return err
+ }
+
+ log.WithFields(log.Fields{
+ "user_id": userID,
+ "num_devices": len(devices),
+ "type": eventType,
+ }).Tracef("Producing to topic '%s'", p.TopicSendToDeviceEvent)
+ for _, device := range devices {
+ ote := &types.OutputSendToDeviceEvent{
+ UserID: userID,
+ DeviceID: device,
+ SendToDeviceEvent: gomatrixserverlib.SendToDeviceEvent{
+ Sender: sender,
+ Type: eventType,
+ Content: js,
+ },
+ }
+
+ eventJSON, err := json.Marshal(ote)
+ if err != nil {
+ log.WithError(err).Error("sendToDevice failed json.Marshal")
+ return err
+ }
+ m := &nats.Msg{
+ Subject: p.TopicSendToDeviceEvent,
+ Data: eventJSON,
+ Header: nats.Header{},
+ }
+ m.Header.Set("sender", sender)
+ m.Header.Set(jetstream.UserID, userID)
+
+ if _, err = p.JetStream.PublishMsg(m, nats.Context(ctx)); err != nil {
+ log.WithError(err).Error("sendToDevice failed t.Producer.SendMessage")
+ return err
+ }
+ }
+ return nil
+}
+
+func (p *SyncAPIProducer) SendTyping(
+ ctx context.Context, userID, roomID string, typing bool, timeoutMS int64,
+) error {
+ m := &nats.Msg{
+ Subject: p.TopicTypingEvent,
+ Header: nats.Header{},
+ }
+ m.Header.Set(jetstream.UserID, userID)
+ m.Header.Set(jetstream.RoomID, roomID)
+ m.Header.Set("typing", strconv.FormatBool(typing))
+ m.Header.Set("timeout_ms", strconv.Itoa(int(timeoutMS)))
+ _, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
+ return err
+}
diff --git a/federationapi/routing/routing.go b/federationapi/routing/routing.go
index 04c88d95..9e5cdb28 100644
--- a/federationapi/routing/routing.go
+++ b/federationapi/routing/routing.go
@@ -19,8 +19,8 @@ import (
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
- eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
+ "github.com/matrix-org/dendrite/federationapi/producers"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/httputil"
keyserverAPI "github.com/matrix-org/dendrite/keyserver/api"
@@ -44,7 +44,6 @@ func Setup(
fedMux, keyMux, wkMux *mux.Router,
cfg *config.FederationAPI,
rsAPI roomserverAPI.RoomserverInternalAPI,
- eduAPI eduserverAPI.EDUServerInputAPI,
fsAPI federationAPI.FederationInternalAPI,
keys gomatrixserverlib.JSONVerifier,
federation *gomatrixserverlib.FederationClient,
@@ -52,6 +51,7 @@ func Setup(
keyAPI keyserverAPI.KeyInternalAPI,
mscCfg *config.MSCs,
servers federationAPI.ServersInRoomProvider,
+ producer *producers.SyncAPIProducer,
) {
v2keysmux := keyMux.PathPrefix("/v2").Subrouter()
v1fedmux := fedMux.PathPrefix("/v1").Subrouter()
@@ -116,7 +116,7 @@ func Setup(
func(httpReq *http.Request, request *gomatrixserverlib.FederationRequest, vars map[string]string) util.JSONResponse {
return Send(
httpReq, request, gomatrixserverlib.TransactionID(vars["txnID"]),
- cfg, rsAPI, eduAPI, keyAPI, keys, federation, mu, servers,
+ cfg, rsAPI, keyAPI, keys, federation, mu, servers, producer,
)
},
)).Methods(http.MethodPut, http.MethodOptions)
diff --git a/federationapi/routing/send.go b/federationapi/routing/send.go
index 745e36de..eacc76db 100644
--- a/federationapi/routing/send.go
+++ b/federationapi/routing/send.go
@@ -23,8 +23,9 @@ import (
"time"
"github.com/matrix-org/dendrite/clientapi/jsonerror"
- eduserverAPI "github.com/matrix-org/dendrite/eduserver/api"
federationAPI "github.com/matrix-org/dendrite/federationapi/api"
+ "github.com/matrix-org/dendrite/federationapi/producers"
+ "github.com/matrix-org/dendrite/federationapi/types"
"github.com/matrix-org/dendrite/internal"
keyapi "github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/roomserver/api"
@@ -87,12 +88,12 @@ func Send(
txnID gomatrixserverlib.TransactionID,
cfg *config.FederationAPI,
rsAPI api.RoomserverInternalAPI,
- eduAPI eduserverAPI.EDUServerInputAPI,
keyAPI keyapi.KeyInternalAPI,
keys gomatrixserverlib.JSONVerifier,
federation *gomatrixserverlib.FederationClient,
mu *internal.MutexByRoom,
servers federationAPI.ServersInRoomProvider,
+ producer *producers.SyncAPIProducer,
) util.JSONResponse {
// First we should check if this origin has already submitted this
// txn ID to us. If they have and the txnIDs map contains an entry,
@@ -127,12 +128,12 @@ func Send(
t := txnReq{
rsAPI: rsAPI,
- eduAPI: eduAPI,
keys: keys,
federation: federation,
servers: servers,
keyAPI: keyAPI,
roomsMu: mu,
+ producer: producer,
}
var txnEvents struct {
@@ -185,12 +186,12 @@ func Send(
type txnReq struct {
gomatrixserverlib.Transaction
rsAPI api.RoomserverInternalAPI
- eduAPI eduserverAPI.EDUServerInputAPI
keyAPI keyapi.KeyInternalAPI
keys gomatrixserverlib.JSONVerifier
federation txnFederationClient
roomsMu *internal.MutexByRoom
servers federationAPI.ServersInRoomProvider
+ producer *producers.SyncAPIProducer
}
// A subset of FederationClient functionality that txn requires. Useful for testing.
@@ -329,8 +330,8 @@ func (t *txnReq) processEDUs(ctx context.Context) {
util.GetLogger(ctx).Debugf("Dropping typing event where sender domain (%q) doesn't match origin (%q)", domain, t.Origin)
continue
}
- if err := eduserverAPI.SendTyping(ctx, t.eduAPI, typingPayload.UserID, typingPayload.RoomID, typingPayload.Typing, 30*1000); err != nil {
- util.GetLogger(ctx).WithError(err).Error("Failed to send typing event to edu server")
+ if err := t.producer.SendTyping(ctx, typingPayload.UserID, typingPayload.RoomID, typingPayload.Typing, 30*1000); err != nil {
+ util.GetLogger(ctx).WithError(err).Error("Failed to send typing event to JetStream")
}
case gomatrixserverlib.MDirectToDevice:
// https://matrix.org/docs/spec/server_server/r0.1.3#m-direct-to-device-schema
@@ -342,12 +343,12 @@ func (t *txnReq) processEDUs(ctx context.Context) {
for userID, byUser := range directPayload.Messages {
for deviceID, message := range byUser {
// TODO: check that the user and the device actually exist here
- if err := eduserverAPI.SendToDevice(ctx, t.eduAPI, directPayload.Sender, userID, deviceID, directPayload.Type, message); err != nil {
+ if err := t.producer.SendToDevice(ctx, directPayload.Sender, userID, deviceID, directPayload.Type, message); err != nil {
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
"sender": directPayload.Sender,
"user_id": userID,
"device_id": deviceID,
- }).Error("Failed to send send-to-device event to edu server")
+ }).Error("Failed to send send-to-device event to JetStream")
}
}
}
@@ -355,7 +356,7 @@ func (t *txnReq) processEDUs(ctx context.Context) {
t.processDeviceListUpdate(ctx, e)
case gomatrixserverlib.MReceipt:
// https://matrix.org/docs/spec/server_server/r0.1.4#receipts
- payload := map[string]eduserverAPI.FederationReceiptMRead{}
+ payload := map[string]types.FederationReceiptMRead{}
if err := json.Unmarshal(e.Content, &payload); err != nil {
util.GetLogger(ctx).WithError(err).Debug("Failed to unmarshal receipt event")
@@ -379,12 +380,12 @@ func (t *txnReq) processEDUs(ctx context.Context) {
"user_id": userID,
"room_id": roomID,
"events": mread.EventIDs,
- }).Error("Failed to send receipt event to edu server")
+ }).Error("Failed to send receipt event to JetStream")
continue
}
}
}
- case eduserverAPI.MSigningKeyUpdate:
+ case types.MSigningKeyUpdate:
if err := t.processSigningKeyUpdate(ctx, e); err != nil {
logrus.WithError(err).Errorf("Failed to process signing key update")
}
@@ -395,7 +396,7 @@ func (t *txnReq) processEDUs(ctx context.Context) {
}
func (t *txnReq) processSigningKeyUpdate(ctx context.Context, e gomatrixserverlib.EDU) error {
- var updatePayload eduserverAPI.CrossSigningKeyUpdate
+ var updatePayload keyapi.CrossSigningKeyUpdate
if err := json.Unmarshal(e.Content, &updatePayload); err != nil {
util.GetLogger(ctx).WithError(err).WithFields(logrus.Fields{
"user_id": updatePayload.UserID,
@@ -422,7 +423,7 @@ func (t *txnReq) processSigningKeyUpdate(ctx context.Context, e gomatrixserverli
return nil
}
-// processReceiptEvent sends receipt events to the edu server
+// processReceiptEvent sends receipt events to JetStream
func (t *txnReq) processReceiptEvent(ctx context.Context,
userID, roomID, receiptType string,
timestamp gomatrixserverlib.Timestamp,
@@ -430,17 +431,7 @@ func (t *txnReq) processReceiptEvent(ctx context.Context,
) error {
// store every event
for _, eventID := range eventIDs {
- req := eduserverAPI.InputReceiptEventRequest{
- InputReceiptEvent: eduserverAPI.InputReceiptEvent{
- UserID: userID,
- RoomID: roomID,
- EventID: eventID,
- Type: receiptType,
- Timestamp: timestamp,
- },
- }
- resp := eduserverAPI.InputReceiptEventResponse{}
- if err := t.eduAPI.InputReceiptEvent(ctx, &req, &resp); err != nil {
+ if err := t.producer.SendReceipt(ctx, userID, roomID, eventID, receiptType, timestamp); err != nil {
return fmt.Errorf("unable to set receipt event: %w", err)
}
}
diff --git a/federationapi/routing/send_test.go b/federationapi/routing/send_test.go
index 4280643e..8d2d8504 100644
--- a/federationapi/routing/send_test.go
+++ b/federationapi/routing/send_test.go
@@ -7,7 +7,6 @@ import (
"testing"
"time"
- eduAPI "github.com/matrix-org/dendrite/eduserver/api"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/test"
"github.com/matrix-org/dendrite/roomserver/api"
@@ -53,44 +52,6 @@ func init() {
}
}
-type testEDUProducer struct {
- // this producer keeps track of calls to InputTypingEvent
- invocations []eduAPI.InputTypingEventRequest
-}
-
-func (p *testEDUProducer) InputTypingEvent(
- ctx context.Context,
- request *eduAPI.InputTypingEventRequest,
- response *eduAPI.InputTypingEventResponse,
-) error {
- p.invocations = append(p.invocations, *request)
- return nil
-}
-
-func (p *testEDUProducer) InputSendToDeviceEvent(
- ctx context.Context,
- request *eduAPI.InputSendToDeviceEventRequest,
- response *eduAPI.InputSendToDeviceEventResponse,
-) error {
- return nil
-}
-
-func (o *testEDUProducer) InputReceiptEvent(
- ctx context.Context,
- request *eduAPI.InputReceiptEventRequest,
- response *eduAPI.InputReceiptEventResponse,
-) error {
- return nil
-}
-
-func (o *testEDUProducer) InputCrossSigningKeyUpdate(
- ctx context.Context,
- request *eduAPI.InputCrossSigningKeyUpdateRequest,
- response *eduAPI.InputCrossSigningKeyUpdateResponse,
-) error {
- return nil
-}
-
type testRoomserverAPI struct {
api.RoomserverInternalAPITrace
inputRoomEvents []api.InputRoomEvent
@@ -225,7 +186,6 @@ func (c *txnFedClient) LookupMissingEvents(ctx context.Context, s gomatrixserver
func mustCreateTransaction(rsAPI api.RoomserverInternalAPI, fedClient txnFederationClient, pdus []json.RawMessage) *txnReq {
t := &txnReq{
rsAPI: rsAPI,
- eduAPI: &testEDUProducer{},
keys: &test.NopJSONVerifier{},
federation: fedClient,
roomsMu: internal.NewMutexByRoom(),
diff --git a/federationapi/types/types.go b/federationapi/types/types.go
index c486c05c..a28a80b2 100644
--- a/federationapi/types/types.go
+++ b/federationapi/types/types.go
@@ -18,6 +18,8 @@ import (
"github.com/matrix-org/gomatrixserverlib"
)
+const MSigningKeyUpdate = "m.signing_key_update" // TODO: move to gomatrixserverlib
+
// A JoinedHost is a server that is joined to a matrix room.
type JoinedHost struct {
// The MemberEventID of a m.room.member join event.
@@ -51,3 +53,16 @@ type InboundPeek struct {
RenewedTimestamp int64
RenewalInterval int64
}
+
+type FederationReceiptMRead struct {
+ User map[string]FederationReceiptData `json:"m.read"`
+}
+
+type FederationReceiptData struct {
+ Data ReceiptTS `json:"data"`
+ EventIDs []string `json:"event_ids"`
+}
+
+type ReceiptTS struct {
+ TS gomatrixserverlib.Timestamp `json:"ts"`
+}
generated by cgit v1.2.3 (git 2.26.2) at 2025-02-17 22:45:15 +0000