Add queries to get the events for /context requests

syncapi/storage/interface.go

@@ -137,4 +137,9 @@ type Database interface {
 	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)
+
+	SelectContextEvent(ctx context.Context, roomID, eventID string) (int, gomatrixserverlib.HeaderedEvent, error)
+	SelectContextBeforeEvent(ctx context.Context, id int, roomID string, limit int) ([]*gomatrixserverlib.HeaderedEvent, error)
+	SelectContextAfterEvent(ctx context.Context, id int, roomID string, limit int) (int, []*gomatrixserverlib.HeaderedEvent, error)
+	SelectEventIDsAfter(ctx context.Context, roomID string, id int) ([]string, error)
 }

syncapi/storage/postgres/output_room_events_table.go

@@ -130,6 +130,18 @@ const selectStateInRangeSQL = "" +
 const deleteEventsForRoomSQL = "" +
 	"DELETE FROM syncapi_output_room_events WHERE room_id = $1"
 
+const selectContextEventSQL = "" +
+	"SELECT id, headered_event_json FROM syncapi_output_room_events WHERE room_id = $1 AND event_id = $2"
+
+const selectContextBeforeEventSQL = "" +
+	"SELECT headered_event_json FROM syncapi_output_room_events WHERE room_id = $1 AND id < $2 ORDER BY id DESC LIMIT $3"
+
+const selectContextAfterEventSQL = "" +
+	"SELECT id, headered_event_json FROM syncapi_output_room_events WHERE room_id = $1 AND id > $2 ORDER BY id ASC LIMIT $3"
+
+const selectEventIDsAfterSQL = "" +
+	"SELECT event_id FROM syncapi_output_room_events WHERE room_id = $1 AND id > $2"
+
 type outputRoomEventsStatements struct {
 	insertEventStmt               *sql.Stmt
 	selectEventsStmt              *sql.Stmt
@@ -140,6 +152,10 @@ type outputRoomEventsStatements struct {
 	selectStateInRangeStmt        *sql.Stmt
 	updateEventJSONStmt           *sql.Stmt
 	deleteEventsForRoomStmt       *sql.Stmt
+	selectContextEventStmt        *sql.Stmt
+	selectContextBeforeEventStmt  *sql.Stmt
+	selectContextAfterEventStmt   *sql.Stmt
+	selectEventIDsAfterStmt       *sql.Stmt
 }
 
 func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
@@ -148,34 +164,21 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
 	if err != nil {
 		return nil, err
 	}
-	if s.insertEventStmt, err = db.Prepare(insertEventSQL); err != nil {
+	return s, sqlutil.StatementList{
-		return nil, err
+		{&s.insertEventStmt, insertEventSQL},
-	}
+		{&s.selectEventsStmt, selectEventsSQL},
-	if s.selectEventsStmt, err = db.Prepare(selectEventsSQL); err != nil {
+		{&s.selectMaxEventIDStmt, selectMaxEventIDSQL},
-		return nil, err
+		{&s.selectRecentEventsStmt, selectRecentEventsSQL},
-	}
+		{&s.selectRecentEventsForSyncStmt, selectRecentEventsForSyncSQL},
-	if s.selectMaxEventIDStmt, err = db.Prepare(selectMaxEventIDSQL); err != nil {
+		{&s.selectEarlyEventsStmt, selectEarlyEventsSQL},
-		return nil, err
+		{&s.selectStateInRangeStmt, selectStateInRangeSQL},
-	}
+		{&s.updateEventJSONStmt, updateEventJSONSQL},
-	if s.selectRecentEventsStmt, err = db.Prepare(selectRecentEventsSQL); err != nil {
+		{&s.deleteEventsForRoomStmt, deleteEventsForRoomSQL},
-		return nil, err
+		{&s.selectContextEventStmt, selectContextEventSQL},
-	}
+		{&s.selectContextBeforeEventStmt, selectContextBeforeEventSQL},
-	if s.selectRecentEventsForSyncStmt, err = db.Prepare(selectRecentEventsForSyncSQL); err != nil {
+		{&s.selectContextAfterEventStmt, selectContextAfterEventSQL},
-		return nil, err
+		{&s.selectEventIDsAfterStmt, selectEventIDsAfterSQL},
-	}
+	}.Prepare(db)
-	if s.selectEarlyEventsStmt, err = db.Prepare(selectEarlyEventsSQL); err != nil {
-		return nil, err
-	}
-	if s.selectStateInRangeStmt, err = db.Prepare(selectStateInRangeSQL); err != nil {
-		return nil, err
-	}
-	if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil {
-		return nil, err
-	}
-	if s.deleteEventsForRoomStmt, err = db.Prepare(deleteEventsForRoomSQL); err != nil {
-		return nil, err
-	}
-	return s, nil
 }
 
 func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error {
@@ -436,6 +439,89 @@ func (s *outputRoomEventsStatements) DeleteEventsForRoom(
 	return err
 }
 
+func (s *outputRoomEventsStatements) SelectContextEvent(ctx context.Context, txn *sql.Tx, roomID, eventID string) (id int, evt gomatrixserverlib.HeaderedEvent, err error) {
+	row := sqlutil.TxStmt(txn, s.selectContextEventStmt).QueryRowContext(ctx, roomID, eventID)
+
+	var eventAsString string
+	if err = row.Scan(&id, &eventAsString); err != nil {
+		return 0, evt, err
+	}
+
+	if err = json.Unmarshal([]byte(eventAsString), &evt); err != nil {
+		return 0, evt, err
+	}
+	return id, evt, nil
+}
+
+func (s *outputRoomEventsStatements) SelectContextBeforeEvent(
+	ctx context.Context, txn *sql.Tx, id int, roomID string, limit int,
+) (evts []*gomatrixserverlib.HeaderedEvent, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectContextBeforeEventStmt).QueryContext(ctx, roomID, id, limit)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var (
+			eventBytes []byte
+			evt        *gomatrixserverlib.HeaderedEvent
+		)
+		if err = rows.Scan(&eventBytes); err != nil {
+			return evts, err
+		}
+		if err = json.Unmarshal(eventBytes, &evt); err != nil {
+			return evts, err
+		}
+		evts = append(evts, evt)
+	}
+
+	return evts, rows.Err()
+}
+
+func (s *outputRoomEventsStatements) SelectContextAfterEvent(
+	ctx context.Context, txn *sql.Tx, id int, roomID string, limit int,
+) (lastID int, evts []*gomatrixserverlib.HeaderedEvent, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectContextAfterEventStmt).QueryContext(ctx, roomID, id, limit)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var (
+			eventBytes []byte
+			evt        *gomatrixserverlib.HeaderedEvent
+		)
+		if err = rows.Scan(&lastID, &eventBytes); err != nil {
+			return 0, evts, err
+		}
+		if err = json.Unmarshal(eventBytes, &evt); err != nil {
+			return 0, evts, err
+		}
+		evts = append(evts, evt)
+	}
+
+	return lastID, evts, rows.Err()
+}
+
+func (s *outputRoomEventsStatements) SelectEventIDsAfter(ctx context.Context, roomID string, id int) (eventIDs []string, err error) {
+	rows, err := s.selectEventIDsAfterStmt.QueryContext(ctx, roomID, id)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var eventID string
+		if err = rows.Scan(&eventID); err != nil {
+			return nil, err
+		}
+		eventIDs = append(eventIDs, eventID)
+	}
+	return eventIDs, rows.Err()
+}
+
 func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
 	var result []types.StreamEvent
 	for rows.Next() {

syncapi/storage/shared/syncserver.go

@@ -955,3 +955,18 @@ func (d *Database) GetRoomReceipts(ctx context.Context, roomIDs []string, stream
 	_, receipts, err := d.Receipts.SelectRoomReceiptsAfter(ctx, roomIDs, streamPos)
 	return receipts, err
 }
+
+func (s *Database) SelectContextEvent(ctx context.Context, roomID, eventID string) (int, gomatrixserverlib.HeaderedEvent, error) {
+	return s.OutputEvents.SelectContextEvent(ctx, nil, roomID, eventID)
+}
+
+func (s *Database) SelectContextBeforeEvent(ctx context.Context, id int, roomID string, limit int) ([]*gomatrixserverlib.HeaderedEvent, error) {
+	return s.OutputEvents.SelectContextBeforeEvent(ctx, nil, id, roomID, limit)
+}
+func (s *Database) SelectContextAfterEvent(ctx context.Context, id int, roomID string, limit int) (int, []*gomatrixserverlib.HeaderedEvent, error) {
+	return s.OutputEvents.SelectContextAfterEvent(ctx, nil, id, roomID, limit)
+}
+
+func (s *Database) SelectEventIDsAfter(ctx context.Context, roomID string, id int) ([]string, error) {
+	return s.OutputEvents.SelectEventIDsAfter(ctx, roomID, id)
+}

syncapi/storage/sqlite3/output_room_events_table.go

@@ -62,17 +62,20 @@ const selectEventsSQL = "" +
 const selectRecentEventsSQL = "" +
 	"SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
 	" WHERE room_id = $1 AND id > $2 AND id <= $3"
-	// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
+
+// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
 
 const selectRecentEventsForSyncSQL = "" +
 	"SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
 	" WHERE room_id = $1 AND id > $2 AND id <= $3 AND exclude_from_sync = FALSE"
-	// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
+
+// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
 
 const selectEarlyEventsSQL = "" +
 	"SELECT event_id, id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
 	" WHERE room_id = $1 AND id > $2 AND id <= $3"
-	// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
+
+// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
 
 const selectMaxEventIDSQL = "" +
 	"SELECT MAX(id) FROM syncapi_output_room_events"
@@ -85,19 +88,36 @@ const selectStateInRangeSQL = "" +
 	" FROM syncapi_output_room_events" +
 	" WHERE (id > $1 AND id <= $2)" +
 	" AND ((add_state_ids IS NOT NULL AND add_state_ids != '') OR (remove_state_ids IS NOT NULL AND remove_state_ids != ''))"
-	// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
+
+// WHEN, ORDER BY and LIMIT are appended by prepareWithFilters
 
 const deleteEventsForRoomSQL = "" +
 	"DELETE FROM syncapi_output_room_events WHERE room_id = $1"
 
+const selectContextEventSQL = "" +
+	"SELECT id, headered_event_json FROM syncapi_output_room_events WHERE room_id = $1 AND event_id = $2"
+
+const selectContextBeforeEventSQL = "" +
+	"SELECT headered_event_json FROM syncapi_output_room_events WHERE room_id = $1 AND id < $2 ORDER BY id DESC LIMIT $3"
+
+const selectContextAfterEventSQL = "" +
+	"SELECT id, headered_event_json FROM syncapi_output_room_events WHERE room_id = $1 AND id > $2 ORDER BY id ASC LIMIT $3"
+
+const selectEventIDsAfterSQL = "" +
+	"SELECT event_id FROM syncapi_output_room_events WHERE room_id = $1 AND id > $2"
+
 type outputRoomEventsStatements struct {
-	db                      *sql.DB
+	db                           *sql.DB
-	streamIDStatements      *streamIDStatements
+	streamIDStatements           *streamIDStatements
-	insertEventStmt         *sql.Stmt
+	insertEventStmt              *sql.Stmt
-	selectEventsStmt        *sql.Stmt
+	selectEventsStmt             *sql.Stmt
-	selectMaxEventIDStmt    *sql.Stmt
+	selectMaxEventIDStmt         *sql.Stmt
-	updateEventJSONStmt     *sql.Stmt
+	updateEventJSONStmt          *sql.Stmt
-	deleteEventsForRoomStmt *sql.Stmt
+	deleteEventsForRoomStmt      *sql.Stmt
+	selectContextEventStmt       *sql.Stmt
+	selectContextBeforeEventStmt *sql.Stmt
+	selectContextAfterEventStmt  *sql.Stmt
+	selectEventIDsAfterStmt      *sql.Stmt
 }
 
 func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Events, error) {
@@ -109,22 +129,17 @@ func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Even
 	if err != nil {
 		return nil, err
 	}
-	if s.insertEventStmt, err = db.Prepare(insertEventSQL); err != nil {
+	return s, sqlutil.StatementList{
-		return nil, err
+		{&s.insertEventStmt, insertEventSQL},
-	}
+		{&s.selectEventsStmt, selectEventsSQL},
-	if s.selectEventsStmt, err = db.Prepare(selectEventsSQL); err != nil {
+		{&s.selectMaxEventIDStmt, selectMaxEventIDSQL},
-		return nil, err
+		{&s.updateEventJSONStmt, updateEventJSONSQL},
-	}
+		{&s.deleteEventsForRoomStmt, deleteEventsForRoomSQL},
-	if s.selectMaxEventIDStmt, err = db.Prepare(selectMaxEventIDSQL); err != nil {
+		{&s.selectContextEventStmt, selectContextEventSQL},
-		return nil, err
+		{&s.selectContextBeforeEventStmt, selectContextBeforeEventSQL},
-	}
+		{&s.selectContextAfterEventStmt, selectContextAfterEventSQL},
-	if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil {
+		{&s.selectEventIDsAfterStmt, selectEventIDsAfterSQL},
-		return nil, err
+	}.Prepare(db)
-	}
-	if s.deleteEventsForRoomStmt, err = db.Prepare(deleteEventsForRoomSQL); err != nil {
-		return nil, err
-	}
-	return s, nil
 }
 
 func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error {
@@ -462,6 +477,89 @@ func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
 	}
 	return result, nil
 }
