sigb.us/dendrite
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases 1 Wiki Activity

Consolidate purge statements as they need to be prepared after other tables are created

Browse source
This commit is contained in:
Neil Alexander 2022-08-22 13:48:09 +01:00
parent 468d4b5bbe
commit 4e4fc400a2
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
26 changed files with 192 additions and 240 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
roomserver/storage/postgres/event_json_table.go
View file

@ -54,16 +54,9 @@ const bulkSelectEventJSONSQL = "" +
" WHERE event_nid = ANY($1)" +
" ORDER BY event_nid ASC"
const purgeEventJSONSQL = `
DELETE FROM roomserver_event_json WHERE event_nid = ANY(
SELECT event_nid FROM roomserver_events WHERE room_nid = $1
)
`
type eventJSONStatements struct {
insertEventJSONStmt *sql.Stmt
bulkSelectEventJSONStmt *sql.Stmt
purgeEventJSONStmt *sql.Stmt
}
func CreateEventJSONTable(db *sql.DB) error {
@ -77,7 +70,6 @@ func PrepareEventJSONTable(db *sql.DB) (tables.EventJSON, error) {
return s, sqlutil.StatementList{
{&s.insertEventJSONStmt, insertEventJSONSQL},
{&s.bulkSelectEventJSONStmt, bulkSelectEventJSONSQL},
{&s.purgeEventJSONStmt, purgeEventJSONSQL},
}.Prepare(db)
}
@ -115,10 +107,3 @@ func (s *eventJSONStatements) BulkSelectEventJSON(
}
return results[:i], rows.Err()
}
func (s *eventJSONStatements) PurgeEventJSONs(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeEventJSONStmt).ExecContext(ctx, roomNID)
return err
}

12
roomserver/storage/postgres/events_table.go
View file

@ -147,9 +147,6 @@ const selectRoomNIDsForEventNIDsSQL = "" +
const selectEventRejectedSQL = "" +
"SELECT is_rejected FROM roomserver_events WHERE room_nid = $1 AND event_id = $2"
const purgeEventsSQL = "" +
"DELETE FROM roomserver_events WHERE room_nid = $1"
type eventStatements struct {
insertEventStmt *sql.Stmt
selectEventStmt *sql.Stmt
@ -169,7 +166,6 @@ type eventStatements struct {
selectMaxEventDepthStmt *sql.Stmt
selectRoomNIDsForEventNIDsStmt *sql.Stmt
selectEventRejectedStmt *sql.Stmt
purgeEventsStmt *sql.Stmt
}
func CreateEventsTable(db *sql.DB) error {
@ -199,7 +195,6 @@ func PrepareEventsTable(db *sql.DB) (tables.Events, error) {
{&s.selectMaxEventDepthStmt, selectMaxEventDepthSQL},
{&s.selectRoomNIDsForEventNIDsStmt, selectRoomNIDsForEventNIDsSQL},
{&s.selectEventRejectedStmt, selectEventRejectedSQL},
{&s.purgeEventsStmt, purgeEventsSQL},
}.Prepare(db)
}
@ -576,10 +571,3 @@ func (s *eventStatements) SelectEventRejected(
err = stmt.QueryRowContext(ctx, roomNID, eventID).Scan(&rejected)
return
}
func (s *eventStatements) PurgeEvents(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeEventsStmt).ExecContext(ctx, roomNID)
return err
}

12
roomserver/storage/postgres/invite_table.go
View file

@ -75,14 +75,10 @@ const updateInviteRetiredSQL = "" +
" WHERE room_nid = $1 AND target_nid = $2 AND NOT retired" +
" RETURNING invite_event_id"
const purgeInvitesSQL = "" +
"DELETE FROM roomserver_invites WHERE room_nid = $1"
type inviteStatements struct {
insertInviteEventStmt *sql.Stmt
selectInviteActiveForUserInRoomStmt *sql.Stmt
updateInviteRetiredStmt *sql.Stmt
purgeInvitesStmt *sql.Stmt
}
func CreateInvitesTable(db *sql.DB) error {
@ -97,7 +93,6 @@ func PrepareInvitesTable(db *sql.DB) (tables.Invites, error) {
{&s.insertInviteEventStmt, insertInviteEventSQL},
{&s.selectInviteActiveForUserInRoomStmt, selectInviteActiveForUserInRoomSQL},
{&s.updateInviteRetiredStmt, updateInviteRetiredSQL},
{&s.purgeInvitesStmt, purgeInvitesSQL},
}.Prepare(db)
}
@ -168,10 +163,3 @@ func (s *inviteStatements) SelectInviteActiveForUserInRoom(
}
return result, eventIDs, rows.Err()
}
func (s *inviteStatements) PurgeInvites(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeInvitesStmt).ExecContext(ctx, roomNID)
return err
}

