diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go index b464ad9cd..aa99dbbf4 100644 --- a/syncapi/storage/interface.go +++ b/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) } diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go index 44de02c92..57d889b3f 100644 --- a/syncapi/storage/postgres/output_room_events_table.go +++ b/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 nil, err - } - if s.selectEventsStmt, err = db.Prepare(selectEventsSQL); err != nil { - return nil, err - } - if s.selectMaxEventIDStmt, err = db.Prepare(selectMaxEventIDSQL); err != nil { - return nil, err - } - if s.selectRecentEventsStmt, err = db.Prepare(selectRecentEventsSQL); err != nil { - return nil, err - } - if s.selectRecentEventsForSyncStmt, err = db.Prepare(selectRecentEventsForSyncSQL); err != nil { - return nil, err - } - 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 + return s, sqlutil.StatementList{ + {&s.insertEventStmt, insertEventSQL}, + {&s.selectEventsStmt, selectEventsSQL}, + {&s.selectMaxEventIDStmt, selectMaxEventIDSQL}, + {&s.selectRecentEventsStmt, selectRecentEventsSQL}, + {&s.selectRecentEventsForSyncStmt, selectRecentEventsForSyncSQL}, + {&s.selectEarlyEventsStmt, selectEarlyEventsSQL}, + {&s.selectStateInRangeStmt, selectStateInRangeSQL}, + {&s.updateEventJSONStmt, updateEventJSONSQL}, + {&s.deleteEventsForRoomStmt, deleteEventsForRoomSQL}, + {&s.selectContextEventStmt, selectContextEventSQL}, + {&s.selectContextBeforeEventStmt, selectContextBeforeEventSQL}, + {&s.selectContextAfterEventStmt, selectContextAfterEventSQL}, + {&s.selectEventIDsAfterStmt, selectEventIDsAfterSQL}, + }.Prepare(db) } 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() { diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index e6c681832..b24f2a94b 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/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) +} diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go index afdbe55ce..0a635dfb5 100644 --- a/syncapi/storage/sqlite3/output_room_events_table.go +++ b/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 - streamIDStatements *streamIDStatements - insertEventStmt *sql.Stmt - selectEventsStmt *sql.Stmt - selectMaxEventIDStmt *sql.Stmt - updateEventJSONStmt *sql.Stmt - deleteEventsForRoomStmt *sql.Stmt + db *sql.DB + streamIDStatements *streamIDStatements + insertEventStmt *sql.Stmt + selectEventsStmt *sql.Stmt + selectMaxEventIDStmt *sql.Stmt + updateEventJSONStmt *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 nil, err - } - if s.selectEventsStmt, err = db.Prepare(selectEventsSQL); err != nil { - return nil, err - } - if s.selectMaxEventIDStmt, err = db.Prepare(selectMaxEventIDSQL); 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 + return s, sqlutil.StatementList{ + {&s.insertEventStmt, insertEventSQL}, + {&s.selectEventsStmt, selectEventsSQL}, + {&s.selectMaxEventIDStmt, selectMaxEventIDSQL}, + {&s.updateEventJSONStmt, updateEventJSONSQL}, + {&s.deleteEventsForRoomStmt, deleteEventsForRoomSQL}, + {&s.selectContextEventStmt, selectContextEventSQL}, + {&s.selectContextBeforeEventStmt, selectContextBeforeEventSQL}, + {&s.selectContextAfterEventStmt, selectContextAfterEventSQL}, + {&s.selectEventIDsAfterStmt, selectEventIDsAfterSQL}, + }.Prepare(db) } 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 { diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index 028872716..5c5ffcd1a 100644 --- a/syncapi/storage/tables/interface.go +++ b/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.