FederationAPI presence implementation

federationapi/consumers/presence.go

@@ -0,0 +1,139 @@
+// 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"
+	"time"
+
+	"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"
+	"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 OutputPresenceConsumer 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 NewOutputPresenceConsumer(
+	process *process.ProcessContext,
+	cfg *config.FederationAPI,
+	js nats.JetStreamContext,
+	queues *queue.OutgoingQueues,
+	store storage.Database,
+) *OutputPresenceConsumer {
+	return &OutputPresenceConsumer{
+		ctx:        process.Context(),
+		jetstream:  js,
+		queues:     queues,
+		db:         store,
+		ServerName: cfg.Matrix.ServerName,
+		durable:    cfg.Matrix.JetStream.Durable("FederationAPIPresenceConsumer"),
+		topic:      cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
+	}
+}
+
+// Start consuming from the clientapi
+func (t *OutputPresenceConsumer) 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 *OutputPresenceConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
+	// only send presence events which originated from us
+	userID := msg.Header.Get(jetstream.UserID)
+	_, serverName, err := gomatrixserverlib.SplitID('@', userID)
+	if err != nil {
+		log.WithError(err).WithField("user_id", userID).Error("failed to extract domain from receipt sender")
+		return true
+	}
+	if serverName != t.ServerName {
+		return true
+	}
+
+	presence := msg.Header.Get("presence")
+	statusMsg := msg.Header.Get("status_msg")
+	nilStatusMsg, _ := strconv.ParseBool(msg.Header.Get("status_msg_nil"))
+	ts, err := strconv.Atoi(msg.Header.Get("last_active_ts"))
+
+	if err != nil {
+		return true
+	}
+
+	timestamp := gomatrixserverlib.Timestamp(ts)
+
+	joined, err := t.db.GetAllJoinedHosts(ctx)
+	if err != nil {
+		log.WithError(err).Error("failed to get joined hosts")
+		return true
+	}
+	if len(joined) == 0 {
+		return true
+	}
+
+	newStatusMsg := &statusMsg
+	if nilStatusMsg {
+		newStatusMsg = nil
+	}
+
+	content := fedTypes.Presence{
+		Push: []fedTypes.PresenceContent{
+			{
+				CurrentlyActive: time.Since(timestamp.Time()).Minutes() < 5,
+				LastActiveAgo:   time.Since(timestamp.Time()).Milliseconds(),
+				Presence:        presence,
+				StatusMsg:       newStatusMsg,
+				UserID:          userID,
+			},
+		},
+	}
+
+	edu := &gomatrixserverlib.EDU{
+		Type:   gomatrixserverlib.MPresence,
+		Origin: string(t.ServerName),
+	}
+	if edu.Content, err = json.Marshal(content); err != nil {
+		log.WithError(err).Error("failed to marshal EDU JSON")
+		return true
+	}
+
+	log.Debugf("sending presence EDU to %d servers", len(joined))
+	if err = t.queues.SendEDU(edu, t.ServerName, joined); err != nil {
+		log.WithError(err).Error("failed to send EDU")
+		return false
+	}
+
+	return true
+}

federationapi/federationapi.go

@@ -66,6 +66,7 @@ func AddPublicRoutes(
 		TopicReceiptEvent:      cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
 		TopicSendToDeviceEvent: cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
 		TopicTypingEvent:       cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
+		TopicPresenceEvent:     cfg.Matrix.JetStream.Prefixed(jetstream.OutputPresenceEvent),
 		ServerName:             cfg.Matrix.ServerName,
 		UserAPI:                userAPI,
 	}
@@ -149,5 +150,11 @@ func NewInternalAPI(
 		logrus.WithError(err).Panic("failed to start key server consumer")
 	}
 
