Finish implementing redactions (#1189)

* Add a bit more logging to the fedsender

* bugfix: continue sending PDUs if ones are added whilst sending another PDU

Without this, the queue goes back to sleep on `<-oq.notifyPDUs` which won't
fire because `pendingPDUs` is already > 0. This should fix a flakey sytest.

* Break if no txn is sent

* WIP syncapi work

* More debugging

* Bump GMSL version to pull in working Event.Redact

* Remove logging

* Make redactions work on v3+

* Fix more tests

1 files changed, 16 insertions, 0 deletions

diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go
index c7c4dc63..5315de24 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, headeredJSON, event.EventID())
+	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.