aboutsummaryrefslogtreecommitdiff
path: root/syncapi
diff options
context:
space:
mode:
authorNeil Alexander <neilalexander@users.noreply.github.com>2020-09-15 11:17:46 +0100
committerGitHub <noreply@github.com>2020-09-15 11:17:46 +0100
commit965f068d1a6298b2ec733b0df983773a6ec8b622 (patch)
treee7a5cda3dba38114189eb857111baaa07bbe4854 /syncapi
parent8dc95062101b3906ffb83604e2abca02d9a3dd03 (diff)
Handle state with input event as new events (#1415)
* SendEventWithState events as new * Use cumulative state IDs for final event * Error wrapping in calculateAndSetState * Handle overwriting same event type and state key * Hacky way to spot historical events * Don't exclude from sync * Don't generate output events when rewriting forward extremities * Update output event check * Historical output events * Define output room event type * Notify key changes on state * Don't send our membership event twice * Deduplicate state entries * Tweaks * Remove unnecessary nolint * Fix current state upsert in sync API * Send auth events as outliers, state events as rewrite * Sync API don't consume state events * Process events actually * Improve outlier check * Fix local room check * Remove extra room check, it seems to break the whole damn world * Fix federated join check * Fix nil pointer exception * Better comments on DeduplicateStateEntries * Reflow forced federated joins * Don't force federated join for possibly even local invites * Comment SendEventWithState better * Rewrite room state in sync API storage * Add TODO * Clean up all room data when receiving create event * Don't generate output events for rewrites, but instead notify that state is rewritten on the final new event * Rename to PurgeRoom * Exclude backfilled messages from /sync * Split out rewriting state from updating state from state res Co-authored-by: Kegan Dougal <kegan@matrix.org>
Diffstat (limited to 'syncapi')
-rw-r--r--syncapi/consumers/roomserver.go6
-rw-r--r--syncapi/storage/interface.go3
-rw-r--r--syncapi/storage/postgres/backwards_extremities_table.go15
-rw-r--r--syncapi/storage/postgres/current_room_state_table.go15
-rw-r--r--syncapi/storage/postgres/output_room_events_table.go14
-rw-r--r--syncapi/storage/postgres/output_room_events_topology_table.go15
-rw-r--r--syncapi/storage/shared/syncserver.go23
-rw-r--r--syncapi/storage/sqlite3/backwards_extremities_table.go15
-rw-r--r--syncapi/storage/sqlite3/current_room_state_table.go17
-rw-r--r--syncapi/storage/sqlite3/output_room_events_table.go14
-rw-r--r--syncapi/storage/sqlite3/output_room_events_topology_table.go14
-rw-r--r--syncapi/storage/tables/interface.go7
12 files changed, 157 insertions, 1 deletions
diff --git a/syncapi/consumers/roomserver.go b/syncapi/consumers/roomserver.go
index b6ab9bd5..d8d0a298 100644
--- a/syncapi/consumers/roomserver.go
+++ b/syncapi/consumers/roomserver.go
@@ -149,6 +149,12 @@ func (s *OutputRoomEventConsumer) onNewRoomEvent(
}
}
+ if msg.RewritesState {
+ if err = s.db.PurgeRoom(ctx, ev.RoomID()); err != nil {
+ return fmt.Errorf("s.db.PurgeRoom: %w", err)
+ }
+ }
+
pduPos, err := s.db.WriteEvent(
ctx,
&ev,
diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go
index 807c7f5e..ce7f1c15 100644
--- a/syncapi/storage/interface.go
+++ b/syncapi/storage/interface.go
@@ -43,6 +43,9 @@ type Database interface {
// Returns an error if there was a problem inserting this event.
WriteEvent(ctx context.Context, ev *gomatrixserverlib.HeaderedEvent, addStateEvents []gomatrixserverlib.HeaderedEvent,
addStateEventIDs []string, removeStateEventIDs []string, transactionID *api.TransactionID, excludeFromSync bool) (types.StreamPosition, error)
+ // PurgeRoom completely purges room state from the sync API. This is done when
+ // receiving an output event that completely resets the state.
+ PurgeRoom(ctx context.Context, roomID string) error
// GetStateEvent returns the Matrix state event of a given type for a given room with a given state key
// If no event could be found, returns nil
// If there was an issue during the retrieval, returns an error
diff --git a/syncapi/storage/postgres/backwards_extremities_table.go b/syncapi/storage/postgres/backwards_extremities_table.go
index 71569a10..13056588 100644
--- a/syncapi/storage/postgres/backwards_extremities_table.go
+++ b/syncapi/storage/postgres/backwards_extremities_table.go
@@ -19,6 +19,7 @@ import (
"database/sql"
"github.com/matrix-org/dendrite/internal"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
)
@@ -46,10 +47,14 @@ const selectBackwardExtremitiesForRoomSQL = "" +
const deleteBackwardExtremitySQL = "" +
"DELETE FROM syncapi_backward_extremities WHERE room_id = $1 AND prev_event_id = $2"
+const deleteBackwardExtremitiesForRoomSQL = "" +
+ "DELETE FROM syncapi_backward_extremities WHERE room_id = $1"
+
type backwardExtremitiesStatements struct {
insertBackwardExtremityStmt *sql.Stmt
selectBackwardExtremitiesForRoomStmt *sql.Stmt
deleteBackwardExtremityStmt *sql.Stmt
+ deleteBackwardExtremitiesForRoomStmt *sql.Stmt
}
func NewPostgresBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities, error) {
@@ -67,6 +72,9 @@ func NewPostgresBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremiti
if s.deleteBackwardExtremityStmt, err = db.Prepare(deleteBackwardExtremitySQL); err != nil {
return nil, err
}
+ if s.deleteBackwardExtremitiesForRoomStmt, err = db.Prepare(deleteBackwardExtremitiesForRoomSQL); err != nil {
+ return nil, err
+ }
return s, nil
}
@@ -105,3 +113,10 @@ func (s *backwardExtremitiesStatements) DeleteBackwardExtremity(
_, err = txn.Stmt(s.deleteBackwardExtremityStmt).ExecContext(ctx, roomID, knownEventID)
return
}
+
+func (s *backwardExtremitiesStatements) DeleteBackwardExtremitiesForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) (err error) {
+ _, err = sqlutil.TxStmt(txn, s.deleteBackwardExtremitiesForRoomStmt).ExecContext(ctx, roomID)
+ return err
+}
diff --git a/syncapi/storage/postgres/current_room_state_table.go b/syncapi/storage/postgres/current_room_state_table.go
index 5cb7baad..0ca9eed9 100644
--- a/syncapi/storage/postgres/current_room_state_table.go
+++ b/syncapi/storage/postgres/current_room_state_table.go
@@ -69,6 +69,9 @@ const upsertRoomStateSQL = "" +
const deleteRoomStateByEventIDSQL = "" +
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
+const DeleteRoomStateForRoomSQL = "" +
+ "DELETE FROM syncapi_current_room_state WHERE event_id = $1"
+
const selectRoomIDsWithMembershipSQL = "" +
"SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
@@ -98,6 +101,7 @@ const selectEventsWithEventIDsSQL = "" +
type currentRoomStateStatements struct {
upsertRoomStateStmt *sql.Stmt
deleteRoomStateByEventIDStmt *sql.Stmt
+ DeleteRoomStateForRoomStmt *sql.Stmt
selectRoomIDsWithMembershipStmt *sql.Stmt
selectCurrentStateStmt *sql.Stmt
selectJoinedUsersStmt *sql.Stmt
@@ -117,6 +121,9 @@ func NewPostgresCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, erro
if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil {
return nil, err
}
+ if s.DeleteRoomStateForRoomStmt, err = db.Prepare(DeleteRoomStateForRoomSQL); err != nil {
+ return nil, err
+ }
if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil {
return nil, err
}
@@ -214,6 +221,14 @@ func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
return err
}
+func (s *currentRoomStateStatements) DeleteRoomStateForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) error {
+ stmt := sqlutil.TxStmt(txn, s.DeleteRoomStateForRoomStmt)
+ _, err := stmt.ExecContext(ctx, roomID)
+ return err
+}
+
func (s *currentRoomStateStatements) UpsertRoomState(
ctx context.Context, txn *sql.Tx,
event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition,
diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go
index 5315de24..4b2101bb 100644
--- a/syncapi/storage/postgres/output_room_events_table.go
+++ b/syncapi/storage/postgres/output_room_events_table.go
@@ -115,6 +115,9 @@ const selectStateInRangeSQL = "" +
" ORDER BY id ASC" +
" LIMIT $8"
+const deleteEventsForRoomSQL = "" +
+ "DELETE FROM syncapi_output_room_events WHERE room_id = $1"
+
type outputRoomEventsStatements struct {
insertEventStmt *sql.Stmt
selectEventsStmt *sql.Stmt
@@ -124,6 +127,7 @@ type outputRoomEventsStatements struct {
selectEarlyEventsStmt *sql.Stmt
selectStateInRangeStmt *sql.Stmt
updateEventJSONStmt *sql.Stmt
+ deleteEventsForRoomStmt *sql.Stmt
}
func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
@@ -156,6 +160,9 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
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
}
@@ -395,6 +402,13 @@ func (s *outputRoomEventsStatements) SelectEvents(
return rowsToStreamEvents(rows)
}
+func (s *outputRoomEventsStatements) DeleteEventsForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) (err error) {
+ _, err = sqlutil.TxStmt(txn, s.deleteEventsForRoomStmt).ExecContext(ctx, roomID)
+ return err
+}
+
func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
var result []types.StreamEvent
for rows.Next() {
diff --git a/syncapi/storage/postgres/output_room_events_topology_table.go b/syncapi/storage/postgres/output_room_events_topology_table.go
index 1ab3a1dc..cbd20a07 100644
--- a/syncapi/storage/postgres/output_room_events_topology_table.go
+++ b/syncapi/storage/postgres/output_room_events_topology_table.go
@@ -19,6 +19,7 @@ import (
"database/sql"
"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"
@@ -71,12 +72,16 @@ const selectMaxPositionInTopologySQL = "" +
"SELECT MAX(topological_position) FROM syncapi_output_room_events_topology WHERE room_id=$1" +
") ORDER BY stream_position DESC LIMIT 1"
+const deleteTopologyForRoomSQL = "" +
+ "DELETE FROM syncapi_output_room_events_topology WHERE room_id = $1"
+
type outputRoomEventsTopologyStatements struct {
insertEventInTopologyStmt *sql.Stmt
selectEventIDsInRangeASCStmt *sql.Stmt
selectEventIDsInRangeDESCStmt *sql.Stmt
selectPositionInTopologyStmt *sql.Stmt
selectMaxPositionInTopologyStmt *sql.Stmt
+ deleteTopologyForRoomStmt *sql.Stmt
}
func NewPostgresTopologyTable(db *sql.DB) (tables.Topology, error) {
@@ -100,6 +105,9 @@ func NewPostgresTopologyTable(db *sql.DB) (tables.Topology, error) {
if s.selectMaxPositionInTopologyStmt, err = db.Prepare(selectMaxPositionInTopologySQL); err != nil {
return nil, err
}
+ if s.deleteTopologyForRoomStmt, err = db.Prepare(deleteTopologyForRoomSQL); err != nil {
+ return nil, err
+ }
return s, nil
}
@@ -167,3 +175,10 @@ func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
err = s.selectMaxPositionInTopologyStmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos)
return
}
+
+func (s *outputRoomEventsTopologyStatements) DeleteTopologyForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) (err error) {
+ _, err = sqlutil.TxStmt(txn, s.deleteTopologyForRoomStmt).ExecContext(ctx, roomID)
+ return err
+}
diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go
index 94580adb..05a8768e 100644
--- a/syncapi/storage/shared/syncserver.go
+++ b/syncapi/storage/shared/syncserver.go
@@ -276,6 +276,29 @@ func (d *Database) handleBackwardExtremities(ctx context.Context, txn *sql.Tx, e
return nil
}
+func (d *Database) PurgeRoom(
+ ctx context.Context, roomID string,
+) error {
+ return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
+ // If the event is a create event then we'll delete all of the existing
+ // data for the room. The only reason that a create event would be replayed
+ // to us in this way is if we're about to receive the entire room state.
+ if err := d.CurrentRoomState.DeleteRoomStateForRoom(ctx, txn, roomID); err != nil {
+ return fmt.Errorf("d.CurrentRoomState.DeleteRoomStateForRoom: %w", err)
+ }
+ if err := d.OutputEvents.DeleteEventsForRoom(ctx, txn, roomID); err != nil {
+ return fmt.Errorf("d.Events.DeleteEventsForRoom: %w", err)
+ }
+ if err := d.Topology.DeleteTopologyForRoom(ctx, txn, roomID); err != nil {
+ return fmt.Errorf("d.Topology.DeleteTopologyForRoom: %w", err)
+ }
+ if err := d.BackwardExtremities.DeleteBackwardExtremitiesForRoom(ctx, txn, roomID); err != nil {
+ return fmt.Errorf("d.BackwardExtremities.DeleteBackwardExtremitiesForRoom: %w", err)
+ }
+ return nil
+ })
+}
+
func (d *Database) WriteEvent(
ctx context.Context,
ev *gomatrixserverlib.HeaderedEvent,
diff --git a/syncapi/storage/sqlite3/backwards_extremities_table.go b/syncapi/storage/sqlite3/backwards_extremities_table.go
index 116c33dc..9a81e8e7 100644
--- a/syncapi/storage/sqlite3/backwards_extremities_table.go
+++ b/syncapi/storage/sqlite3/backwards_extremities_table.go
@@ -19,6 +19,7 @@ import (
"database/sql"
"github.com/matrix-org/dendrite/internal"
+ "github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/syncapi/storage/tables"
)
@@ -46,11 +47,15 @@ const selectBackwardExtremitiesForRoomSQL = "" +
const deleteBackwardExtremitySQL = "" +
"DELETE FROM syncapi_backward_extremities WHERE room_id = $1 AND prev_event_id = $2"
+const deleteBackwardExtremitiesForRoomSQL = "" +
+ "DELETE FROM syncapi_backward_extremities WHERE room_id = $1"
+
type backwardExtremitiesStatements struct {
db *sql.DB
insertBackwardExtremityStmt *sql.Stmt
selectBackwardExtremitiesForRoomStmt *sql.Stmt
deleteBackwardExtremityStmt *sql.Stmt
+ deleteBackwardExtremitiesForRoomStmt *sql.Stmt
}
func NewSqliteBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities, error) {
@@ -70,6 +75,9 @@ func NewSqliteBackwardsExtremitiesTable(db *sql.DB) (tables.BackwardsExtremities
if s.deleteBackwardExtremityStmt, err = db.Prepare(deleteBackwardExtremitySQL); err != nil {
return nil, err
}
+ if s.deleteBackwardExtremitiesForRoomStmt, err = db.Prepare(deleteBackwardExtremitiesForRoomSQL); err != nil {
+ return nil, err
+ }
return s, nil
}
@@ -108,3 +116,10 @@ func (s *backwardExtremitiesStatements) DeleteBackwardExtremity(
_, err = txn.Stmt(s.deleteBackwardExtremityStmt).ExecContext(ctx, roomID, knownEventID)
return err
}
+
+func (s *backwardExtremitiesStatements) DeleteBackwardExtremitiesForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) (err error) {
+ _, err = sqlutil.TxStmt(txn, s.deleteBackwardExtremitiesForRoomStmt).ExecContext(ctx, roomID)
+ return err
+}
diff --git a/syncapi/storage/sqlite3/current_room_state_table.go b/syncapi/storage/sqlite3/current_room_state_table.go
index 6f822c90..13d23be5 100644
--- a/syncapi/storage/sqlite3/current_room_state_table.go
+++ b/syncapi/storage/sqlite3/current_room_state_table.go
@@ -51,12 +51,15 @@ CREATE UNIQUE INDEX IF NOT EXISTS syncapi_event_id_idx ON syncapi_current_room_s
const upsertRoomStateSQL = "" +
"INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, contains_url, state_key, headered_event_json, membership, added_at)" +
" VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)" +
- " ON CONFLICT (event_id, room_id, type, sender, contains_url)" +
+ " ON CONFLICT (room_id, type, state_key)" +
" DO UPDATE SET event_id = $2, sender=$4, contains_url=$5, headered_event_json = $7, membership = $8, added_at = $9"
const deleteRoomStateByEventIDSQL = "" +
"DELETE FROM syncapi_current_room_state WHERE event_id = $1"
+const DeleteRoomStateForRoomSQL = "" +
+ "DELETE FROM syncapi_current_room_state WHERE event_id = $1"
+
const selectRoomIDsWithMembershipSQL = "" +
"SELECT DISTINCT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
@@ -88,6 +91,7 @@ type currentRoomStateStatements struct {
streamIDStatements *streamIDStatements
upsertRoomStateStmt *sql.Stmt
deleteRoomStateByEventIDStmt *sql.Stmt
+ DeleteRoomStateForRoomStmt *sql.Stmt
selectRoomIDsWithMembershipStmt *sql.Stmt
selectCurrentStateStmt *sql.Stmt
selectJoinedUsersStmt *sql.Stmt
@@ -109,6 +113,9 @@ func NewSqliteCurrentRoomStateTable(db *sql.DB, streamID *streamIDStatements) (t
if s.deleteRoomStateByEventIDStmt, err = db.Prepare(deleteRoomStateByEventIDSQL); err != nil {
return nil, err
}
+ if s.DeleteRoomStateForRoomStmt, err = db.Prepare(DeleteRoomStateForRoomSQL); err != nil {
+ return nil, err
+ }
if s.selectRoomIDsWithMembershipStmt, err = db.Prepare(selectRoomIDsWithMembershipSQL); err != nil {
return nil, err
}
@@ -203,6 +210,14 @@ func (s *currentRoomStateStatements) DeleteRoomStateByEventID(
return err
}
+func (s *currentRoomStateStatements) DeleteRoomStateForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) error {
+ stmt := sqlutil.TxStmt(txn, s.DeleteRoomStateForRoomStmt)
+ _, err := stmt.ExecContext(ctx, roomID)
+ return err
+}
+
func (s *currentRoomStateStatements) UpsertRoomState(
ctx context.Context, txn *sql.Tx,
event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition,
diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go
index f10d0106..587a4072 100644
--- a/syncapi/storage/sqlite3/output_room_events_table.go
+++ b/syncapi/storage/sqlite3/output_room_events_table.go
@@ -103,6 +103,9 @@ const selectStateInRangeSQL = "" +
" ORDER BY id ASC" +
" LIMIT $8" // limit
+const deleteEventsForRoomSQL = "" +
+ "DELETE FROM syncapi_output_room_events WHERE room_id = $1"
+
type outputRoomEventsStatements struct {
db *sql.DB
streamIDStatements *streamIDStatements
@@ -114,6 +117,7 @@ type outputRoomEventsStatements struct {
selectEarlyEventsStmt *sql.Stmt
selectStateInRangeStmt *sql.Stmt
updateEventJSONStmt *sql.Stmt
+ deleteEventsForRoomStmt *sql.Stmt
}
func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Events, error) {
@@ -149,6 +153,9 @@ func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Even
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
}
@@ -410,6 +417,13 @@ func (s *outputRoomEventsStatements) SelectEvents(
return returnEvents, nil
}
+func (s *outputRoomEventsStatements) DeleteEventsForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) (err error) {
+ _, err = sqlutil.TxStmt(txn, s.deleteEventsForRoomStmt).ExecContext(ctx, roomID)
+ return err
+}
+
func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
var result []types.StreamEvent
for rows.Next() {
diff --git a/syncapi/storage/sqlite3/output_room_events_topology_table.go b/syncapi/storage/sqlite3/output_room_events_topology_table.go
index d8c97b7e..d3ba9af6 100644
--- a/syncapi/storage/sqlite3/output_room_events_topology_table.go
+++ b/syncapi/storage/sqlite3/output_room_events_topology_table.go
@@ -65,6 +65,9 @@ const selectMaxPositionInTopologySQL = "" +
"SELECT MAX(topological_position), stream_position FROM syncapi_output_room_events_topology" +
" WHERE room_id = $1 ORDER BY stream_position DESC"
+const deleteTopologyForRoomSQL = "" +
+ "DELETE FROM syncapi_output_room_events_topology WHERE room_id = $1"
+
type outputRoomEventsTopologyStatements struct {
db *sql.DB
insertEventInTopologyStmt *sql.Stmt
@@ -72,6 +75,7 @@ type outputRoomEventsTopologyStatements struct {
selectEventIDsInRangeDESCStmt *sql.Stmt
selectPositionInTopologyStmt *sql.Stmt
selectMaxPositionInTopologyStmt *sql.Stmt
+ deleteTopologyForRoomStmt *sql.Stmt
}
func NewSqliteTopologyTable(db *sql.DB) (tables.Topology, error) {
@@ -97,6 +101,9 @@ func NewSqliteTopologyTable(db *sql.DB) (tables.Topology, error) {
if s.selectMaxPositionInTopologyStmt, err = db.Prepare(selectMaxPositionInTopologySQL); err != nil {
return nil, err
}
+ if s.deleteTopologyForRoomStmt, err = db.Prepare(deleteTopologyForRoomSQL); err != nil {
+ return nil, err
+ }
return s, nil
}
@@ -164,3 +171,10 @@ func (s *outputRoomEventsTopologyStatements) SelectMaxPositionInTopology(
err = stmt.QueryRowContext(ctx, roomID).Scan(&pos, &spos)
return
}
+
+func (s *outputRoomEventsTopologyStatements) DeleteTopologyForRoom(
+ ctx context.Context, txn *sql.Tx, roomID string,
+) (err error) {
+ _, err = sqlutil.TxStmt(txn, s.deleteTopologyForRoomStmt).ExecContext(ctx, roomID)
+ return err
+}
diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go
index 631746c6..da095be5 100644
--- a/syncapi/storage/tables/interface.go
+++ b/syncapi/storage/tables/interface.go
@@ -60,6 +60,8 @@ type Events interface {
SelectEarlyEvents(ctx context.Context, txn *sql.Tx, roomID string, r types.Range, limit int) ([]types.StreamEvent, error)
SelectEvents(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error)
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)
}
// Topology keeps track of the depths and stream positions for all events.
@@ -77,6 +79,8 @@ type Topology interface {
SelectPositionInTopology(ctx context.Context, txn *sql.Tx, eventID string) (depth, spos types.StreamPosition, err error)
// SelectMaxPositionInTopology returns the event which has the highest depth, and if there are multiple, the event with the highest stream position.
SelectMaxPositionInTopology(ctx context.Context, txn *sql.Tx, roomID string) (depth types.StreamPosition, spos types.StreamPosition, err error)
+ // DeleteTopologyForRoom removes all topological information for a room. This should only be done when removing the room entirely.
+ DeleteTopologyForRoom(ctx context.Context, txn *sql.Tx, roomID string) (err error)
}
type CurrentRoomState interface {
@@ -84,6 +88,7 @@ type CurrentRoomState interface {
SelectEventsWithEventIDs(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]types.StreamEvent, error)
UpsertRoomState(ctx context.Context, txn *sql.Tx, event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition) error
DeleteRoomStateByEventID(ctx context.Context, txn *sql.Tx, eventID string) error
+ DeleteRoomStateForRoom(ctx context.Context, txn *sql.Tx, roomID string) error
// SelectCurrentState returns all the current state events for the given room.
SelectCurrentState(ctx context.Context, txn *sql.Tx, roomID string, stateFilter *gomatrixserverlib.StateFilter) ([]gomatrixserverlib.HeaderedEvent, error)
// SelectRoomIDsWithMembership returns the list of room IDs which have the given user in the given membership state.
@@ -118,6 +123,8 @@ type BackwardsExtremities interface {
SelectBackwardExtremitiesForRoom(ctx context.Context, roomID string) (bwExtrems map[string][]string, err error)
// DeleteBackwardExtremity removes a backwards extremity for a room, if one existed.
DeleteBackwardExtremity(ctx context.Context, txn *sql.Tx, roomID, knownEventID string) (err error)
+ // DeleteBackwardExtremitiesFoorRoomID removes all backward extremities for a room. This should only be done when removing the room entirely.
+ DeleteBackwardExtremitiesForRoom(ctx context.Context, txn *sql.Tx, roomID string) (err error)
}
// SendToDevice tracks send-to-device messages which are sent to individual