Peeking updates (#1607)

* Add unpeek

* Don't allow peeks into encrypted rooms

* Fix send tests

* Update consumers

syncapi/consumers/roomserver.go

@@ -105,6 +105,8 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 		return s.onRetireInviteEvent(context.TODO(), *output.RetireInviteEvent)
 	case api.OutputTypeNewPeek:
 		return s.onNewPeek(context.TODO(), *output.NewPeek)
+	case api.OutputTypeRetirePeek:
+		return s.onRetirePeek(context.TODO(), *output.RetirePeek)
 	case api.OutputTypeRedactedEvent:
 		return s.onRedactEvent(context.TODO(), *output.RedactedEvent)
 	default:
@@ -309,6 +311,26 @@ func (s *OutputRoomEventConsumer) onNewPeek(
 	return nil
 }
 
+func (s *OutputRoomEventConsumer) onRetirePeek(
+	ctx context.Context, msg api.OutputRetirePeek,
+) error {
+	sp, err := s.db.DeletePeek(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.OnRetirePeek(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

@@ -91,6 +91,9 @@ type Database interface {
 	// AddPeek adds a new peek to our DB for a given room by a given user's device.
 	// Returns an error if there was a problem communicating with the database.
 	AddPeek(ctx context.Context, RoomID, UserID, DeviceID string) (types.StreamPosition, error)
+	// DeletePeek removes an existing peek from the database for a given room by a user's device.
+	// Returns an error if there was a problem communicating with the database.
+	DeletePeek(ctx context.Context, roomID, userID, deviceID string) (sp types.StreamPosition, err error)
 	// DeletePeek deletes all peeks for a given room by a given user
 	// Returns an error if there was a problem communicating with the database.
 	DeletePeeks(ctx context.Context, RoomID, UserID string) (types.StreamPosition, error)

syncapi/storage/shared/syncserver.go

@@ -178,6 +178,23 @@ func (d *Database) AddPeek(
 	return
 }
 
+// DeletePeeks tracks the fact that a user has stopped peeking from the specified
+// device. If the peeks was successfully deleted this returns the stream ID it was
+// stored at. Returns an error if there was a problem communicating with the database.
+func (d *Database) DeletePeek(
+	ctx context.Context, roomID, userID, deviceID string,
+) (sp types.StreamPosition, err error) {
+	err = d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+		sp, err = d.Peeks.DeletePeek(ctx, txn, roomID, userID, deviceID)
+		return err
+	})
+	if err == sql.ErrNoRows {
+		sp = 0
+		err = nil
+	}
+	return
+}
+
 // DeletePeeks tracks the fact that a user has stopped peeking from all devices
 // If the peeks was successfully deleted this returns the stream ID it was stored at.
 // Returns an error if there was a problem communicating with the database.

syncapi/sync/notifier.go

@@ -137,6 +137,18 @@ func (n *Notifier) OnNewPeek(
 	// by calling OnNewEvent.
 }
 
+func (n *Notifier) OnRetirePeek(
+	roomID, userID, deviceID string,
+) {
+	n.streamLock.Lock()
+	defer n.streamLock.Unlock()
+
+	n.removePeekingDevice(roomID, userID, deviceID)
+
+	// we don't wake up devices here given the roomserver consumer will do this shortly afterwards
+	// by calling OnRetireEvent.
+}
+
 func (n *Notifier) OnNewSendToDevice(
 	userID string, deviceIDs []string,
 	posUpdate types.StreamingToken,