diff --git a/currentstateserver/storage/shared/storage.go b/currentstateserver/storage/shared/storage.go index 362dafe96..66b979d88 100644 --- a/currentstateserver/storage/shared/storage.go +++ b/currentstateserver/storage/shared/storage.go @@ -23,7 +23,6 @@ import ( "github.com/matrix-org/dendrite/internal/eventutil" "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/gomatrixserverlib" - "github.com/matrix-org/util" ) type Database struct { @@ -45,10 +44,7 @@ func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, reda return err } if len(events) != 1 { - // this should never happen but is non-fatal - util.GetLogger(ctx).WithField("redacted_event_id", redactedEventID).WithField("redaction_event_id", redactedBecause.EventID()).Warnf( - "RedactEvent: missing redacted event", - ) + // this will happen for all non-state events return nil } redactionEvent := redactedBecause.Unwrap() diff --git a/syncapi/consumers/roomserver.go b/syncapi/consumers/roomserver.go index af7f612b3..e4af623ec 100644 --- a/syncapi/consumers/roomserver.go +++ b/syncapi/consumers/roomserver.go @@ -86,6 +86,8 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { return s.onNewInviteEvent(context.TODO(), *output.NewInviteEvent) case api.OutputTypeRetireInviteEvent: return s.onRetireInviteEvent(context.TODO(), *output.RetireInviteEvent) + case api.OutputTypeRedactedEvent: + return s.onRedactEvent(context.TODO(), *output.RedactedEvent) default: log.WithField("type", output.Type).Debug( "roomserver output log: ignoring unknown output type", @@ -94,6 +96,12 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error { } } +func (c *OutputRoomEventConsumer) onRedactEvent( + ctx context.Context, msg api.OutputRedactedEvent, +) error { + return c.db.RedactEvent(ctx, msg.RedactedEventID, &msg.RedactedBecause) +} + func (s *OutputRoomEventConsumer) onNewRoomEvent( ctx context.Context, msg api.OutputNewRoomEvent, ) error { @@ -173,12 +181,10 @@ func (s *OutputRoomEventConsumer) onRetireInviteEvent( } func (s *OutputRoomEventConsumer) updateStateEvent(event gomatrixserverlib.HeaderedEvent) (gomatrixserverlib.HeaderedEvent, error) { - var stateKey string if event.StateKey() == nil { - stateKey = "" - } else { - stateKey = *event.StateKey() + return event, nil } + stateKey := *event.StateKey() prevEvent, err := s.db.GetStateEvent( context.TODO(), event.RoomID(), event.Type(), stateKey, diff --git a/syncapi/storage/interface.go b/syncapi/storage/interface.go index c4dae4d09..a5e13b674 100644 --- a/syncapi/storage/interface.go +++ b/syncapi/storage/interface.go @@ -136,4 +136,6 @@ type Database interface { // Returns the filterID as a string. Otherwise returns an error if something // goes wrong. PutFilter(ctx context.Context, localpart string, filter *gomatrixserverlib.Filter) (string, error) + // RedactEvent wipes an event in the database and sets the unsigned.redacted_because key to the redaction event + RedactEvent(ctx context.Context, redactedEventID string, redactedBecause *gomatrixserverlib.HeaderedEvent) error } diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go index c7c4dc63b..79257051f 100644 --- a/syncapi/storage/postgres/output_room_events_table.go +++ b/syncapi/storage/postgres/output_room_events_table.go @@ -99,6 +99,9 @@ const selectEarlyEventsSQL = "" + const selectMaxEventIDSQL = "" + "SELECT MAX(id) FROM syncapi_output_room_events" +const updateEventJSONSQL = "" + + "UPDATE syncapi_output_room_events SET headered_event_json=$1 WHERE event_id=$2" + // In order for us to apply the state updates correctly, rows need to be ordered in the order they were received (id). const selectStateInRangeSQL = "" + "SELECT id, headered_event_json, exclude_from_sync, add_state_ids, remove_state_ids" + @@ -120,6 +123,7 @@ type outputRoomEventsStatements struct { selectRecentEventsForSyncStmt *sql.Stmt selectEarlyEventsStmt *sql.Stmt selectStateInRangeStmt *sql.Stmt + updateEventJSONStmt *sql.Stmt } func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) { @@ -149,9 +153,21 @@ func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) { if s.selectStateInRangeStmt, err = db.Prepare(selectStateInRangeSQL); err != nil { return nil, err } + if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil { + return nil, err + } return s, nil } +func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error { + headeredJSON, err := json.Marshal(event) + if err != nil { + return err + } + _, err = s.updateEventJSONStmt.ExecContext(ctx, event.EventID(), headeredJSON) + return err +} + // selectStateInRange returns the state events between the two given PDU stream positions, exclusive of oldPos, inclusive of newPos. // Results are bucketed based on the room ID. If the same state is overwritten multiple times between the // two positions, only the most recent state is returned. diff --git a/syncapi/storage/shared/syncserver.go b/syncapi/storage/shared/syncserver.go index 01362ddd6..7988b199a 100644 --- a/syncapi/storage/shared/syncserver.go +++ b/syncapi/storage/shared/syncserver.go @@ -24,6 +24,7 @@ import ( userapi "github.com/matrix-org/dendrite/userapi/api" "github.com/matrix-org/dendrite/eduserver/cache" + "github.com/matrix-org/dendrite/internal/eventutil" "github.com/matrix-org/dendrite/internal/sqlutil" "github.com/matrix-org/dendrite/roomserver/api" "github.com/matrix-org/dendrite/syncapi/storage/tables" @@ -597,6 +598,26 @@ func (d *Database) IncrementalSync( return res, nil } +func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, redactedBecause *gomatrixserverlib.HeaderedEvent) error { + redactedEvents, err := d.Events(ctx, []string{redactedEventID}) + if err != nil { + return err + } + if len(redactedEvents) == 0 { + logrus.WithField("event_id", redactedEventID).WithField("redaction_event", redactedBecause.EventID()).Warnf("missing redacted event for redaction") + return nil + } + eventToRedact := redactedEvents[0].Unwrap() + redactionEvent := redactedBecause.Unwrap() + ev, err := eventutil.RedactEvent(&redactionEvent, &eventToRedact) + if err != nil { + return err + } + + newEvent := ev.Headered(redactedBecause.RoomVersion) + return d.OutputEvents.UpdateEventJSON(ctx, &newEvent) +} + // getResponseWithPDUsForCompleteSync creates a response and adds all PDUs needed // to it. It returns toPos and joinedRoomIDs for use of adding EDUs. // nolint:nakedret @@ -619,6 +640,7 @@ func (d *Database) getResponseWithPDUsForCompleteSync( } var succeeded bool defer func() { + logrus.Infof("getResponseWithPDUsForCompleteSync: limit:%d rooms:%v", numRecentEventsPerRoom, joinedRoomIDs) txerr := sqlutil.EndTransaction(txn, &succeeded) if err == nil && txerr != nil { err = txerr @@ -790,6 +812,7 @@ func (d *Database) addRoomDeltaToResponse( // This is all "okay" assuming history_visibility == "shared" which it is by default. r.To = delta.membershipPos } + logrus.Infof("addRoomDeltaToResponse range:%+v roomID:%s limit:%d", r, delta.roomID, numRecentEventsPerRoom) recentStreamEvents, limited, err := d.OutputEvents.SelectRecentEvents( ctx, txn, delta.roomID, r, numRecentEventsPerRoom, true, true, @@ -797,6 +820,7 @@ func (d *Database) addRoomDeltaToResponse( if err != nil { return err } + logrus.Infof("addRoomDeltaToResponse produced %d events, limited:%v", len(recentStreamEvents), limited) recentEvents := d.StreamEventsToEvents(device, recentStreamEvents) delta.stateEvents = removeDuplicates(delta.stateEvents, recentEvents) // roll back prevBatch, err := d.getBackwardTopologyPos(ctx, txn, recentStreamEvents) diff --git a/syncapi/storage/sqlite3/output_room_events_table.go b/syncapi/storage/sqlite3/output_room_events_table.go index 0c909cc4d..73527879a 100644 --- a/syncapi/storage/sqlite3/output_room_events_table.go +++ b/syncapi/storage/sqlite3/output_room_events_table.go @@ -76,6 +76,9 @@ const selectEarlyEventsSQL = "" + const selectMaxEventIDSQL = "" + "SELECT MAX(id) FROM syncapi_output_room_events" +const updateEventJSONSQL = "" + + "UPDATE syncapi_output_room_events SET headered_event_json=$1 WHERE event_id=$2" + // In order for us to apply the state updates correctly, rows need to be ordered in the order they were received (id). /* $1 = oldPos, @@ -109,6 +112,7 @@ type outputRoomEventsStatements struct { selectRecentEventsForSyncStmt *sql.Stmt selectEarlyEventsStmt *sql.Stmt selectStateInRangeStmt *sql.Stmt + updateEventJSONStmt *sql.Stmt } func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Events, error) { @@ -140,9 +144,21 @@ func NewSqliteEventsTable(db *sql.DB, streamID *streamIDStatements) (tables.Even if s.selectStateInRangeStmt, err = db.Prepare(selectStateInRangeSQL); err != nil { return nil, err } + if s.updateEventJSONStmt, err = db.Prepare(updateEventJSONSQL); err != nil { + return nil, err + } return s, nil } +func (s *outputRoomEventsStatements) UpdateEventJSON(ctx context.Context, event *gomatrixserverlib.HeaderedEvent) error { + headeredJSON, err := json.Marshal(event) + if err != nil { + return err + } + _, err = s.updateEventJSONStmt.ExecContext(ctx, event.EventID(), headeredJSON) + return err +} + // selectStateInRange returns the state events between the two given PDU stream positions, exclusive of oldPos, inclusive of newPos. // Results are bucketed based on the room ID. If the same state is overwritten multiple times between the // two positions, only the most recent state is returned. diff --git a/syncapi/storage/tables/interface.go b/syncapi/storage/tables/interface.go index 4ac0be4ec..9d239d233 100644 --- a/syncapi/storage/tables/interface.go +++ b/syncapi/storage/tables/interface.go @@ -49,6 +49,7 @@ type Events interface { // SelectEarlyEvents returns the earliest events in the given room. 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 } // Topology keeps track of the depths and stream positions for all events.