a very very WIP first cut of peeking via MSC2753.

doesn't yet compile or work.
needs to actually add the peeking block into the sync response.
checking in now before it gets any bigger, and to gather any initial feedback on the vague shape of it.

syncapi/consumers/roomserver.go

@@ -99,6 +99,8 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 		return s.onNewInviteEvent(context.TODO(), *output.NewInviteEvent)
 	case api.OutputTypeRetireInviteEvent:
 		return s.onRetireInviteEvent(context.TODO(), *output.RetireInviteEvent)
+	case api.OutputTypeNewPeek:
+		return s.onNewPeek(context.TODO(), *output.NewPeek)
 	case api.OutputTypeRedactedEvent:
 		return s.onRedactEvent(context.TODO(), *output.RedactedEvent)
 	default:
@@ -218,6 +220,26 @@ func (s *OutputRoomEventConsumer) onRetireInviteEvent(
 	return nil
 }
 
+func (s *OutputRoomEventConsumer) onNewPeek(
+	ctx context.Context, msg api.OutputNewPeek,
+) error {
+	sp, err := s.db.AddPeek(ctx, msg.RoomID, msg.UserID, msg.DeviceID)
+	if err != nil {
+		// panic rather than continue with an inconsistent database
+		log.WithFields(log.Fields{
+			log.ErrorKey: err,
+		}).Panicf("roomserver output log: write peek failure")
+		return nil
+	}
+	// tell the notifier about the new peek so it knows to wake up new devices
+	s.notifier.OnNewPeek(msg.RoomID, msg.UserID, msg.DeviceID)
+
+	// we need to wake up the users who might need to now be peeking into this room,
+	// so we send in a dummy event to trigger a wakeup
+	s.notifier.OnNewEvent(nil, msg.RoomID, nil, types.NewStreamToken(sp, 0, nil))
+	return nil
+}
+
 func (s *OutputRoomEventConsumer) updateStateEvent(event gomatrixserverlib.HeaderedEvent) (gomatrixserverlib.HeaderedEvent, error) {
 	if event.StateKey() == nil {
 		return event, nil

syncapi/storage/interface.go

@@ -30,6 +30,8 @@ type Database interface {
 	internal.PartitionStorer
 	// AllJoinedUsersInRooms returns a map of room ID to a list of all joined user IDs.
 	AllJoinedUsersInRooms(ctx context.Context) (map[string][]string, error)
+	// AllPeekingDevicesInRooms returns a map of room ID to a list of all peeking devices.
+	AllPeekingDevicesInRooms(ctx context.Context) (map[string][]PeekingDevice, error)
 	// Events lookups a list of event by their event ID.
 	// Returns a list of events matching the requested IDs found in the database.
 	// If an event is not found in the database then it will be omitted from the list.

syncapi/storage/shared/syncserver.go

@@ -120,6 +120,10 @@ func (d *Database) AllJoinedUsersInRooms(ctx context.Context) (map[string][]stri
 	return d.CurrentRoomState.SelectJoinedUsers(ctx)
 }
 
+func (d *Database) AllPeekingDevicesInRooms(ctx context.Context) (map[string][]PeekingDevice, error) {
+	return d.Peeks.SelectPeekingDevices(ctx)
+}
+
 func (d *Database) GetStateEvent(
 	ctx context.Context, roomID, evType, stateKey string,
 ) (*gomatrixserverlib.HeaderedEvent, error) {
@@ -187,6 +191,19 @@ func (d *Database) RetireInviteEvent(
 	return
 }
 
+// AddPeek tracks the fact that a user has started peeking.
+// If the peek was successfully stored this returns the stream ID it was stored at.
+// Returns an error if there was a problem communicating with the database.
+func (d *Database) AddPeek(
+	ctx context.Context, roomID, userID, deviceID string,
+) (sp types.StreamPosition, err error) {
+	_ = d.Writer.Do(nil, nil, func(_ *sql.Tx) error {
+		sp, err = d.Peeks.InsertPeek(ctx, nil, inviteEvent)
+		return nil
+	})
+	return
+}
+
 // GetAccountDataInRange returns all account data for a given user inserted or
 // updated between two given positions
 // Returns a map following the format data[roomID] = []dataTypes

syncapi/storage/sqlite3/peeks_table.go

@@ -0,0 +1,151 @@
+// Copyright 2020 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 sqlite3
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+
+	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/internal/sqlutil"
+	"github.com/matrix-org/dendrite/syncapi/storage/tables"
+	"github.com/matrix-org/dendrite/syncapi/types"
+	"github.com/matrix-org/gomatrixserverlib"
+)
+
+const peeksSchema = `
+CREATE TABLE IF NOT EXISTS syncapi_peeks (
+	id INTEGER PRIMARY KEY,
+	room_id TEXT NOT NULL,
+	user_id TEXT NOT NULL,
+	device_id TEXT NOT NULL,
+    -- When the peek was created in UNIX epoch ms.
+    creation_ts INTEGER NOT NULL,
+);
+
+CREATE INDEX IF NOT EXISTS syncapi_peeks_room_id_idx ON syncapi_peeks(room_id);
+CREATE INDEX IF NOT EXISTS syncapi_peeks_user_id_device_id_idx ON syncapi_peeks(user_Id, device_id);
+`
+
+const insertPeekSQL = "" +
+	"INSERT INTO syncapi_peeks" +
+	" (id, room_id, user_id, device_id, creation_ts" +
+	" VALUES ($1, $2, $3, $4, $5)"
+
+const deletePeekSQL = "" +
+	"DELETE FROM syncapi_peeks WHERE room_id = $1 AND user_id = $2 and device_id = $3"
+
+const selectPeeksSQL == "" +
+	"SELECT room_id FROM syncapi_peeks WHERE user_id = $1 and device_id = $2"
+
+const selectPeekingDevicesSQL == "" +
+	"SELECT room_id, user_id, device_id FROM syncapi_peeks"
+
+type peekStatements struct {
+	db                            *sql.DB
+	insertPeekStmt        		  *sql.Stmt
+	deletePeekStmt         		  *sql.Stmt
+	selectPeeksStmt				  *sql.Stmt
+	selectPeekingDevicesStmt	  *sql.Stmt
+}
+
+func NewSqlitePeeksTable(db *sql.DB) (tables.Peeks, error) {
+	_, err := db.Exec(filterSchema)
+	if err != nil {
+		return nil, err
+	}
+	s := &peekStatements{
+		db: db,
+	}
+	if s.insertPeekStmt, err = db.Prepare(insertPeekSQL); err != nil {
+		return nil, err
+	}
+	if s.deletePeekStmt, err = db.Prepare(deletePeekSQL); err != nil {
+		return nil, err
+	}
+	if s.selectPeeksStmt, err = db.Prepare(selectPeeksSQL); err != nil {
+		return nil, err
+	}
+	if s.selectPeekingDevicesStmt, err = db.Prepare(selectPeekingDevicesSQL); err != nil {
+		return nil, err
+	}
+	return s, nil
+}
+
+func (s *peekStatements) InsertPeek(
+	ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string,
+) (streamPos types.StreamPosition, err error) {
+	streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn)
+	if err != nil {
+		return
+	}
+	nowMilli := time.Now().UnixNano() / int64(time.Millisecond)
+	_, err = sqlutil.TxStmt(txn, s.insertPeekStmt).ExecContext(ctx, roomID, userID, deviceID, nowMilli)
+	return
+}
+
+func (s *peekStatements) DeletePeek(
+	ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string,
+) (streamPos types.StreamPosition, err error) {
+	streamPos, err = s.streamIDStatements.nextStreamID(ctx, txn)
+	if err != nil {
+		return
+	}
+	_, err = sqlutil.TxStmt(txn, s.deletePeekStmt).ExecContext(ctx, roomID, userID, deviceID)
+	return
+}
+
+func (s *peekStatements) SelectPeeks(
+	ctx context.Context, txn *sql.Tx, userID, deviceID string,
+) (roomIDs []string, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectPeeksStmt).QueryContext(ctx, userID, deviceID)
+	if err != nil {
+		return
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectPeeks: rows.close() failed")
+
+	for rows.Next() {
+		var roomID string
+		if err = rows.Scan(&roomId); err != nil {
+			return
+		}
+		roomIDs = append(roomIDs, roomID)
+	}
+
+	return roomIDs, rows.Err()
+}
+
+func (s *peekStatements) SelectPeekingDevices(
+	ctx context.Context,
+) (peekingDevices map[string][]PeekingDevice, err error) {
+	rows, err := s.selectPeekingDevicesStmt.QueryContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer internal.CloseAndLogIfError(ctx, rows, "SelectPeekingDevices: rows.close() failed")
+
+	result := make(map[string][]PeekingDevice)
+	for rows.Next() {
+		var roomID, userID, deviceID string
+		if err := rows.Scan(&roomID, &userID, &deviceID); err != nil {
+			return nil, err
+		}
+		devices := result[roomID]
+		devices = append(devices, PeekingDevice{userID, deviceID})
+		result[roomID] = devices
+	}
+	return result, nil
+}

syncapi/storage/tables/interface.go

@@ -39,6 +39,13 @@ type Invites interface {
 	SelectMaxInviteID(ctx context.Context, txn *sql.Tx) (id int64, err error)
 }
 
+type Peeks interface {
+	InsertPeek(ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string) (streamPos types.StreamPosition, err error)
+	DeletePeek(ctx context.Context, txn *sql.Tx, roomID, userID, deviceID string) (streamPos types.StreamPosition, err error)
+	SelectPeeks(ctxt context.Context, txn *sql.Tx, userID, deviceID string) (peeks []string, err error)
+	SelectPeekingDevices((ctxt context.Context) (peekingDevices map[string][]PeekingDevice, err error)
+}
+
 type Events interface {
 	SelectStateInRange(ctx context.Context, txn *sql.Tx, r types.Range, stateFilter *gomatrixserverlib.StateFilter) (map[string]map[string]bool, map[string]types.StreamEvent, error)
 	SelectMaxEventID(ctx context.Context, txn *sql.Tx) (id int64, err error)

syncapi/sync/notifier.go

@@ -33,6 +33,8 @@ import (
 type Notifier struct {
 	// A map of RoomID => Set<UserID> : Must only be accessed by the OnNewEvent goroutine
 	roomIDToJoinedUsers map[string]userIDSet
+	// A map of RoomID => Set<UserID> : Must only be accessed by the OnNewEvent goroutine
+	roomIDToPeekingDevices map[string]PeekingDeviceSet
 	// Protects currPos and userStreams.
 	streamLock *sync.Mutex
 	// The latest sync position
@@ -48,11 +50,11 @@ type Notifier struct {
 // the joined users within each of them by calling Notifier.Load(*storage.SyncServerDatabase).
 func NewNotifier(pos types.StreamingToken) *Notifier {
 	return &Notifier{
-		currPos:             pos,
-		roomIDToJoinedUsers: make(map[string]userIDSet),
-		userDeviceStreams:   make(map[string]map[string]*UserDeviceStream),
-		streamLock:          &sync.Mutex{},
-		lastCleanUpTime:     time.Now(),
+		currPos:                pos,
+		roomIDToJoinedUsers:    make(map[string]userIDSet),
+		roomIDToPeekingDevices: make(map[string]PeekingDeviceSet),
+		userDeviceStreams:      make(map[string]map[string]*UserDeviceStream),
+		streamLock:             &sync.Mutex{},
 	}
 }
 
@@ -82,6 +84,8 @@ func (n *Notifier) OnNewEvent(
 	if ev != nil {
 		// Map this event's room_id to a list of joined users, and wake them up.
 		usersToNotify := n.joinedUsers(ev.RoomID())
+		// Map this event's room_id to a list of peeking devices, and wake them up.
+		peekingDevicesToNotify := n.PeekingDevices(ev.RoomID())
 		// If this is an invite, also add in the invitee to this list.
 		if ev.Type() == "m.room.member" && ev.StateKey() != nil {
 			targetUserID := *ev.StateKey()
@@ -108,11 +112,11 @@ func (n *Notifier) OnNewEvent(
 			}
 		}
 
-		n.wakeupUsers(usersToNotify, latestPos)
+		n.wakeupUsers(usersToNotify, peekingDevicesToNotify, latestPos)
 	} else if roomID != "" {
-		n.wakeupUsers(n.joinedUsers(roomID), latestPos)
+		n.wakeupUsers(n.joinedUsers(roomID), n.PeekingDevices(roomID), latestPos)
 	} else if len(userIDs) > 0 {
-		n.wakeupUsers(userIDs, latestPos)
+		n.wakeupUsers(userIDs, nil, latestPos)
 	} else {
 		log.WithFields(log.Fields{
 			"posUpdate": posUpdate.String,
@@ -120,6 +124,18 @@ func (n *Notifier) OnNewEvent(
 	}
 }
 
+func (n *Notifier) OnNewPeek(
+	roomID, userID, deviceID string,
+) {
+	n.streamLock.Lock()
+	defer n.streamLock.Unlock()
+
+	n.addPeekingDevice(roomID, userID, deviceID)
+
+	// we don't wake up devices here given the roomserver consumer will do this shortly afterwards
+	// by calling OnNewEvent.
+}
+
 func (n *Notifier) OnNewSendToDevice(
 	userID string, deviceIDs []string,
 	posUpdate types.StreamingToken,
@@ -139,7 +155,7 @@ func (n *Notifier) OnNewKeyChange(
 	defer n.streamLock.Unlock()
 	latestPos := n.currPos.WithUpdates(posUpdate)
 	n.currPos = latestPos
-	n.wakeupUsers([]string{wakeUserID}, latestPos)
+	n.wakeupUsers([]string{wakeUserID}, nil, latestPos)
 }
 
 // GetListener returns a UserStreamListener that can be used to wait for
@@ -169,6 +185,13 @@ func (n *Notifier) Load(ctx context.Context, db storage.Database) error {
 		return err
 	}
 	n.setUsersJoinedToRooms(roomToUsers)
+
+	roomToPeekingDevices, err := db.AllPeekingDevicesInRooms(ctx)
+	if err != nil {
+		return err
+	}
+	n.setPeekingDevices(roomToPeekingDevices)
+
 	return nil
 }
 
@@ -195,9 +218,24 @@ func (n *Notifier) setUsersJoinedToRooms(roomIDToUserIDs map[string][]string) {
 	}
 }
 
+// setPeekingDevices marks the given devices as peeking in the given rooms, such that new events from
+// these rooms will wake the given devices' /sync requests. This should be called prior to ANY calls to
+// OnNewEvent (eg on startup) to prevent racing.
+func (n *Notifier) setPeekingDevices(roomIDToPeekingDevices map[string][]PeekingDevices) {
+	// This is just the bulk form of addPeekingDevice
+	for roomID, peekingDevices := range roomIDToPeekingDevices {
+		if _, ok := n.roomIDToPeekingDevices[roomID]; !ok {
+			n.roomIDToPeekingDevices[roomID] = make(PeekingDeviceSet)
+		}
+		for _, peekingDevice := range peekingDevices {
+			n.roomIDToPeekingDevices[roomID].add(peekingDevice)
+		}
+	}
+}
+
 // wakeupUsers will wake up the sync strems for all of the devices for all of the
-// specified user IDs.
-func (n *Notifier) wakeupUsers(userIDs []string, newPos types.StreamingToken) {
+// specified user IDs, and also the specified peekingDevices
+func (n *Notifier) wakeupUsers(userIDs []string, peekingDevices []PeekingDevice, newPos types.StreamingToken) {
 	for _, userID := range userIDs {
 		for _, stream := range n.fetchUserStreams(userID) {
 			if stream == nil {
@@ -206,6 +244,15 @@ func (n *Notifier) wakeupUsers(userIDs []string, newPos types.StreamingToken) {
 			stream.Broadcast(newPos) // wake up all goroutines Wait()ing on this stream
 		}
 	}
+
+	if peekingDevices != nil {
+		for _, peekingDevice := range peekingDevices {
+			// TODO: don't bother waking up for devices whose users we already woke up
+			if stream := n.fetchUserDeviceStream(peekingDevice.UserID, peekingDevice.DeviceID, false); stream != nil {
+				stream.Broadcast(newPos) // wake up all goroutines Wait()ing on this stream
+			}
+		}
+	}
 }
 
 // wakeupUserDevice will wake up the sync stream for a specific user device. Other
@@ -284,6 +331,34 @@ func (n *Notifier) joinedUsers(roomID string) (userIDs []string) {
 	return n.roomIDToJoinedUsers[roomID].values()
 }
 
+
+// Not thread-safe: must be called on the OnNewEvent goroutine only
+func (n *Notifier) addPeekingDevice(roomID, userID, deviceID string) {
+	if _, ok := n.roomIDToPeekingDevices[roomID]; !ok {
+		n.roomIDToPeekingDevices[roomID] = make(PeekingDeviceSet)
+	}
+	n.roomIDToPeekingDevices[roomID].add(PeekingDevice{deviceID, userID})
+}
+
+// Not thread-safe: must be called on the OnNewEvent goroutine only
+func (n *Notifier) removePeekingDevice(roomID, userID, deviceID string) {
+	if _, ok := n.roomIDToPeekingDevices[roomID]; !ok {
+		n.roomIDToPeekingDevices[roomID] = make(PeekingDeviceSet)
+	}
+	// XXX: is this going to work as a key?
+	n.roomIDToPeekingDevices[roomID].remove(PeekingDevice{deviceID, userID})
+}
+
+// Not thread-safe: must be called on the OnNewEvent goroutine only
+func (n *Notifier) PeekingDevices(roomID string) (peekingDevices []PeekingDevices) {
+	if _, ok := n.roomIDToPeekingDevices[roomID]; !ok {
+		return
+	}
+	return n.roomIDToPeekingDevices[roomID].values()
+}
+
+
+
 // removeEmptyUserStreams iterates through the user stream map and removes any
 // that have been empty for a certain amount of time. This is a crude way of
 // ensuring that the userStreams map doesn't grow forver.

syncapi/types/types.go

@@ -507,3 +507,27 @@ type SendToDeviceEvent struct {
 	DeviceID    string
 	SentByToken *StreamingToken
 }
+
+// For tracking peeking devices
+
+type PeekingDevice struct {
+	ID     string
+	UserID string
+}
+
+type PeekingDeviceSet map[PeekingDevice]bool
+
+func (s PeekingDeviceSet) add(d PeekingDevice) {
+	s[d] = true
+}
+
+func (s PeekingDeviceSet) remove(d PeekingDevice) {
+	delete(s, d)
+}
+
+func (s PeekingDeviceSet) values() (vals []PeekingDevice) {
+	for d := range s {
+		vals = append(vals, d)
+	}
+	return
+}