12
roomserver/storage/postgres/membership_table.go
View file

@ -153,9 +153,6 @@ const selectServerInRoomSQL = "" +
" JOIN roomserver_event_state_keys ON roomserver_membership.target_nid = roomserver_event_state_keys.event_state_key_nid" +
" WHERE membership_nid = $1 AND room_nid = $2 AND event_state_key LIKE '%:' || $3 LIMIT 1"
const purgeMembershipsSQL = "" +
"DELETE FROM roomserver_membership WHERE room_nid = $1"
type membershipStatements struct {
insertMembershipStmt *sql.Stmt
selectMembershipForUpdateStmt *sql.Stmt
@ -173,7 +170,6 @@ type membershipStatements struct {
selectLocalServerInRoomStmt *sql.Stmt
selectServerInRoomStmt *sql.Stmt
deleteMembershipStmt *sql.Stmt
purgeMembershipsStmt *sql.Stmt
}
func CreateMembershipTable(db *sql.DB) error {
@ -209,7 +205,6 @@ func PrepareMembershipTable(db *sql.DB) (tables.Membership, error) {
{&s.selectLocalServerInRoomStmt, selectLocalServerInRoomSQL},
{&s.selectServerInRoomStmt, selectServerInRoomSQL},
{&s.deleteMembershipStmt, deleteMembershipSQL},
{&s.purgeMembershipsStmt, purgeMembershipsSQL},
}.Prepare(db)
}
@ -441,10 +436,3 @@ func (s *membershipStatements) DeleteMembership(
)
return err
}
func (s *membershipStatements) PurgeMemberships(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeMembershipsStmt).ExecContext(ctx, roomNID)
return err
}

15
roomserver/storage/postgres/previous_events_table.go
View file

@ -59,16 +59,9 @@ const selectPreviousEventExistsSQL = "" +
"SELECT 1 FROM roomserver_previous_events" +
" WHERE previous_event_id = $1 AND previous_reference_sha256 = $2"
const purgePreviousEventsSQL = `
DELETE FROM roomserver_previous_events WHERE event_nids && ANY(
SELECT ARRAY_AGG(event_nid) FROM roomserver_events WHERE room_nid = $1
)
`
type previousEventStatements struct {
insertPreviousEventStmt *sql.Stmt
selectPreviousEventExistsStmt *sql.Stmt
purgePreviousEventsStmt *sql.Stmt
}
func CreatePrevEventsTable(db *sql.DB) error {
@ -82,7 +75,6 @@ func PreparePrevEventsTable(db *sql.DB) (tables.PreviousEvents, error) {
return s, sqlutil.StatementList{
{&s.insertPreviousEventStmt, insertPreviousEventSQL},
{&s.selectPreviousEventExistsStmt, selectPreviousEventExistsSQL},
{&s.purgePreviousEventsStmt, purgePreviousEventsSQL},
}.Prepare(db)
}
@ -109,10 +101,3 @@ func (s *previousEventStatements) SelectPreviousEventExists(
stmt := sqlutil.TxStmt(txn, s.selectPreviousEventExistsStmt)
return stmt.QueryRowContext(ctx, eventID, eventReferenceSHA256).Scan(&ok)
}
func (s *previousEventStatements) PurgePreviousEvents(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgePreviousEventsStmt).ExecContext(ctx, roomNID)
return err
}

12
roomserver/storage/postgres/published_table.go
View file

@ -43,14 +43,10 @@ const selectAllPublishedSQL = "" +
const selectPublishedSQL = "" +
"SELECT published FROM roomserver_published WHERE room_id = $1"
const purgePublishedSQL = "" +
"DELETE FROM roomserver_published WHERE room_id = $1"
type publishedStatements struct {
upsertPublishedStmt *sql.Stmt
selectAllPublishedStmt *sql.Stmt
selectPublishedStmt *sql.Stmt
purgePublishedStmt *sql.Stmt
}
func CreatePublishedTable(db *sql.DB) error {
@ -65,7 +61,6 @@ func PreparePublishedTable(db *sql.DB) (tables.Published, error) {
{&s.upsertPublishedStmt, upsertPublishedSQL},
{&s.selectAllPublishedStmt, selectAllPublishedSQL},
{&s.selectPublishedStmt, selectPublishedSQL},
{&s.purgePublishedStmt, purgePublishedSQL},
}.Prepare(db)
}
@ -109,10 +104,3 @@ func (s *publishedStatements) SelectAllPublishedRooms(
}
return roomIDs, rows.Err()
}
func (s *publishedStatements) PurgePublished(
ctx context.Context, txn *sql.Tx, roomID string,
) error {
_, err := sqlutil.TxStmt(txn, s.purgePublishedStmt).ExecContext(ctx, roomID)
return err
}

