Refactor notifications (#2688)

This PR changes the handling of notifications
- removes the `StreamEvent` and `ReadUpdate` stream
- listens on the `OutputRoomEvent` stream in the UserAPI to inform the
SyncAPI about unread notifications
- listens on the `OutputReceiptEvent` stream in the UserAPI to set
receipts/update notifications
- sets the `read_markers` directly from within the internal UserAPI

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

2 files changed, 0 insertions, 122 deletions

diff --git a/syncapi/producers/userapi_readupdate.go b/syncapi/producers/userapi_readupdate.go
deleted file mode 100644
index d56cab77..00000000
--- a/syncapi/producers/userapi_readupdate.go
+++ /dev/null
@@ -1,62 +0,0 @@
-// Copyright 2017 Vector Creations Ltd
-//
-// 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 (
-	"encoding/json"
-
-	"github.com/matrix-org/dendrite/setup/jetstream"
-	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/nats-io/nats.go"
-	log "github.com/sirupsen/logrus"
-)
-
-// UserAPIProducer produces events for the user API server to consume
-type UserAPIReadProducer struct {
-	Topic     string
-	JetStream nats.JetStreamContext
-}
-
-// SendData sends account data to the user API server
-func (p *UserAPIReadProducer) SendReadUpdate(userID, roomID string, readPos, fullyReadPos types.StreamPosition) error {
-	m := &nats.Msg{
-		Subject: p.Topic,
-		Header:  nats.Header{},
-	}
-	m.Header.Set(jetstream.UserID, userID)
-	m.Header.Set(jetstream.RoomID, roomID)
-
-	data := types.ReadUpdate{
-		UserID:    userID,
-		RoomID:    roomID,
-		Read:      readPos,
-		FullyRead: fullyReadPos,
-	}
-	var err error
-	m.Data, err = json.Marshal(data)
-	if err != nil {
-		return err
-	}
-
-	log.WithFields(log.Fields{
-		"user_id":        userID,
-		"room_id":        roomID,
-		"read_pos":       readPos,
-		"fully_read_pos": fullyReadPos,
-	}).Tracef("Producing to topic '%s'", p.Topic)
-
-	_, err = p.JetStream.PublishMsg(m)
-	return err
-}
diff --git a/syncapi/producers/userapi_streamevent.go b/syncapi/producers/userapi_streamevent.go
deleted file mode 100644
index 2bbd19c0..00000000
--- a/syncapi/producers/userapi_streamevent.go
+++ /dev/null
@@ -1,60 +0,0 @@
-// Copyright 2017 Vector Creations Ltd
-//
-// 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 (
-	"encoding/json"
-
-	"github.com/matrix-org/dendrite/setup/jetstream"
-	"github.com/matrix-org/dendrite/syncapi/types"
-	"github.com/matrix-org/gomatrixserverlib"
-	"github.com/nats-io/nats.go"
-	log "github.com/sirupsen/logrus"
-)
-
-// UserAPIProducer produces events for the user API server to consume
-type UserAPIStreamEventProducer struct {
-	Topic     string
-	JetStream nats.JetStreamContext
-}
-
-// SendData sends account data to the user API server
-func (p *UserAPIStreamEventProducer) SendStreamEvent(roomID string, event *gomatrixserverlib.HeaderedEvent, pos types.StreamPosition) error {
-	m := &nats.Msg{
-		Subject: p.Topic,
-		Header:  nats.Header{},
-	}
-	m.Header.Set(jetstream.RoomID, roomID)
-
-	data := types.StreamedEvent{
-		Event:          event,
-		StreamPosition: pos,
-	}
-	var err error
-	m.Data, err = json.Marshal(data)
-	if err != nil {
-		return err
-	}
-
-	log.WithFields(log.Fields{
-		"room_id":    roomID,
-		"event_id":   event.EventID(),
-		"event_type": event.Type(),
-		"stream_pos": pos,
-	}).Tracef("Producing to topic '%s'", p.Topic)
-
-	_, err = p.JetStream.PublishMsg(m)
-	return err
-}