Add presence module V2 (#2312)

* Syncapi presence

* Clientapi http presence handler

* Why is this here?

* Missing files

* FederationAPI presence implementation

* Add new presence stream

* Pinecone update

* Pinecone update

* Add passing tests

* Make linter happy

* Add presence producer

* Add presence config option

* Set user to unavailable after x minutes

* Only set currently_active if online
Avoid unneeded presence updates when syncing

* Tweaks

* Query devices for last_active_ts
Fixes & tweaks

* Export SharedUsers/SharedUsers

* Presence stream in MemoryStorage

* Remove status_msg_nil

* Fix sytest crashes

* Make presence types const and use stringer for it

* Change options to allow inbound/outbound presence

* Fix option & typo

* Update configs

Co-authored-by: Neil Alexander <neilalexander@users.noreply.github.com>

federationapi/consumers/presence.go

@@ -0,0 +1,143 @@
+// 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"
+	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/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 OutputPresenceConsumer struct {
+	ctx                     context.Context
+	jetstream               nats.JetStreamContext
+	durable                 string
+	db                      storage.Database
+	queues                  *queue.OutgoingQueues
+	ServerName              gomatrixserverlib.ServerName
+	topic                   string
+	outboundPresenceEnabled bool
+}
+
+// NewOutputPresenceConsumer creates a new OutputPresenceConsumer. Call Start() to begin consuming 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),
+		outboundPresenceEnabled: cfg.Matrix.Presence.EnableOutbound,
+	}
+}
+
+// Start consuming from the clientapi
+func (t *OutputPresenceConsumer) Start() error {
+	if !t.outboundPresenceEnabled {
+		return nil
+	}
+	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 presence
+// 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")
+
+	ts, err := strconv.Atoi(msg.Header.Get("last_active_ts"))
+	if err != nil {
+		return true
+	}
+
+	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
+	}
+
+	var statusMsg *string = nil
+	if data, ok := msg.Header["status_msg"]; ok && len(data) > 0 {
+		status := msg.Header.Get("status_msg")
+		statusMsg = &status
+	}
+
+	p := types.PresenceInternal{LastActiveTS: gomatrixserverlib.Timestamp(ts)}
+
+	content := fedTypes.Presence{
+		Push: []fedTypes.PresenceContent{
+			{
+				CurrentlyActive: p.CurrentlyActive(),
+				LastActiveAgo:   p.LastActiveAgo(),
+				Presence:        presence,
+				StatusMsg:       statusMsg,
+				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,20 @@ func (p *SyncAPIProducer) SendTyping(
 	_, err := p.JetStream.PublishMsg(m, nats.Context(ctx))
 	return err
 }
+
+func (p *SyncAPIProducer) SendPresence(
+	ctx context.Context, userID string, presence types.Presence, statusMsg *string, lastActiveAgo int64,
+) error {
+	m := nats.NewMsg(p.TopicPresenceEvent)
+	m.Header.Set(jetstream.UserID, userID)
+	m.Header.Set("presence", presence.String())
+	if statusMsg != nil {
+		m.Header.Set("status_msg", *statusMsg)
+	}
+	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

@@ -30,6 +30,7 @@ import (
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
+	syncTypes "github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
 	"github.com/matrix-org/util"
 	"github.com/prometheus/client_golang/prometheus"
@@ -127,13 +128,14 @@ func Send(
 	defer inFlightTxnsPerOrigin.Delete(index)
 
 	t := txnReq{
-		rsAPI:      rsAPI,
-		keys:       keys,
-		federation: federation,
-		servers:    servers,
-		keyAPI:     keyAPI,
-		roomsMu:    mu,
-		producer:   producer,
+		rsAPI:                  rsAPI,
+		keys:                   keys,
+		federation:             federation,
+		servers:                servers,
+		keyAPI:                 keyAPI,
+		roomsMu:                mu,
+		producer:               producer,
+		inboundPresenceEnabled: cfg.Matrix.Presence.EnableInbound,
 	}
 
 	var txnEvents struct {
@@ -185,13 +187,14 @@ func Send(
 
 type txnReq struct {
 	gomatrixserverlib.Transaction
-	rsAPI      api.RoomserverInternalAPI
-	keyAPI     keyapi.KeyInternalAPI
-	keys       gomatrixserverlib.JSONVerifier
-	federation txnFederationClient
-	roomsMu    *internal.MutexByRoom
-	servers    federationAPI.ServersInRoomProvider
-	producer   *producers.SyncAPIProducer
+	rsAPI                  api.RoomserverInternalAPI
+	keyAPI                 keyapi.KeyInternalAPI
+	keys                   gomatrixserverlib.JSONVerifier
+	federation             txnFederationClient
+	roomsMu                *internal.MutexByRoom
+	servers                federationAPI.ServersInRoomProvider
+	producer               *producers.SyncAPIProducer
+	inboundPresenceEnabled bool
 }
 
 // A subset of FederationClient functionality that txn requires. Useful for testing.
@@ -389,12 +392,37 @@ 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 t.inboundPresenceEnabled {
+				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 {
+		presence, ok := syncTypes.PresenceFromString(content.Presence)
+		if !ok {
+			logrus.Warnf("invalid presence '%s', skipping.", content.Presence)
+			continue
+		}
+		if err := t.producer.SendPresence(ctx, content.UserID, 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"`
+}