Remove eduserver (#2306)

* Move receipt sending to own JetStream producer

* Move SendToDevice to producer

* Remove most parts of the EDU server

* Fix SendToDevice & copyrights

* Move structs, cleanup EDU Server traces

* Use HeadersOnly subscription

* Missing file

* Fix linter issues

* Move consumers to own files

* Rename durable consumer; Consumer cleanup

* Docs/config cleanup

14 files changed, 106 insertions, 75 deletions

diff --git a/syncapi/consumers/clientapi.go b/syncapi/consumers/clientapi.go
index 40c1cd3d..c28da460 100644
--- a/syncapi/consumers/clientapi.go
+++ b/syncapi/consumers/clientapi.go
@@ -62,7 +62,7 @@ func NewOutputClientDataConsumer(
 		ctx:        process.Context(),
 		jetstream:  js,
 		topic:      cfg.Matrix.JetStream.Prefixed(jetstream.OutputClientData),
-		durable:    cfg.Matrix.JetStream.Durable("SyncAPIClientAPIConsumer"),
+		durable:    cfg.Matrix.JetStream.Durable("SyncAPIAccountDataConsumer"),
 		db:         store,
 		notifier:   notifier,
 		stream:     stream,
diff --git a/syncapi/consumers/eduserver_receipts.go b/syncapi/consumers/receipts.go
index ab79998e..6bb0747f 100644
--- a/syncapi/consumers/eduserver_receipts.go
+++ b/syncapi/consumers/receipts.go
@@ -17,11 +17,10 @@ package consumers
 import (
 	"context"
 	"database/sql"
-	"encoding/json"
 	"fmt"
+	"strconv"
 
 	"github.com/getsentry/sentry-go"
-	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
 	"github.com/matrix-org/dendrite/setup/process"
@@ -63,7 +62,7 @@ func NewOutputReceiptEventConsumer(
 		ctx:        process.Context(),
 		jetstream:  js,
 		topic:      cfg.Matrix.JetStream.Prefixed(jetstream.OutputReceiptEvent),
-		durable:    cfg.Matrix.JetStream.Durable("SyncAPIEDUServerReceiptConsumer"),
+		durable:    cfg.Matrix.JetStream.Durable("SyncAPIReceiptConsumer"),
 		db:         store,
 		notifier:   notifier,
 		stream:     stream,
@@ -72,7 +71,7 @@ func NewOutputReceiptEventConsumer(
 	}
 }
 
-// Start consuming from EDU api
+// Start consuming receipts events.
 func (s *OutputReceiptEventConsumer) Start() error {
 	return jetstream.JetStreamConsumer(
 		s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
@@ -81,14 +80,23 @@ func (s *OutputReceiptEventConsumer) Start() error {
 }
 
 func (s *OutputReceiptEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
-	var output api.OutputReceiptEvent
-	if err := json.Unmarshal(msg.Data, &output); err != nil {
+	output := types.OutputReceiptEvent{
+		UserID:  msg.Header.Get(jetstream.UserID),
+		RoomID:  msg.Header.Get(jetstream.RoomID),
+		EventID: msg.Header.Get(jetstream.EventID),
+		Type:    msg.Header.Get("type"),
+	}
+
+	timestamp, err := strconv.Atoi(msg.Header.Get("timestamp"))
+	if err != nil {
 		// If the message was invalid, log it and move on to the next message in the stream
-		log.WithError(err).Errorf("EDU server output log: message parse failure")
+		log.WithError(err).Errorf("output log: message parse failure")
 		sentry.CaptureException(err)
 		return true
 	}
 
+	output.Timestamp = gomatrixserverlib.Timestamp(timestamp)
+
 	streamPos, err := s.db.StoreReceipt(
 		s.ctx,
 		output.RoomID,
@@ -117,7 +125,7 @@ func (s *OutputReceiptEventConsumer) onMessage(ctx context.Context, msg *nats.Ms
 	return true
 }
 
-func (s *OutputReceiptEventConsumer) sendReadUpdate(ctx context.Context, output api.OutputReceiptEvent) error {
+func (s *OutputReceiptEventConsumer) sendReadUpdate(ctx context.Context, output types.OutputReceiptEvent) error {
 	if output.Type != "m.read" {
 		return nil
 	}
diff --git a/syncapi/consumers/eduserver_sendtodevice.go b/syncapi/consumers/sendtodevice.go
index bdbe7735..0b9153fc 100644
--- a/syncapi/consumers/eduserver_sendtodevice.go
+++ b/syncapi/consumers/sendtodevice.go
@@ -19,7 +19,6 @@ import (
 	"encoding/json"
 
 	"github.com/getsentry/sentry-go"
-	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
 	"github.com/matrix-org/dendrite/setup/jetstream"
 	"github.com/matrix-org/dendrite/setup/process"
@@ -58,7 +57,7 @@ func NewOutputSendToDeviceEventConsumer(
 		ctx:        process.Context(),
 		jetstream:  js,
 		topic:      cfg.Matrix.JetStream.Prefixed(jetstream.OutputSendToDeviceEvent),
-		durable:    cfg.Matrix.JetStream.Durable("SyncAPIEDUServerSendToDeviceConsumer"),
+		durable:    cfg.Matrix.JetStream.Durable("SyncAPISendToDeviceConsumer"),
 		db:         store,
 		serverName: cfg.Matrix.ServerName,
 		notifier:   notifier,
@@ -66,7 +65,7 @@ func NewOutputSendToDeviceEventConsumer(
 	}
 }
 
-// Start consuming from EDU api
+// Start consuming send-to-device events.
 func (s *OutputSendToDeviceEventConsumer) Start() error {
 	return jetstream.JetStreamConsumer(
 		s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
@@ -75,15 +74,8 @@ func (s *OutputSendToDeviceEventConsumer) Start() error {
 }
 
 func (s *OutputSendToDeviceEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
-	var output api.OutputSendToDeviceEvent
-	if err := json.Unmarshal(msg.Data, &output); err != nil {
-		// If the message was invalid, log it and move on to the next message in the stream
-		log.WithError(err).Errorf("EDU server output log: message parse failure")
-		sentry.CaptureException(err)
-		return true
-	}
-
-	_, domain, err := gomatrixserverlib.SplitID('@', output.UserID)
+	userID := msg.Header.Get(jetstream.UserID)
+	_, domain, err := gomatrixserverlib.SplitID('@', userID)
 	if err != nil {
 		sentry.CaptureException(err)
 		return true
@@ -92,12 +84,20 @@ func (s *OutputSendToDeviceEventConsumer) onMessage(ctx context.Context, msg *na
 		return true
 	}
 
+	var output types.OutputSendToDeviceEvent
+	if err = json.Unmarshal(msg.Data, &output); err != nil {
+		// If the message was invalid, log it and move on to the next message in the stream
+		log.WithError(err).Errorf("output log: message parse failure")
+		sentry.CaptureException(err)
+		return true
+	}
+
 	util.GetLogger(context.TODO()).WithFields(log.Fields{
 		"sender":     output.Sender,
 		"user_id":    output.UserID,
 		"device_id":  output.DeviceID,
 		"event_type": output.Type,
-	}).Info("sync API received send-to-device event from EDU server")
+	}).Debugf("sync API received send-to-device event from the clientapi/federationsender")
 
 	streamPos, err := s.db.StoreNewSendForDeviceMessage(
 		s.ctx, output.UserID, output.DeviceID, output.SendToDeviceEvent,
diff --git a/syncapi/consumers/eduserver_typing.go b/syncapi/consumers/typing.go
index c2828c7f..48e484ec 100644
--- a/syncapi/consumers/eduserver_typing.go
+++ b/syncapi/consumers/typing.go
@@ -16,16 +16,14 @@ package consumers
 
 import (
 	"context"
-	"encoding/json"
+	"strconv"
+	"time"
 
-	"github.com/getsentry/sentry-go"
-	"github.com/matrix-org/dendrite/eduserver/api"
-	"github.com/matrix-org/dendrite/eduserver/cache"
+	"github.com/matrix-org/dendrite/internal/caching"
 	"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/notifier"
-	"github.com/matrix-org/dendrite/syncapi/storage"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/nats-io/nats.go"
 	log "github.com/sirupsen/logrus"
@@ -37,7 +35,7 @@ type OutputTypingEventConsumer struct {
 	jetstream nats.JetStreamContext
 	durable   string
 	topic     string
-	eduCache  *cache.EDUCache
+	eduCache  *caching.EDUCache
 	stream    types.StreamProvider
 	notifier  *notifier.Notifier
 }
@@ -48,8 +46,7 @@ func NewOutputTypingEventConsumer(
 	process *process.ProcessContext,
 	cfg *config.SyncAPI,
 	js nats.JetStreamContext,
-	store storage.Database,
-	eduCache *cache.EDUCache,
+	eduCache *caching.EDUCache,
 	notifier *notifier.Notifier,
 	stream types.StreamProvider,
 ) *OutputTypingEventConsumer {
@@ -57,14 +54,14 @@ func NewOutputTypingEventConsumer(
 		ctx:       process.Context(),
 		jetstream: js,
 		topic:     cfg.Matrix.JetStream.Prefixed(jetstream.OutputTypingEvent),
-		durable:   cfg.Matrix.JetStream.Durable("SyncAPIEDUServerTypingConsumer"),
+		durable:   cfg.Matrix.JetStream.Durable("SyncAPITypingConsumer"),
 		eduCache:  eduCache,
 		notifier:  notifier,
 		stream:    stream,
 	}
 }
 
-// Start consuming from EDU api
+// Start consuming typing events.
 func (s *OutputTypingEventConsumer) Start() error {
 	return jetstream.JetStreamConsumer(
 		s.ctx, s.jetstream, s.topic, s.durable, s.onMessage,
@@ -73,34 +70,40 @@ func (s *OutputTypingEventConsumer) Start() error {
 }
 
 func (s *OutputTypingEventConsumer) onMessage(ctx context.Context, msg *nats.Msg) bool {
-	var output api.OutputTypingEvent
-	if err := json.Unmarshal(msg.Data, &output); err != nil {
-		// If the message was invalid, log it and move on to the next message in the stream
-		log.WithError(err).Errorf("EDU server output log: message parse failure")
-		sentry.CaptureException(err)
+	roomID := msg.Header.Get(jetstream.RoomID)
+	userID := msg.Header.Get(jetstream.UserID)
+	typing, err := strconv.ParseBool(msg.Header.Get("typing"))
+	if err != nil {
+		log.WithError(err).Errorf("output log: typing parse failure")
+		return true
+	}
+	timeout, err := strconv.Atoi(msg.Header.Get("timeout_ms"))
+	if err != nil {
+		log.WithError(err).Errorf("output log: timeout_ms parse failure")
 		return true
 	}
 
 	log.WithFields(log.Fields{
-		"room_id": output.Event.RoomID,
-		"user_id": output.Event.UserID,
-		"typing":  output.Event.Typing,
-	}).Debug("received data from EDU server")
+		"room_id": roomID,
+		"user_id": userID,
+		"typing":  typing,
+		"timeout": timeout,
+	}).Debug("syncapi received EDU data from client api")
 
 	var typingPos types.StreamPosition
-	typingEvent := output.Event
-	if typingEvent.Typing {
+	if typing {
+		expiry := time.Now().Add(time.Duration(timeout) * time.Millisecond)
 		typingPos = types.StreamPosition(
-			s.eduCache.AddTypingUser(typingEvent.UserID, typingEvent.RoomID, output.ExpireTime),
+			s.eduCache.AddTypingUser(userID, roomID, &expiry),
 		)
 	} else {
 		typingPos = types.StreamPosition(
-			s.eduCache.RemoveUser(typingEvent.UserID, typingEvent.RoomID),
+			s.eduCache.RemoveUser(userID, roomID),
 		)
 	}
 
 	s.stream.Advance(typingPos)
-	s.notifier.OnNewTyping(output.Event.RoomID, types.StreamingToken{TypingPosition: typingPos})
+	s.notifier.OnNewTyping(roomID, types.StreamingToken{TypingPosition: typingPos})
 
 	return true
 }
diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go
index b6ac5be1..647fffad 100644
--- a/syncapi/storage/interface.go
+++ b/syncapi/storage/interface.go
@@ -17,7 +17,6 @@ package storage
 import (
 	"context"
 
-	eduAPI "github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/internal/eventutil"
 
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -46,7 +45,7 @@ type Database interface {
 
 	InviteEventsInRange(ctx context.Context, targetUserID string, r types.Range) (map[string]*gomatrixserverlib.HeaderedEvent, map[string]*gomatrixserverlib.HeaderedEvent, error)
 	PeeksInRange(ctx context.Context, userID, deviceID string, r types.Range) (peeks []types.Peek, err error)
-	RoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []eduAPI.OutputReceiptEvent, error)
+	RoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error)
 
 	// AllJoinedUsersInRooms returns a map of room ID to a list of all joined user IDs.
 	AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error)
@@ -136,7 +135,7 @@ type Database interface {
 	// StoreReceipt stores new receipt events
 	StoreReceipt(ctx context.Context, roomId, receiptType, userId, eventId string, timestamp gomatrixserverlib.Timestamp) (pos types.StreamPosition, err error)
 	// GetRoomReceipts gets all receipts for a given roomID
-	GetRoomReceipts(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]eduAPI.OutputReceiptEvent, error)
+	GetRoomReceipts(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]types.OutputReceiptEvent, error)
 
 	// UpsertRoomUnreadNotificationCounts updates the notification statistics about a (user, room) key.
 	UpsertRoomUnreadNotificationCounts(ctx context.Context, userID, roomID string, notificationCount, highlightCount int) (types.StreamPosition, error)
diff --git a/syncapi/storage/postgres/receipt_table.go b/syncapi/storage/postgres/receipt_table.go
index 474d0c02..2a42ffd7 100644
--- a/syncapi/storage/postgres/receipt_table.go
+++ b/syncapi/storage/postgres/receipt_table.go
@@ -21,7 +21,6 @@ import (
 
 	"github.com/lib/pq"
 
-	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
@@ -95,16 +94,16 @@ func (r *receiptStatements) UpsertReceipt(ctx context.Context, txn *sql.Tx, room
 	return
 }
 
-func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []api.OutputReceiptEvent, error) {
+func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
 	var lastPos types.StreamPosition
 	rows, err := r.selectRoomReceipts.QueryContext(ctx, pq.Array(roomIDs), streamPos)
 	if err != nil {
 		return 0, nil, fmt.Errorf("unable to query room receipts: %w", err)
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "SelectRoomReceiptsAfter: rows.close() failed")
-	var res []api.OutputReceiptEvent
+	var res []types.OutputReceiptEvent
 	for rows.Next() {
-		r := api.OutputReceiptEvent{}
+		r := types.OutputReceiptEvent{}
 		var id types.StreamPosition
 		err = rows.Scan(&id, &r.RoomID, &r.Type, &r.UserID, &r.EventID, &r.Timestamp)
 		if err != nil {
diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go
index 9a2dc0d4..349e4452 100644
--- a/syncapi/storage/shared/syncserver.go
+++ b/syncapi/storage/shared/syncserver.go
@@ -20,7 +20,6 @@ import (
 	"encoding/json"
 	"fmt"
 
-	eduAPI "github.com/matrix-org/dendrite/eduserver/api"
 	userapi "github.com/matrix-org/dendrite/userapi/api"
 
 	"github.com/matrix-org/dendrite/internal/eventutil"
@@ -135,7 +134,7 @@ func (d *Database) PeeksInRange(ctx context.Context, userID, deviceID string, r
 	return d.Peeks.SelectPeeksInRange(ctx, nil, userID, deviceID, r)
 }
 
-func (d *Database) RoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []eduAPI.OutputReceiptEvent, error) {
+func (d *Database) RoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
 	return d.Receipts.SelectRoomReceiptsAfter(ctx, roomIDs, streamPos)
 }
 
@@ -972,7 +971,7 @@ func (d *Database) StoreReceipt(ctx context.Context, roomId, receiptType, userId
 	return
 }
 
-func (d *Database) GetRoomReceipts(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]eduAPI.OutputReceiptEvent, error) {
+func (d *Database) GetRoomReceipts(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) ([]types.OutputReceiptEvent, error) {
 	_, receipts, err := d.Receipts.SelectRoomReceiptsAfter(ctx, roomIDs, streamPos)
 	return receipts, err
 }
diff --git a/syncapi/storage/sqlite3/receipt_table.go b/syncapi/storage/sqlite3/receipt_table.go
index 9111a39f..dea05771 100644
--- a/syncapi/storage/sqlite3/receipt_table.go
+++ b/syncapi/storage/sqlite3/receipt_table.go
@@ -20,7 +20,6 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/internal"
 	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/dendrite/syncapi/storage/tables"
@@ -99,7 +98,7 @@ func (r *receiptStatements) UpsertReceipt(ctx context.Context, txn *sql.Tx, room
 }
 
 // SelectRoomReceiptsAfter select all receipts for a given room after a specific timestamp
-func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []api.OutputReceiptEvent, error) {
+func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error) {
 	selectSQL := strings.Replace(selectRoomReceipts, "($2)", sqlutil.QueryVariadicOffset(len(roomIDs), 1), 1)
 	var lastPos types.StreamPosition
 	params := make([]interface{}, len(roomIDs)+1)
@@ -112,9 +111,9 @@ func (r *receiptStatements) SelectRoomReceiptsAfter(ctx context.Context, roomIDs
 		return 0, nil, fmt.Errorf("unable to query room receipts: %w", err)
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "SelectRoomReceiptsAfter: rows.close() failed")
-	var res []api.OutputReceiptEvent
+	var res []types.OutputReceiptEvent
 	for rows.Next() {
-		r := api.OutputReceiptEvent{}
+		r := types.OutputReceiptEvent{}
 		var id types.StreamPosition
 		err = rows.Scan(&id, &r.RoomID, &r.Type, &r.UserID, &r.EventID, &r.Timestamp)
 		if err != nil {
diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go
index 640b7dc3..ba0076e2 100644
--- a/syncapi/storage/tables/interface.go
+++ b/syncapi/storage/tables/interface.go
@@ -18,7 +18,6 @@ import (
 	"context"
 	"database/sql"
 
-	eduAPI "github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/internal/eventutil"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/syncapi/types"
@@ -168,7 +167,7 @@ type Filter interface {
 
 type Receipts interface {
 	UpsertReceipt(ctx context.Context, txn *sql.Tx, roomId, receiptType, userId, eventId string, timestamp gomatrixserverlib.Timestamp) (pos types.StreamPosition, err error)
-	SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []eduAPI.OutputReceiptEvent, error)
+	SelectRoomReceiptsAfter(ctx context.Context, roomIDs []string, streamPos types.StreamPosition) (types.StreamPosition, []types.OutputReceiptEvent, error)
 	SelectMaxReceiptID(ctx context.Context, txn *sql.Tx) (id int64, err error)
 }
 
diff --git a/syncapi/streams/stream_receipt.go b/syncapi/streams/stream_receipt.go
index 35ffd3a1..680f8cd8 100644
--- a/syncapi/streams/stream_receipt.go
+++ b/syncapi/streams/stream_receipt.go
@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 
-	eduAPI "github.com/matrix-org/dendrite/eduserver/api"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
 )
@@ -53,7 +52,7 @@ func (p *ReceiptStreamProvider) IncrementalSync(
 	}
 
 	// Group receipts by room, so we can create one ClientEvent for every room
-	receiptsByRoom := make(map[string][]eduAPI.OutputReceiptEvent)
+	receiptsByRoom := make(map[string][]types.OutputReceiptEvent)
 	for _, receipt := range receipts {
 		receiptsByRoom[receipt.RoomID] = append(receiptsByRoom[receipt.RoomID], receipt)
 	}
@@ -68,15 +67,15 @@ func (p *ReceiptStreamProvider) IncrementalSync(
 			Type:   gomatrixserverlib.MReceipt,
 			RoomID: roomID,
 		}
-		content := make(map[string]eduAPI.ReceiptMRead)
+		content := make(map[string]ReceiptMRead)
 		for _, receipt := range receipts {
 			read, ok := content[receipt.EventID]
 			if !ok {
-				read = eduAPI.ReceiptMRead{
-					User: make(map[string]eduAPI.ReceiptTS),
+				read = ReceiptMRead{
+					User: make(map[string]ReceiptTS),
 				}
 			}
-			read.User[receipt.UserID] = eduAPI.ReceiptTS{TS: receipt.Timestamp}
+			read.User[receipt.UserID] = ReceiptTS{TS: receipt.Timestamp}
 			content[receipt.EventID] = read
 		}
 		ev.Content, err = json.Marshal(content)
@@ -91,3 +90,11 @@ func (p *ReceiptStreamProvider) IncrementalSync(
 
 	return lastPos
 }
+
+type ReceiptMRead struct {
+	User map[string]ReceiptTS `json:"m.read"`
+}
+
+type ReceiptTS struct {
+	TS gomatrixserverlib.Timestamp `json:"ts"`
+}
diff --git a/syncapi/streams/stream_typing.go b/syncapi/streams/stream_typing.go
index 1e7a46bd..e46cd447 100644
--- a/syncapi/streams/stream_typing.go
+++ b/syncapi/streams/stream_typing.go
@@ -4,14 +4,14 @@ import (
 	"context"
 	"encoding/json"
 
-	"github.com/matrix-org/dendrite/eduserver/cache"
+	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/matrix-org/dendrite/syncapi/types"
 	"github.com/matrix-org/gomatrixserverlib"
 )
 
 type TypingStreamProvider struct {
 	StreamProvider
-	EDUCache *cache.EDUCache
+	EDUCache *caching.EDUCache
 }
 
 func (p *TypingStreamProvider) CompleteSync(
diff --git a/syncapi/streams/streams.go b/syncapi/streams/streams.go
index 17951acb..b2273aad 100644
--- a/syncapi/streams/streams.go
+++ b/syncapi/streams/streams.go
@@ -3,7 +3,7 @@ package streams
 import (
 	"context"
 
-	"github.com/matrix-org/dendrite/eduserver/cache"
+	"github.com/matrix-org/dendrite/internal/caching"
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	rsapi "github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/syncapi/storage"
@@ -25,7 +25,7 @@ type Streams struct {
 func NewSyncStreamProviders(
 	d storage.Database, userAPI userapi.UserInternalAPI,
 	rsAPI rsapi.RoomserverInternalAPI, keyAPI keyapi.KeyInternalAPI,
-	eduCache *cache.EDUCache,
+	eduCache *caching.EDUCache,
 ) *Streams {
 	streams := &Streams{
 		PDUStreamProvider: &PDUStreamProvider{
diff --git a/syncapi/syncapi.go b/syncapi/syncapi.go
index ed8118bf..b579467a 100644
--- a/syncapi/syncapi.go
+++ b/syncapi/syncapi.go
@@ -18,9 +18,9 @@ import (
 	"context"
 
 	"github.com/gorilla/mux"
+	"github.com/matrix-org/dendrite/internal/caching"
 	"github.com/sirupsen/logrus"
 
-	"github.com/matrix-org/dendrite/eduserver/cache"
 	keyapi "github.com/matrix-org/dendrite/keyserver/api"
 	"github.com/matrix-org/dendrite/roomserver/api"
 	"github.com/matrix-org/dendrite/setup/config"
@@ -56,7 +56,7 @@ func AddPublicRoutes(
 		logrus.WithError(err).Panicf("failed to connect to sync db")
 	}
 
-	eduCache := cache.New()
+	eduCache := caching.NewTypingCache()
 	streams := streams.NewSyncStreamProviders(syncDB, userAPI, rsAPI, keyAPI, eduCache)
 	notifier := notifier.NewNotifier(streams.Latest(context.Background()))
 	if err = notifier.Load(context.Background(), syncDB); err != nil {
@@ -110,7 +110,7 @@ func AddPublicRoutes(
 	}
 
 	typingConsumer := consumers.NewOutputTypingEventConsumer(
-		process, cfg, js, syncDB, eduCache, notifier, streams.TypingStreamProvider,
+		process, cfg, js, eduCache, notifier, streams.TypingStreamProvider,
 	)
 	if err = typingConsumer.Start(); err != nil {
 		logrus.WithError(err).Panicf("failed to start typing consumer")
diff --git a/syncapi/types/types.go b/syncapi/types/types.go
index 4150e6c9..f964b80b 100644
--- a/syncapi/types/types.go
+++ b/syncapi/types/types.go
@@ -487,3 +487,21 @@ type StreamedEvent struct {
 	Event          *gomatrixserverlib.HeaderedEvent `json:"event"`
 	StreamPosition StreamPosition                   `json:"stream_position"`
 }
+
+// OutputReceiptEvent is an entry in the receipt output kafka log
+type OutputReceiptEvent struct {
+	UserID    string                      `json:"user_id"`
+	RoomID    string                      `json:"room_id"`
+	EventID   string                      `json:"event_id"`
+	Type      string                      `json:"type"`
+	Timestamp gomatrixserverlib.Timestamp `json:"timestamp"`
+}
+
+// OutputSendToDeviceEvent is an entry in the send-to-device output kafka log.
+// This contains the full event content, along with the user ID and device ID
+// to which it is destined.
+type OutputSendToDeviceEvent struct {
+	UserID   string `json:"user_id"`
+	DeviceID string `json:"device_id"`
+	gomatrixserverlib.SendToDeviceEvent
+}