Don't leak timeline after leaving room

syncapi/storage/postgres/output_room_events_table.go

@@ -130,6 +130,10 @@ const selectStateInRangeSQL = "" +
 const deleteEventsForRoomSQL = "" +
 	"DELETE FROM syncapi_output_room_events WHERE room_id = $1"
 
+const selectPositionInStreamSQL = "" +
+	"SELECT id FROM syncapi_output_room_events" +
+	" WHERE event_id = $1"
+
 type outputRoomEventsStatements struct {
 	insertEventStmt               *sql.Stmt
 	selectEventsStmt              *sql.Stmt
@@ -140,6 +144,7 @@ type outputRoomEventsStatements struct {
 	selectStateInRangeStmt        *sql.Stmt
 	updateEventJSONStmt           *sql.Stmt
 	deleteEventsForRoomStmt       *sql.Stmt
+	selectPositionInStreamStmt    *sql.Stmt
 }
 
 func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
@@ -175,6 +180,9 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
 	if s.deleteEventsForRoomStmt, err = db.Prepare(deleteEventsForRoomSQL); err != nil {
 		return nil, err
 	}
+	if s.selectPositionInStreamStmt, err = db.Prepare(selectPositionInStreamSQL); err != nil {
+		return nil, err
+	}
 	return s, nil
 }
 
@@ -435,6 +443,15 @@ func (s *outputRoomEventsStatements) DeleteEventsForRoom(
 	return err
 }
 
+// SelectPositionInStream returns the position of a given event in the
+// global stream topology.
+func (s *outputRoomEventsStatements) SelectPositionInStream(
+	ctx context.Context, txn *sql.Tx, eventID string,
+) (pos types.StreamPosition, err error) {
+	err = s.selectPositionInStreamStmt.QueryRowContext(ctx, eventID).Scan(&pos)
+	return
+}
+
 func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
 	var result []types.StreamEvent
 	for rows.Next() {