diff --git a/federationsender/consumers/eduserver.go b/federationsender/consumers/eduserver.go index d9ac41b3b..95e2ee898 100644 --- a/federationsender/consumers/eduserver.go +++ b/federationsender/consumers/eduserver.go @@ -34,6 +34,7 @@ import ( type OutputEDUConsumer struct { typingConsumer *internal.ContinualConsumer sendToDeviceConsumer *internal.ContinualConsumer + receiptConsumer *internal.ContinualConsumer db storage.Database queues *queue.OutgoingQueues ServerName gomatrixserverlib.ServerName @@ -51,24 +52,31 @@ func NewOutputEDUConsumer( c := &OutputEDUConsumer{ typingConsumer: &internal.ContinualConsumer{ ComponentName: "eduserver/typing", - Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputTypingEvent)), + Topic: cfg.Matrix.Kafka.TopicFor(config.TopicOutputTypingEvent), Consumer: kafkaConsumer, PartitionStore: store, }, sendToDeviceConsumer: &internal.ContinualConsumer{ ComponentName: "eduserver/sendtodevice", - Topic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputSendToDeviceEvent)), + Topic: cfg.Matrix.Kafka.TopicFor(config.TopicOutputSendToDeviceEvent), + Consumer: kafkaConsumer, + PartitionStore: store, + }, + receiptConsumer: &internal.ContinualConsumer{ + ComponentName: "eduserver/receipt", + Topic: cfg.Matrix.Kafka.TopicFor(config.TopicOutputReceiptEvent), Consumer: kafkaConsumer, PartitionStore: store, }, queues: queues, db: store, ServerName: cfg.Matrix.ServerName, - TypingTopic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputTypingEvent)), - SendToDeviceTopic: string(cfg.Matrix.Kafka.TopicFor(config.TopicOutputSendToDeviceEvent)), + TypingTopic: cfg.Matrix.Kafka.TopicFor(config.TopicOutputTypingEvent), + SendToDeviceTopic: cfg.Matrix.Kafka.TopicFor(config.TopicOutputSendToDeviceEvent), } c.typingConsumer.ProcessMessage = c.onTypingEvent c.sendToDeviceConsumer.ProcessMessage = c.onSendToDeviceEvent + c.receiptConsumer.ProcessMessage = c.onReceiptEvent return c } @@ -81,6 +89,9 @@ func (t *OutputEDUConsumer) Start() error { if err := t.sendToDeviceConsumer.Start(); err != nil { return fmt.Errorf("t.sendToDeviceConsumer.Start: %w", err) } + if err := t.receiptConsumer.Start(); err != nil { + return fmt.Errorf("t.receiptConsumer.Start: %w", err) + } return nil } @@ -177,3 +188,66 @@ func (t *OutputEDUConsumer) onTypingEvent(msg *sarama.ConsumerMessage) error { return t.queues.SendEDU(edu, t.ServerName, names) } + +type userData struct { + Data struct { + Ts gomatrixserverlib.Timestamp `json:"ts"` + } `json:"data"` + EventIds []string `json:"event_ids"` +} + +// onReceiptEvent is called in response to a message received on the receipt +// events topic from the EDU server. +func (t *OutputEDUConsumer) onReceiptEvent(msg *sarama.ConsumerMessage) error { + // Extract the typing event from msg. + var receipt api.OutputReceiptEvent + if err := json.Unmarshal(msg.Value, &receipt); err != nil { + // Skip this msg but continue processing messages. + log.WithError(err).Errorf("eduserver output log: message parse failed (expected receipt)") + return nil + } + + // 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 nil + } + if receiptServerName != t.ServerName { + log.WithField("other_server", receiptServerName).Info("Suppressing receipt notif: originated elsewhere") + return nil + } + + joined, err := t.db.GetJoinedHosts(context.TODO(), receipt.RoomID) + if err != nil { + return err + } + + names := make([]gomatrixserverlib.ServerName, len(joined)) + for i := range joined { + names[i] = joined[i].ServerName + } + + // TODO: easier/nicer creation of receipt EDUs + content := map[string]interface{}{} + content[receipt.RoomID] = map[string]interface{}{ + "m.read": map[string]interface{}{ + receipt.UserID: userData{ + Data: struct { + Ts gomatrixserverlib.Timestamp `json:"ts"` + }{receipt.Timestamp}, + EventIds: []string{receipt.EventID}, + }, + }, + } + + edu := &gomatrixserverlib.EDU{ + Type: "m.receipt", + Origin: string(t.ServerName), + } + if edu.Content, err = json.Marshal(content); err != nil { + return err + } + + return t.queues.SendEDU(edu, t.ServerName, names) +} diff --git a/sytest-whitelist b/sytest-whitelist index 820e93435..f31b5f0b7 100644 --- a/sytest-whitelist +++ b/sytest-whitelist @@ -487,4 +487,5 @@ Inbound federation rejects typing notifications from wrong remote POST /rooms/:room_id/receipt can create receipts Receipts must be m.read Read receipts appear in initial v2 /sync -New read receipts appear in incremental v2 /sync \ No newline at end of file +New read receipts appear in incremental v2 /sync +Outbound federation sends receipts \ No newline at end of file