+	presenceConsumer := consumers.NewOutputPresenceConsumer(
+		base.ProcessContext, cfg, js, queues, federationDB,
+	)
+	if err = presenceConsumer.Start(); err != nil {
+		logrus.WithError(err).Panic("failed to start presence consumer")
+	}
 	return internal.NewFederationInternalAPI(federationDB, cfg, rsAPI, federation, stats, caches, queues, keyRing)
 }

federationapi/producers/syncapi.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"encoding/json"
 	"strconv"
+	"time"
 
 	"github.com/matrix-org/dendrite/setup/jetstream"
 	"github.com/matrix-org/dendrite/syncapi/types"
@@ -32,6 +33,7 @@ type SyncAPIProducer struct {
 	TopicReceiptEvent      string
 	TopicSendToDeviceEvent string
 	TopicTypingEvent       string
+	TopicPresenceEvent     string
 	JetStream              nats.JetStreamContext
 	ServerName             gomatrixserverlib.ServerName
 	UserAPI                userapi.UserInternalAPI
@@ -142,3 +144,21 @@ func (p *SyncAPIProducer) SendTyping(
 	_, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
 	return err
 }
+
+func (p *SyncAPIProducer) SendPresence(
+	ctx context.Context, userID, presence string, statusMsg *string, lastActiveAgo int64,
+) error {
+	m := nats.NewMsg(p.TopicPresenceEvent)
+	m.Header.Set(jetstream.UserID, userID)
+	m.Header.Set("presence", presence)
+	if statusMsg != nil {
+		m.Header.Set("status_msg", *statusMsg)
+	}
+	m.Header.Set("status_msg_nil", strconv.FormatBool(statusMsg == nil))
+	lastActiveTS := gomatrixserverlib.AsTimestamp(time.Now().Add(-(time.Duration(lastActiveAgo) * time.Millisecond)))
+
+	m.Header.Set("last_active_ts", strconv.Itoa(int(lastActiveTS)))
+	log.Debugf("Sending presence to syncAPI: %+v", m.Header)
+	_, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
+	return err
+}

federationapi/routing/send.go

@@ -389,12 +389,30 @@ func (t *txnReq) processEDUs(ctx context.Context) {
 			if err := t.processSigningKeyUpdate(ctx, e); err != nil {
 				logrus.WithError(err).Errorf("Failed to process signing key update")
 			}
+		case gomatrixserverlib.MPresence:
+			if err := t.processPresence(ctx, e); err != nil {
+				logrus.WithError(err).Errorf("Failed to process presence update")
+			}
 		default:
 			util.GetLogger(ctx).WithField("type", e.Type).Debug("Unhandled EDU")
 		}
 	}
 }
 
+// processPresence handles m.receipt events
+func (t *txnReq) processPresence(ctx context.Context, e gomatrixserverlib.EDU) error {
+	payload := types.Presence{}
+	if err := json.Unmarshal(e.Content, &payload); err != nil {
+		return err
+	}
+	for _, content := range payload.Push {
+		if err := t.producer.SendPresence(ctx, content.UserID, content.Presence, content.StatusMsg, content.LastActiveAgo); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (t *txnReq) processSigningKeyUpdate(ctx context.Context, e gomatrixserverlib.EDU) error {
 	var updatePayload keyapi.CrossSigningKeyUpdate
 	if err := json.Unmarshal(e.Content, &updatePayload); err != nil {

federationapi/types/types.go

@@ -66,3 +66,15 @@ type FederationReceiptData struct {
 type ReceiptTS struct {
 	TS gomatrixserverlib.Timestamp `json:"ts"`
 }
+
+type Presence struct {
+	Push []PresenceContent `json:"push"`
+}
+
+type PresenceContent struct {
+	CurrentlyActive bool    `json:"currently_active,omitempty"`
+	LastActiveAgo   int64   `json:"last_active_ago"`
+	Presence        string  `json:"presence"`
+	StatusMsg       *string `json:"status_msg,omitempty"`
+	UserID          string  `json:"user_id"`
+}