Emit redacted_event from the roomserver when redactions are validated (#1186)

* Emit redacted_event from the roomserver when redactions are validated

- Consume them in the currentstateserver and act accordingly.
- Add integration test for the roomserver to check that injecting
  `m.room.redaction` events result in `redacted_event` being emitted.

* Linting

* Ignore events that redact themselves

currentstateserver/consumers/roomserver.go

@@ -61,6 +61,8 @@ func (c *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
 		return c.onNewRoomEvent(context.TODO(), *output.NewRoomEvent)
 	case api.OutputTypeNewInviteEvent:
 	case api.OutputTypeRetireInviteEvent:
+	case api.OutputTypeRedactedEvent:
+		return c.onRedactEvent(context.Background(), *output.RedactedEvent)
 	default:
 		log.WithField("type", output.Type).Debug(
 			"roomserver output log: ignoring unknown output type",
@@ -105,6 +107,12 @@ func (c *OutputRoomEventConsumer) onNewRoomEvent(
 	return nil
 }
 
+func (c *OutputRoomEventConsumer) onRedactEvent(
+	ctx context.Context, msg api.OutputRedactedEvent,
+) error {
+	return c.db.RedactEvent(ctx, msg.RedactedEventID, msg.RedactedBecause)
+}
+
 // Start consuming from room servers
 func (c *OutputRoomEventConsumer) Start() error {
 	return c.rsConsumer.Start()

currentstateserver/storage/interface.go

@@ -35,4 +35,6 @@ type Database interface {
 	// GetBulkStateContent returns all state events which match a given room ID and a given state key tuple. Both must be satisfied for a match.
 	// If a tuple has the StateKey of '*' and allowWildcards=true then all state events with the EventType should be returned.
 	GetBulkStateContent(ctx context.Context, roomIDs []string, tuples []gomatrixserverlib.StateKeyTuple, allowWildcards bool) ([]tables.StrippedEvent, error)
+	// Redact a state event
+	RedactEvent(ctx context.Context, redactedEventID string, redactedBecause gomatrixserverlib.HeaderedEvent) error
 }

currentstateserver/storage/postgres/current_room_state_table.go

@@ -189,7 +189,6 @@ func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
 		if err := rows.Scan(&eventBytes); err != nil {
 			return nil, err
 		}
-		// TODO: Handle redacted events
 		var ev gomatrixserverlib.HeaderedEvent
 		if err := json.Unmarshal(eventBytes, &ev); err != nil {
 			return nil, err

currentstateserver/storage/shared/storage.go

@@ -17,10 +17,13 @@ package shared
 import (
 	"context"
 	"database/sql"
+	"fmt"
 
 	"github.com/matrix-org/dendrite/currentstateserver/storage/tables"
+	"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 {
@@ -36,6 +39,28 @@ func (d *Database) GetBulkStateContent(ctx context.Context, roomIDs []string, tu
 	return d.CurrentRoomState.SelectBulkStateContent(ctx, roomIDs, tuples, allowWildcards)
 }
 
+func (d *Database) RedactEvent(ctx context.Context, redactedEventID string, redactedBecause gomatrixserverlib.HeaderedEvent) error {
+	events, err := d.CurrentRoomState.SelectEventsWithEventIDs(ctx, nil, []string{redactedEventID})
+	if err != nil {
+		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",
+		)
+		return nil
+	}
+	redactionEvent := redactedBecause.Unwrap()
+	eventBeingRedacted := events[0].Unwrap()
+	redactedEvent, err := eventutil.RedactEvent(&redactionEvent, &eventBeingRedacted)
+	if err != nil {
+		return fmt.Errorf("RedactEvent failed: %w", err)
+	}
+	// replace the state event with a redacted version of itself
+	return d.StoreStateEvents(ctx, []gomatrixserverlib.HeaderedEvent{redactedEvent.Headered(redactedBecause.RoomVersion)}, []string{redactedEventID})
+}
+
 func (d *Database) StoreStateEvents(ctx context.Context, addStateEvents []gomatrixserverlib.HeaderedEvent,
 	removeStateEventIDs []string) error {
 	return sqlutil.WithTransaction(d.DB, func(txn *sql.Tx) error {

currentstateserver/storage/sqlite3/current_room_state_table.go

@@ -162,7 +162,13 @@ func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
 		iEventIDs[k] = v
 	}
 	query := strings.Replace(selectEventsWithEventIDsSQL, "($1)", sqlutil.QueryVariadic(len(iEventIDs)), 1)
-	rows, err := txn.QueryContext(ctx, query, iEventIDs...)
+	var rows *sql.Rows
+	var err error
+	if txn != nil {
+		rows, err = txn.QueryContext(ctx, query, iEventIDs...)
+	} else {
+		rows, err = s.db.QueryContext(ctx, query, iEventIDs...)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -173,7 +179,6 @@ func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
 		if err := rows.Scan(&eventBytes); err != nil {
 			return nil, err
 		}
-		// TODO: Handle redacted events
 		var ev gomatrixserverlib.HeaderedEvent
 		if err := json.Unmarshal(eventBytes, &ev); err != nil {
 			return nil, err

currentstateserver/storage/tables/interface.go

@@ -24,6 +24,8 @@ import (
 
 type CurrentRoomState interface {
 	SelectStateEvent(ctx context.Context, roomID, evType, stateKey string) (*gomatrixserverlib.HeaderedEvent, error)
+	// SelectEventsWithEventIDs returns the events for the given event IDs. If the event(s) are missing, they are not returned
+	// and no error is returned.
 	SelectEventsWithEventIDs(ctx context.Context, txn *sql.Tx, eventIDs []string) ([]gomatrixserverlib.HeaderedEvent, error)
 	// UpsertRoomState stores the given event in the database, along with an extracted piece of content.
 	// The piece of content will vary depending on the event type, and table implementations may use this information to optimise