+func (s *outputRoomEventsStatements) SelectContextEvent(
+	ctx context.Context, txn *sql.Tx, roomID, eventID string,
+) (id int, evt gomatrixserverlib.HeaderedEvent, err error) {
+	row := sqlutil.TxStmt(txn, s.selectContextEventStmt).QueryRowContext(ctx, roomID, eventID)
+	var eventAsString string
+	if err = row.Scan(&id, &eventAsString); err != nil {
+		return 0, evt, err
+	}
+
+	if err = json.Unmarshal([]byte(eventAsString), &evt); err != nil {
+		return 0, evt, err
+	}
+	return id, evt, nil
+}
+
+func (s *outputRoomEventsStatements) SelectContextBeforeEvent(
+	ctx context.Context, txn *sql.Tx, id int, roomID string, limit int,
+) (evts []*gomatrixserverlib.HeaderedEvent, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectContextBeforeEventStmt).QueryContext(ctx, roomID, id, limit)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var (
+			eventBytes []byte
+			evt        *gomatrixserverlib.HeaderedEvent
+		)
+		if err = rows.Scan(&eventBytes); err != nil {
+			return evts, err
+		}
+		if err = json.Unmarshal(eventBytes, &evt); err != nil {
+			return evts, err
+		}
+		evts = append(evts, evt)
+	}
+
+	return evts, rows.Err()
+}
+
+func (s *outputRoomEventsStatements) SelectContextAfterEvent(
+	ctx context.Context, txn *sql.Tx, id int, roomID string, limit int,
+) (lastID int, evts []*gomatrixserverlib.HeaderedEvent, err error) {
+	rows, err := sqlutil.TxStmt(txn, s.selectContextAfterEventStmt).QueryContext(ctx, roomID, id, limit)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var (
+			eventBytes []byte
+			evt        *gomatrixserverlib.HeaderedEvent
+		)
+		if err = rows.Scan(&lastID, &eventBytes); err != nil {
+			return 0, evts, err
+		}
+		if err = json.Unmarshal(eventBytes, &evt); err != nil {
+			return 0, evts, err
+		}
+		evts = append(evts, evt)
+	}
+
+	return lastID, evts, rows.Err()
+}
+
+func (s *outputRoomEventsStatements) SelectEventIDsAfter(ctx context.Context, roomID string, id int) (eventIDs []string, err error) {
+	rows, err := s.selectEventIDsAfterStmt.QueryContext(ctx, roomID, id)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var eventID string
+		if err = rows.Scan(&eventID); err != nil {
+			return nil, err
+		}
+		eventIDs = append(eventIDs, eventID)
+	}
+	return eventIDs, rows.Err()
+}
 
 func unmarshalStateIDs(addIDsJSON, delIDsJSON string) (addIDs []string, delIDs []string, err error) {
 	if len(addIDsJSON) > 0 {

syncapi/storage/tables/interface.go

@@ -63,6 +63,11 @@ type Events interface {
 	UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error
 	// DeleteEventsForRoom removes all event information for a room. This should only be done when removing the room entirely.
 	DeleteEventsForRoom(ctx context.Context, txn *sql.Tx, roomID string) (err error)
+
+	SelectContextEvent(ctx context.Context, txn *sql.Tx, roomID, eventID string) (int, gomatrixserverlib.HeaderedEvent, error)
+	SelectContextAfterEvent(ctx context.Context, txn *sql.Tx, id int, roomID string, limit int) (int, []*gomatrixserverlib.HeaderedEvent, error)
+	SelectContextBeforeEvent(ctx context.Context, txn *sql.Tx, id int, roomID string, limit int) ([]*gomatrixserverlib.HeaderedEvent, error)
+	SelectEventIDsAfter(ctx context.Context, roomID string, id int) ([]string, error)
 }
 
 // Topology keeps track of the depths and stream positions for all events.