159
roomserver/storage/postgres/purge_statements.go Normal file
View file

@ -0,0 +1,159 @@
package postgres
import (
"context"
"database/sql"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/types"
)
const purgeEventJSONSQL = "" +
"DELETE FROM roomserver_event_json WHERE event_nid = ANY(" +
" SELECT event_nid FROM roomserver_events WHERE room_nid = $1" +
")"
const purgeEventsSQL = "" +
"DELETE FROM roomserver_events WHERE room_nid = $1"
const purgeInvitesSQL = "" +
"DELETE FROM roomserver_invites WHERE room_nid = $1"
const purgeMembershipsSQL = "" +
"DELETE FROM roomserver_membership WHERE room_nid = $1"
const purgePreviousEventsSQL = "" +
"DELETE FROM roomserver_previous_events WHERE event_nids && ANY(" +
" SELECT ARRAY_AGG(event_nid) FROM roomserver_events WHERE room_nid = $1" +
")"
const purgePublishedSQL = "" +
"DELETE FROM roomserver_published WHERE room_id = $1"
const purgeRedactionsSQL = "" +
"DELETE FROM roomserver_redactions WHERE redaction_event_id = ANY(" +
" SELECT event_id FROM roomserver_events WHERE room_nid = $1" +
")"
const purgeRoomAliasesSQL = "" +
"DELETE FROM roomserver_room_aliases WHERE room_id = $1"
const purgeRoomSQL = "" +
"DELETE FROM roomserver_rooms WHERE room_nid = $1"
const purgeStateBlockEntriesSQL = "" +
"DELETE FROM roomserver_state_block WHERE state_block_nid = ANY(" +
" SELECT DISTINCT UNNEST(state_block_nids) FROM roomserver_state_snapshots WHERE room_nid = $1" +
")"
const purgeStateSnapshotEntriesSQL = "" +
"DELETE FROM roomserver_state_snapshots WHERE room_nid = $1"
type purgeStatements struct {
purgeEventJSONStmt *sql.Stmt
purgeEventsStmt *sql.Stmt
purgeInvitesStmt *sql.Stmt
purgeMembershipsStmt *sql.Stmt
purgePreviousEventsStmt *sql.Stmt
purgePublishedStmt *sql.Stmt
purgeRedactionStmt *sql.Stmt
purgeRoomAliasesStmt *sql.Stmt
purgeRoomStmt *sql.Stmt
purgeStateBlockEntriesStmt *sql.Stmt
purgeStateSnapshotEntriesStmt *sql.Stmt
}
func PreparePurgeStatements(db *sql.DB) (*purgeStatements, error) {
s := &purgeStatements{}
return s, sqlutil.StatementList{
{&s.purgeEventJSONStmt, purgeEventJSONSQL},
{&s.purgeEventsStmt, purgeEventsSQL},
{&s.purgeInvitesStmt, purgeInvitesSQL},
{&s.purgeMembershipsStmt, purgeMembershipsSQL},
{&s.purgePublishedStmt, purgePublishedSQL},
{&s.purgePreviousEventsStmt, purgePreviousEventsSQL},
{&s.purgeRedactionStmt, purgeRedactionsSQL},
{&s.purgeRoomAliasesStmt, purgeRoomAliasesSQL},
{&s.purgeRoomStmt, purgeRoomSQL},
{&s.purgeStateBlockEntriesStmt, purgeStateBlockEntriesSQL},
{&s.purgeStateSnapshotEntriesStmt, purgeStateSnapshotEntriesSQL},
}.Prepare(db)
}
func (s *purgeStatements) PurgeEventJSONs(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeEventJSONStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgeEvents(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeEventsStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgeInvites(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeInvitesStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgeMemberships(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeMembershipsStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgePreviousEvents(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgePreviousEventsStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgePublished(
ctx context.Context, txn *sql.Tx, roomID string,
) error {
_, err := sqlutil.TxStmt(txn, s.purgePublishedStmt).ExecContext(ctx, roomID)
return err
}
func (s *purgeStatements) PurgeRedactions(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeRedactionStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgeRoomAliases(
ctx context.Context, txn *sql.Tx, roomID string,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeRoomAliasesStmt).ExecContext(ctx, roomID)
return err
}
func (s *purgeStatements) PurgeRoom(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeRoomStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgeStateBlocks(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeStateBlockEntriesStmt).ExecContext(ctx, roomNID)
return err
}
func (s *purgeStatements) PurgeStateSnapshots(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeStateSnapshotEntriesStmt).ExecContext(ctx, roomNID)
return err
}

16
roomserver/storage/postgres/redactions_table.go
View file

@ -20,7 +20,6 @@ import (
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
const redactionsSchema = `
@ -53,18 +52,11 @@ const selectRedactionInfoByEventBeingRedactedSQL = "" +
const markRedactionValidatedSQL = "" +
" UPDATE roomserver_redactions SET validated = $2 WHERE redaction_event_id = $1"
const purgeRedactionsSQL = `
DELETE FROM roomserver_redactions WHERE redaction_event_id = ANY(
SELECT event_id FROM roomserver_events WHERE room_nid = $1
)
`
type redactionStatements struct {
insertRedactionStmt *sql.Stmt
selectRedactionInfoByRedactionEventIDStmt *sql.Stmt
selectRedactionInfoByEventBeingRedactedStmt *sql.Stmt
markRedactionValidatedStmt *sql.Stmt
purgeRedactionStmt *sql.Stmt
}
func CreateRedactionsTable(db *sql.DB) error {
@ -80,7 +72,6 @@ func PrepareRedactionsTable(db *sql.DB) (tables.Redactions, error) {
{&s.selectRedactionInfoByRedactionEventIDStmt, selectRedactionInfoByRedactionEventIDSQL},
{&s.selectRedactionInfoByEventBeingRedactedStmt, selectRedactionInfoByEventBeingRedactedSQL},
{&s.markRedactionValidatedStmt, markRedactionValidatedSQL},
{&s.purgeRedactionStmt, purgeRedactionsSQL},
}.Prepare(db)
}
@ -129,10 +120,3 @@ func (s *redactionStatements) MarkRedactionValidated(
_, err := stmt.ExecContext(ctx, redactionEventID, validated)
return err
}
func (s *redactionStatements) PurgeRedactions(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeRedactionStmt).ExecContext(ctx, roomNID)
return err
}

12
roomserver/storage/postgres/room_aliases_table.go
View file

@ -53,16 +53,12 @@ const selectCreatorIDFromAliasSQL = "" +
const deleteRoomAliasSQL = "" +
"DELETE FROM roomserver_room_aliases WHERE alias = $1"
const purgeRoomAliasesSQL = "" +
"DELETE FROM roomserver_room_aliases WHERE room_id = $1"
type roomAliasesStatements struct {
insertRoomAliasStmt *sql.Stmt
selectRoomIDFromAliasStmt *sql.Stmt
selectAliasesFromRoomIDStmt *sql.Stmt
selectCreatorIDFromAliasStmt *sql.Stmt
deleteRoomAliasStmt *sql.Stmt
purgeRoomAliasesStmt *sql.Stmt
}
func CreateRoomAliasesTable(db *sql.DB) error {
@ -79,7 +75,6 @@ func PrepareRoomAliasesTable(db *sql.DB) (tables.RoomAliases, error) {
{&s.selectAliasesFromRoomIDStmt, selectAliasesFromRoomIDSQL},
{&s.selectCreatorIDFromAliasStmt, selectCreatorIDFromAliasSQL},
{&s.deleteRoomAliasStmt, deleteRoomAliasSQL},
{&s.purgeRoomAliasesStmt, purgeRoomAliasesSQL},
}.Prepare(db)
}
@ -142,10 +137,3 @@ func (s *roomAliasesStatements) DeleteRoomAlias(
_, err = stmt.ExecContext(ctx, alias)
return
}
func (s *roomAliasesStatements) PurgeRoomAliases(
ctx context.Context, txn *sql.Tx, roomID string,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeRoomAliasesStmt).ExecContext(ctx, roomID)
return err
}

12
roomserver/storage/postgres/rooms_table.go
View file

@ -82,9 +82,6 @@ const bulkSelectRoomIDsSQL = "" +
const bulkSelectRoomNIDsSQL = "" +
"SELECT room_nid FROM roomserver_rooms WHERE room_id = ANY($1)"
const purgeRoomSQL = "" +
"DELETE FROM roomserver_rooms WHERE room_nid = $1"
type roomStatements struct {
insertRoomNIDStmt *sql.Stmt
selectRoomNIDStmt *sql.Stmt
@ -96,7 +93,6 @@ type roomStatements struct {
selectRoomIDsStmt *sql.Stmt
bulkSelectRoomIDsStmt *sql.Stmt
bulkSelectRoomNIDsStmt *sql.Stmt
purgeRoomStmt *sql.Stmt
}
func CreateRoomsTable(db *sql.DB) error {
@ -118,7 +114,6 @@ func PrepareRoomsTable(db *sql.DB) (tables.Rooms, error) {
{&s.selectRoomIDsStmt, selectRoomIDsSQL},
{&s.bulkSelectRoomIDsStmt, bulkSelectRoomIDsSQL},
{&s.bulkSelectRoomNIDsStmt, bulkSelectRoomNIDsSQL},
{&s.purgeRoomStmt, purgeRoomSQL},
}.Prepare(db)
}
@ -293,13 +288,6 @@ func (s *roomStatements) BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, ro
return roomNIDs, nil
}
func (s *roomStatements) PurgeRoom(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeRoomStmt).ExecContext(ctx, roomNID)
return err
}
func roomNIDsAsArray(roomNIDs []types.RoomNID) pq.Int64Array {
nids := make([]int64, len(roomNIDs))
for i := range roomNIDs {

15
roomserver/storage/postgres/state_block_table.go
View file

@ -65,16 +65,9 @@ const bulkSelectStateBlockEntriesSQL = "" +
"SELECT state_block_nid, event_nids" +
" FROM roomserver_state_block WHERE state_block_nid = ANY($1) ORDER BY state_block_nid ASC"
const purgeStateBlockEntriesSQL = `
DELETE FROM roomserver_state_block WHERE state_block_nid = ANY(
SELECT DISTINCT UNNEST(state_block_nids) FROM roomserver_state_snapshots WHERE room_nid = $1
)
`
type stateBlockStatements struct {
insertStateDataStmt *sql.Stmt
bulkSelectStateBlockEntriesStmt *sql.Stmt
purgeStateBlockEntriesStmt *sql.Stmt
}
func CreateStateBlockTable(db *sql.DB) error {
@ -88,7 +81,6 @@ func PrepareStateBlockTable(db *sql.DB) (tables.StateBlock, error) {
return s, sqlutil.StatementList{
{&s.insertStateDataStmt, insertStateDataSQL},
{&s.bulkSelectStateBlockEntriesStmt, bulkSelectStateBlockEntriesSQL},
{&s.purgeStateBlockEntriesStmt, purgeStateBlockEntriesSQL},
}.Prepare(db)
}
@ -141,13 +133,6 @@ func (s *stateBlockStatements) BulkSelectStateBlockEntries(
return results, err
}
func (s *stateBlockStatements) PurgeStateBlocks(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeStateBlockEntriesStmt).ExecContext(ctx, roomNID)
return err
}
func stateBlockNIDsAsArray(stateBlockNIDs []types.StateBlockNID) pq.Int64Array {
nids := make([]int64, len(stateBlockNIDs))
for i := range stateBlockNIDs {

14
roomserver/storage/postgres/state_snapshot_table.go
View file

@ -72,11 +72,6 @@ const bulkSelectStateBlockNIDsSQL = "" +
"SELECT state_snapshot_nid, state_block_nids FROM roomserver_state_snapshots" +
" WHERE state_snapshot_nid = ANY($1) ORDER BY state_snapshot_nid ASC"
// Look up state snapshot NIDs for the given room.
const purgeStateSnapshotEntriesSQL = `
DELETE FROM roomserver_state_snapshots WHERE room_nid = $1
`
// Looks up both the history visibility event and relevant membership events from
// a given domain name from a given state snapshot. This is used to optimise the
// helpers.CheckServerAllowedToSeeEvent function.
@ -106,7 +101,6 @@ type stateSnapshotStatements struct {
insertStateStmt *sql.Stmt
bulkSelectStateBlockNIDsStmt *sql.Stmt
bulkSelectStateForHistoryVisibilityStmt *sql.Stmt
purgeStateSnapshotEntriesStmt *sql.Stmt
}
func CreateStateSnapshotTable(db *sql.DB) error {
@ -121,7 +115,6 @@ func PrepareStateSnapshotTable(db *sql.DB) (tables.StateSnapshot, error) {
{&s.insertStateStmt, insertStateSQL},
{&s.bulkSelectStateBlockNIDsStmt, bulkSelectStateBlockNIDsSQL},
{&s.bulkSelectStateForHistoryVisibilityStmt, bulkSelectStateForHistoryVisibilitySQL},
{&s.purgeStateSnapshotEntriesStmt, purgeStateSnapshotEntriesSQL},
}.Prepare(db)
}
@ -190,10 +183,3 @@ func (s *stateSnapshotStatements) BulkSelectStateForHistoryVisibility(
}
return results, rows.Err()
}
func (s *stateSnapshotStatements) PurgeStateSnapshots(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
_, err := sqlutil.TxStmt(txn, s.purgeStateSnapshotEntriesStmt).ExecContext(ctx, roomNID)
return err
}

5
roomserver/storage/postgres/storage.go
View file

@ -184,6 +184,10 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
if err != nil {
return err
}
purge, err := PreparePurgeStatements(db)
if err != nil {
return err
}
d.Database = shared.Database{
DB: db,
Cache: cache,
@ -201,6 +205,7 @@ func (d *Database) prepare(db *sql.DB, writer sqlutil.Writer, cache caching.Room
MembershipTable: membership,
PublishedTable: published,
RedactionsTable: redactions,
Purge: purge,
}
return nil
}

26
roomserver/storage/shared/storage.go
View file

@ -42,6 +42,7 @@ type Database struct {
MembershipTable tables.Membership
PublishedTable tables.Published
RedactionsTable tables.Redactions
Purge tables.Purge
GetRoomUpdaterFn func(ctx context.Context, roomInfo *types.RoomInfo) (*RoomUpdater, error)
}
@ -1355,6 +1356,9 @@ func (d *Database) ForgetRoom(ctx context.Context, userID, roomID string, forget
// PurgeRoom removes all information about a given room from the roomserver.
// For large rooms this operation may take a considerable amount of time.
func (d *Database) PurgeRoom(ctx context.Context, roomID string) error {
if d.Purge == nil {
return fmt.Errorf("not supported on this database engine")
}
return d.Writer.Do(d.DB, nil, func(txn *sql.Tx) error {
roomNID, err := d.RoomsTable.SelectRoomNID(ctx, txn, roomID)
switch err {
@ -1364,37 +1368,37 @@ func (d *Database) PurgeRoom(ctx context.Context, roomID string) error {
default:
return fmt.Errorf("failed to find room NID: %w", err)
}
if err := d.StateBlockTable.PurgeStateBlocks(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeStateBlocks(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge state blocks: %w", err)
}
if err := d.StateSnapshotTable.PurgeStateSnapshots(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeStateSnapshots(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge state blocks: %w", err)
}
if err := d.InvitesTable.PurgeInvites(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeInvites(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge invites: %w", err)
}
if err := d.MembershipTable.PurgeMemberships(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeMemberships(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge memberships: %w", err)
}
if err := d.RoomAliasesTable.PurgeRoomAliases(ctx, txn, roomID); err != nil {
if err := d.Purge.PurgeRoomAliases(ctx, txn, roomID); err != nil {
return fmt.Errorf("failed to purge room aliases: %w", err)
}
if err := d.PublishedTable.PurgePublished(ctx, txn, roomID); err != nil {
if err := d.Purge.PurgePublished(ctx, txn, roomID); err != nil {
return fmt.Errorf("failed to purge published: %w", err)
}
if err := d.PrevEventsTable.PurgePreviousEvents(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgePreviousEvents(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge previous events: %w", err)
}
if err := d.EventJSONTable.PurgeEventJSONs(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeEventJSONs(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge event JSONs: %w", err)
}
if err := d.RedactionsTable.PurgeRedactions(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeRedactions(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge redactions: %w", err)
}
if err := d.EventsTable.PurgeEvents(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeEvents(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge events: %w", err)
}
if err := d.RoomsTable.PurgeRoom(ctx, txn, roomNID); err != nil {
if err := d.Purge.PurgeRoom(ctx, txn, roomNID); err != nil {
return fmt.Errorf("failed to purge room: %w", err)
}
return nil

7
roomserver/storage/sqlite3/event_json_table.go
View file

@ -18,7 +18,6 @@ package sqlite3
import (
"context"
"database/sql"
"fmt"
"strings"
"github.com/matrix-org/dendrite/internal"
@ -112,9 +111,3 @@ func (s *eventJSONStatements) BulkSelectEventJSON(
}
return results[:i], nil
}
func (s *eventJSONStatements) PurgeEventJSONs(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

6
roomserver/storage/sqlite3/events_table.go
View file

@ -646,9 +646,3 @@ func (s *eventStatements) SelectEventRejected(
err = stmt.QueryRowContext(ctx, roomNID, eventID).Scan(&rejected)
return
}
func (s *eventStatements) PurgeEvents(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

7
roomserver/storage/sqlite3/invite_table.go
View file

@ -18,7 +18,6 @@ package sqlite3
import (
"context"
"database/sql"
"fmt"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -159,9 +158,3 @@ func (s *inviteStatements) SelectInviteActiveForUserInRoom(
}
return result, eventIDs, nil
}
func (s *inviteStatements) PurgeInvites(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

6
roomserver/storage/sqlite3/membership_table.go
View file

@ -407,9 +407,3 @@ func (s *membershipStatements) DeleteMembership(
)
return err
}
func (s *membershipStatements) PurgeMemberships(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

6
roomserver/storage/sqlite3/previous_events_table.go
View file

@ -129,9 +129,3 @@ func (s *previousEventStatements) SelectPreviousEventExists(
stmt := sqlutil.TxStmt(txn, s.selectPreviousEventExistsStmt)
return stmt.QueryRowContext(ctx, eventID, eventReferenceSHA256).Scan(&ok)
}
func (s *previousEventStatements) PurgePreviousEvents(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

7
roomserver/storage/sqlite3/published_table.go
View file

@ -17,7 +17,6 @@ package sqlite3
import (
"context"
"database/sql"
"fmt"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -107,9 +106,3 @@ func (s *publishedStatements) SelectAllPublishedRooms(
}
return roomIDs, rows.Err()
}
func (s *publishedStatements) PurgePublished(
ctx context.Context, txn *sql.Tx, roomID string,
) error {
return fmt.Errorf("not implemented on SQLite")
}

8
roomserver/storage/sqlite3/redactions_table.go
View file

@ -17,11 +17,9 @@ package sqlite3
import (
"context"
"database/sql"
"fmt"
"github.com/matrix-org/dendrite/internal/sqlutil"
"github.com/matrix-org/dendrite/roomserver/storage/tables"
"github.com/matrix-org/dendrite/roomserver/types"
)
const redactionsSchema = `
@ -123,9 +121,3 @@ func (s *redactionStatements) MarkRedactionValidated(
_, err := stmt.ExecContext(ctx, validated, redactionEventID)
return err
}
func (s *redactionStatements) PurgeRedactions(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

7
roomserver/storage/sqlite3/room_aliases_table.go
View file

@ -18,7 +18,6 @@ package sqlite3
import (
"context"
"database/sql"
"fmt"
"github.com/matrix-org/dendrite/internal"
"github.com/matrix-org/dendrite/internal/sqlutil"
@ -144,9 +143,3 @@ func (s *roomAliasesStatements) DeleteRoomAlias(
_, err := stmt.ExecContext(ctx, alias)
return err
}
func (s *roomAliasesStatements) PurgeRoomAliases(
ctx context.Context, txn *sql.Tx, roomID string,
) error {
return fmt.Errorf("not implemented on SQLite")
}

6
roomserver/storage/sqlite3/rooms_table.go
View file

@ -309,9 +309,3 @@ func (s *roomStatements) BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, ro
}
return roomNIDs, nil
}
func (s *roomStatements) PurgeRoom(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

6
roomserver/storage/sqlite3/state_block_table.go
View file

@ -141,9 +141,3 @@ func (s *stateBlockStatements) BulkSelectStateBlockEntries(
}
return results, err
}
func (s *stateBlockStatements) PurgeStateBlocks(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

6
roomserver/storage/sqlite3/state_snapshot_table.go
View file

@ -146,9 +146,3 @@ func (s *stateSnapshotStatements) BulkSelectStateForHistoryVisibility(
) ([]types.EventNID, error) {
return nil, tables.OptimisationNotSupportedError
}
func (s *stateSnapshotStatements) PurgeStateSnapshots(
ctx context.Context, txn *sql.Tx, roomNID types.RoomNID,
) error {
return fmt.Errorf("not implemented on SQLite")
}

23
roomserver/storage/tables/interface.go
View file

@ -22,7 +22,6 @@ type EventJSON interface {
// Insert the event JSON. On conflict, replace the event JSON with the new value (for redactions).
InsertEventJSON(ctx context.Context, tx *sql.Tx, eventNID types.EventNID, eventJSON []byte) error
BulkSelectEventJSON(ctx context.Context, tx *sql.Tx, eventNIDs []types.EventNID) ([]EventJSONPair, error)
PurgeEventJSONs(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type EventTypes interface {
@ -68,7 +67,6 @@ type Events interface {
SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error)
SelectRoomNIDsForEventNIDs(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (roomNIDs map[types.EventNID]types.RoomNID, err error)
SelectEventRejected(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, eventID string) (rejected bool, err error)
PurgeEvents(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type Rooms interface {
@ -82,7 +80,6 @@ type Rooms interface {
SelectRoomIDsWithEvents(ctx context.Context, txn *sql.Tx) ([]string, error)
BulkSelectRoomIDs(ctx context.Context, txn *sql.Tx, roomNIDs []types.RoomNID) ([]string, error)
BulkSelectRoomNIDs(ctx context.Context, txn *sql.Tx, roomIDs []string) ([]types.RoomNID, error)
PurgeRoom(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type StateSnapshot interface {
@ -92,14 +89,12 @@ type StateSnapshot interface {
// which users are in a room faster than having to load the entire room state. In the
// case of SQLite, this will return tables.OptimisationNotSupportedError.
BulkSelectStateForHistoryVisibility(ctx context.Context, txn *sql.Tx, stateSnapshotNID types.StateSnapshotNID, domain string) ([]types.EventNID, error)
PurgeStateSnapshots(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type StateBlock interface {
BulkInsertStateData(ctx context.Context, txn *sql.Tx, entries types.StateEntries) (types.StateBlockNID, error)
BulkSelectStateBlockEntries(ctx context.Context, txn *sql.Tx, stateBlockNIDs types.StateBlockNIDs) ([][]types.EventNID, error)
//BulkSelectFilteredStateBlockEntries(ctx context.Context, stateBlockNIDs []types.StateBlockNID, stateKeyTuples []types.StateKeyTuple) ([]types.StateEntryList, error)
PurgeStateBlocks(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type RoomAliases interface {
@ -108,7 +103,6 @@ type RoomAliases interface {
SelectAliasesFromRoomID(ctx context.Context, txn *sql.Tx, roomID string) ([]string, error)
SelectCreatorIDFromAlias(ctx context.Context, txn *sql.Tx, alias string) (creatorID string, err error)
DeleteRoomAlias(ctx context.Context, txn *sql.Tx, alias string) (err error)
PurgeRoomAliases(ctx context.Context, txn *sql.Tx, roomID string) error
}
type PreviousEvents interface {
@ -116,7 +110,6 @@ type PreviousEvents interface {
// Check if the event reference exists
// Returns sql.ErrNoRows if the event reference doesn't exist.
SelectPreviousEventExists(ctx context.Context, txn *sql.Tx, eventID string, eventReferenceSHA256 []byte) error
PurgePreviousEvents(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type Invites interface {
@ -124,7 +117,6 @@ type Invites interface {
UpdateInviteRetired(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID) ([]string, error)
// SelectInviteActiveForUserInRoom returns a list of sender state key NIDs and invite event IDs matching those nids.
SelectInviteActiveForUserInRoom(ctx context.Context, txn *sql.Tx, targetUserNID types.EventStateKeyNID, roomNID types.RoomNID) ([]types.EventStateKeyNID, []string, error)
PurgeInvites(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type MembershipState int64
@ -151,14 +143,12 @@ type Membership interface {
SelectLocalServerInRoom(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) (bool, error)
SelectServerInRoom(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, serverName gomatrixserverlib.ServerName) (bool, error)
DeleteMembership(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID, targetUserNID types.EventStateKeyNID) error
PurgeMemberships(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
}
type Published interface {
UpsertRoomPublished(ctx context.Context, txn *sql.Tx, roomID string, published bool) (err error)
SelectPublishedFromRoomID(ctx context.Context, txn *sql.Tx, roomID string) (published bool, err error)
SelectAllPublishedRooms(ctx context.Context, txn *sql.Tx, published bool) ([]string, error)
PurgePublished(ctx context.Context, txn *sql.Tx, roomID string) error
}
type RedactionInfo struct {
@ -179,7 +169,20 @@ type Redactions interface {
// Mark this redaction event as having been validated. This means we have both sides of the redaction and have
// successfully redacted the event JSON.
MarkRedactionValidated(ctx context.Context, txn *sql.Tx, redactionEventID string, validated bool) error
}
type Purge interface {
PurgeEventJSONs(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgeEvents(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgeRoom(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgeStateSnapshots(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgeStateBlocks(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgePreviousEvents(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgeInvites(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgeMemberships(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgePublished(ctx context.Context, txn *sql.Tx, roomID string) error
PurgeRedactions(ctx context.Context, txn *sql.Tx, roomNID types.RoomNID) error
PurgeRoomAliases(ctx context.Context, txn *sql.Tx, roomID string) error
}
// StrippedEvent represents a stripped event for returning extracted content values.