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>

1 files changed, 11 insertions, 3 deletions

diff --git a/syncapi/types/types.go b/syncapi/types/types.go
index d0efa1bb..d21203b5 100644
--- a/syncapi/types/types.go
+++ b/syncapi/types/types.go
@@ -103,6 +103,7 @@ type StreamingToken struct {
 	AccountDataPosition      StreamPosition
 	DeviceListPosition       StreamPosition
 	NotificationDataPosition StreamPosition
+	PresencePosition         StreamPosition
 }
 
 // This will be used as a fallback by json.Marshal.
@@ -118,11 +119,12 @@ func (s *StreamingToken) UnmarshalText(text []byte) (err error) {
 
 func (t StreamingToken) String() string {
 	posStr := fmt.Sprintf(
-		"s%d_%d_%d_%d_%d_%d_%d_%d",
+		"s%d_%d_%d_%d_%d_%d_%d_%d_%d",
 		t.PDUPosition, t.TypingPosition,
 		t.ReceiptPosition, t.SendToDevicePosition,
 		t.InvitePosition, t.AccountDataPosition,
 		t.DeviceListPosition, t.NotificationDataPosition,
+		t.PresencePosition,
 	)
 	return posStr
 }
@@ -146,12 +148,14 @@ func (t *StreamingToken) IsAfter(other StreamingToken) bool {
 		return true
 	case t.NotificationDataPosition > other.NotificationDataPosition:
 		return true
+	case t.PresencePosition > other.PresencePosition:
+		return true
 	}
 	return false
 }
 
 func (t *StreamingToken) IsEmpty() bool {
-	return t == nil || t.PDUPosition+t.TypingPosition+t.ReceiptPosition+t.SendToDevicePosition+t.InvitePosition+t.AccountDataPosition+t.DeviceListPosition+t.NotificationDataPosition == 0
+	return t == nil || t.PDUPosition+t.TypingPosition+t.ReceiptPosition+t.SendToDevicePosition+t.InvitePosition+t.AccountDataPosition+t.DeviceListPosition+t.NotificationDataPosition+t.PresencePosition == 0
 }
 
 // WithUpdates returns a copy of the StreamingToken with updates applied from another StreamingToken.
@@ -192,6 +196,9 @@ func (t *StreamingToken) ApplyUpdates(other StreamingToken) {
 	if other.NotificationDataPosition > t.NotificationDataPosition {
 		t.NotificationDataPosition = other.NotificationDataPosition
 	}
+	if other.PresencePosition > t.PresencePosition {
+		t.PresencePosition = other.PresencePosition
+	}
 }
 
 type TopologyToken struct {
@@ -284,7 +291,7 @@ func NewStreamTokenFromString(tok string) (token StreamingToken, err error) {
 	// s478_0_0_0_0_13.dl-0-2 but we have now removed partitioned stream positions
 	tok = strings.Split(tok, ".")[0]
 	parts := strings.Split(tok[1:], "_")
-	var positions [8]StreamPosition
+	var positions [9]StreamPosition
 	for i, p := range parts {
 		if i >= len(positions) {
 			break
@@ -306,6 +313,7 @@ func NewStreamTokenFromString(tok string) (token StreamingToken, err error) {
 		AccountDataPosition:      positions[5],
 		DeviceListPosition:       positions[6],
 		NotificationDataPosition: positions[7],
+		PresencePosition:         positions[8],
 	}
 	return token, nil
 }