HeaderedEvents in sync API (#922)

* Use HeaderedEvent in syncapi

* Update notifier test

* Fix persisting headered event

* Clean up unused API function

* Fix overshadowed err from linter

* Write headered JSON to invites table too

* Rename event_json to headered_event_json in syncapi database schemae

* Fix invites_table queries

* Update QueryRoomVersionCapabilitiesResponse comment

* Fix syncapi SQLite

5 files changed, 86 insertions, 64 deletions

diff --git a/syncapi/storage/postgres/current_room_state_table.go b/syncapi/storage/postgres/current_room_state_table.go
index 48cc2b5f..ab8f07b2 100644
--- a/syncapi/storage/postgres/current_room_state_table.go
+++ b/syncapi/storage/postgres/current_room_state_table.go
@@ -42,7 +42,7 @@ CREATE TABLE IF NOT EXISTS syncapi_current_room_state (
     -- The state_key value for this state event e.g ''
     state_key TEXT NOT NULL,
     -- The JSON for the event. Stored as TEXT because this should be valid UTF-8.
-    event_json TEXT NOT NULL,
+    headered_event_json TEXT NOT NULL,
     -- The 'content.membership' value if this event is an m.room.member event. For other
     -- events, this will be NULL.
     membership TEXT,
@@ -59,10 +59,10 @@ CREATE INDEX IF NOT EXISTS syncapi_membership_idx ON syncapi_current_room_state(
 `
 
 const upsertRoomStateSQL = "" +
-	"INSERT INTO syncapi_current_room_state (room_id, event_id, type, sender, contains_url, state_key, event_json, membership, added_at)" +
+	"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 ON CONSTRAINT syncapi_room_state_unique" +
-	" DO UPDATE SET event_id = $2, sender=$4, contains_url=$5, event_json = $7, membership = $8, added_at = $9"
+	" 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"
@@ -71,7 +71,7 @@ const selectRoomIDsWithMembershipSQL = "" +
 	"SELECT room_id FROM syncapi_current_room_state WHERE type = 'm.room.member' AND state_key = $1 AND membership = $2"
 
 const selectCurrentStateSQL = "" +
-	"SELECT event_json FROM syncapi_current_room_state WHERE room_id = $1" +
+	"SELECT headered_event_json FROM syncapi_current_room_state WHERE room_id = $1" +
 	" AND ( $2::text[] IS NULL OR     sender  = ANY($2)  )" +
 	" AND ( $3::text[] IS NULL OR NOT(sender  = ANY($3)) )" +
 	" AND ( $4::text[] IS NULL OR     type LIKE ANY($4)  )" +
@@ -83,14 +83,14 @@ const selectJoinedUsersSQL = "" +
 	"SELECT room_id, state_key FROM syncapi_current_room_state WHERE type = 'm.room.member' AND membership = 'join'"
 
 const selectStateEventSQL = "" +
-	"SELECT event_json FROM syncapi_current_room_state WHERE room_id = $1 AND type = $2 AND state_key = $3"
+	"SELECT headered_event_json FROM syncapi_current_room_state WHERE room_id = $1 AND type = $2 AND state_key = $3"
 
 const selectEventsWithEventIDsSQL = "" +
 	// TODO: The session_id and transaction_id blanks are here because otherwise
 	// the rowsToStreamEvents expects there to be exactly five columns. We need to
 	// figure out if these really need to be in the DB, and if so, we need a
 	// better permanent fix for this. - neilalexander, 2 Jan 2020
-	"SELECT added_at, event_json, 0 AS session_id, false AS exclude_from_sync, '' AS transaction_id" +
+	"SELECT added_at, headered_event_json, 0 AS session_id, false AS exclude_from_sync, '' AS transaction_id" +
 	" FROM syncapi_current_room_state WHERE event_id = ANY($1)"
 
 type currentRoomStateStatements struct {
@@ -185,7 +185,7 @@ func (s *currentRoomStateStatements) selectRoomIDsWithMembership(
 func (s *currentRoomStateStatements) selectCurrentState(
 	ctx context.Context, txn *sql.Tx, roomID string,
 	stateFilter *gomatrixserverlib.StateFilter,
-) ([]gomatrixserverlib.Event, error) {
+) ([]gomatrixserverlib.HeaderedEvent, error) {
 	stmt := common.TxStmt(txn, s.selectCurrentStateStmt)
 	rows, err := stmt.QueryContext(ctx, roomID,
 		pq.StringArray(stateFilter.Senders),
@@ -213,7 +213,7 @@ func (s *currentRoomStateStatements) deleteRoomStateByEventID(
 
 func (s *currentRoomStateStatements) upsertRoomState(
 	ctx context.Context, txn *sql.Tx,
-	event gomatrixserverlib.Event, membership *string, addedAt types.StreamPosition,
+	event gomatrixserverlib.HeaderedEvent, membership *string, addedAt types.StreamPosition,
 ) error {
 	// Parse content as JSON and search for an "url" key
 	containsURL := false
@@ -223,9 +223,14 @@ func (s *currentRoomStateStatements) upsertRoomState(
 		_, containsURL = content["url"]
 	}
 
+	headeredJSON, err := json.Marshal(event)
+	if err != nil {
+		return err
+	}
+
 	// upsert state event
 	stmt := common.TxStmt(txn, s.upsertRoomStateStmt)
-	_, err := stmt.ExecContext(
+	_, err = stmt.ExecContext(
 		ctx,
 		event.RoomID(),
 		event.EventID(),
@@ -233,7 +238,7 @@ func (s *currentRoomStateStatements) upsertRoomState(
 		event.Sender(),
 		containsURL,
 		*event.StateKey(),
-		event.JSON(),
+		headeredJSON,
 		membership,
 		addedAt,
 	)
@@ -252,16 +257,16 @@ func (s *currentRoomStateStatements) selectEventsWithEventIDs(
 	return rowsToStreamEvents(rows)
 }
 
-func rowsToEvents(rows *sql.Rows) ([]gomatrixserverlib.Event, error) {
-	result := []gomatrixserverlib.Event{}
+func rowsToEvents(rows *sql.Rows) ([]gomatrixserverlib.HeaderedEvent, error) {
+	result := []gomatrixserverlib.HeaderedEvent{}
 	for rows.Next() {
 		var eventBytes []byte
 		if err := rows.Scan(&eventBytes); err != nil {
 			return nil, err
 		}
 		// TODO: Handle redacted events
-		ev, err := gomatrixserverlib.NewEventFromTrustedJSON(eventBytes, false)
-		if err != nil {
+		var ev gomatrixserverlib.HeaderedEvent
+		if err := json.Unmarshal(eventBytes, &ev); err != nil {
 			return nil, err
 		}
 		result = append(result, ev)
@@ -271,7 +276,7 @@ func rowsToEvents(rows *sql.Rows) ([]gomatrixserverlib.Event, error) {
 
 func (s *currentRoomStateStatements) selectStateEvent(
 	ctx context.Context, roomID, evType, stateKey string,
-) (*gomatrixserverlib.Event, error) {
+) (*gomatrixserverlib.HeaderedEvent, error) {
 	stmt := s.selectStateEventStmt
 	var res []byte
 	err := stmt.QueryRowContext(ctx, roomID, evType, stateKey).Scan(&res)
@@ -281,6 +286,9 @@ func (s *currentRoomStateStatements) selectStateEvent(
 	if err != nil {
 		return nil, err
 	}
-	ev, err := gomatrixserverlib.NewEventFromTrustedJSON(res, false)
+	var ev gomatrixserverlib.HeaderedEvent
+	if err = json.Unmarshal(res, &ev); err != nil {
+		return nil, err
+	}
 	return &ev, err
 }
diff --git a/syncapi/storage/postgres/invites_table.go b/syncapi/storage/postgres/invites_table.go
index 6a7f9ef2..ca0c64fb 100644
--- a/syncapi/storage/postgres/invites_table.go
+++ b/syncapi/storage/postgres/invites_table.go
@@ -18,6 +18,7 @@ package postgres
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 
 	"github.com/matrix-org/dendrite/common"
 	"github.com/matrix-org/dendrite/syncapi/types"
@@ -30,7 +31,7 @@ CREATE TABLE IF NOT EXISTS syncapi_invite_events (
 	event_id TEXT NOT NULL,
 	room_id TEXT NOT NULL,
 	target_user_id TEXT NOT NULL,
-	event_json TEXT NOT NULL
+	headered_event_json TEXT NOT NULL
 );
 
 -- For looking up the invites for a given user.
@@ -44,14 +45,14 @@ CREATE INDEX IF NOT EXISTS syncapi_invites_event_id_idx
 
 const insertInviteEventSQL = "" +
 	"INSERT INTO syncapi_invite_events (" +
-	" room_id, event_id, target_user_id, event_json" +
+	" room_id, event_id, target_user_id, headered_event_json" +
 	") VALUES ($1, $2, $3, $4) RETURNING id"
 
 const deleteInviteEventSQL = "" +
 	"DELETE FROM syncapi_invite_events WHERE event_id = $1"
 
 const selectInviteEventsInRangeSQL = "" +
-	"SELECT room_id, event_json FROM syncapi_invite_events" +
+	"SELECT room_id, headered_event_json FROM syncapi_invite_events" +
 	" WHERE target_user_id = $1 AND id > $2 AND id <= $3" +
 	" ORDER BY id DESC"
 
@@ -86,14 +87,20 @@ func (s *inviteEventsStatements) prepare(db *sql.DB) (err error) {
 }
 
 func (s *inviteEventsStatements) insertInviteEvent(
-	ctx context.Context, inviteEvent gomatrixserverlib.Event,
+	ctx context.Context, inviteEvent gomatrixserverlib.HeaderedEvent,
 ) (streamPos types.StreamPosition, err error) {
+	var headeredJSON []byte
+	headeredJSON, err = json.Marshal(inviteEvent)
+	if err != nil {
+		return
+	}
+
 	err = s.insertInviteEventStmt.QueryRowContext(
 		ctx,
 		inviteEvent.RoomID(),
 		inviteEvent.EventID(),
 		*inviteEvent.StateKey(),
-		inviteEvent.JSON(),
+		headeredJSON,
 	).Scan(&streamPos)
 	return
 }
@@ -109,14 +116,14 @@ func (s *inviteEventsStatements) deleteInviteEvent(
 // active invites for the target user ID in the supplied range.
 func (s *inviteEventsStatements) selectInviteEventsInRange(
 	ctx context.Context, txn *sql.Tx, targetUserID string, startPos, endPos types.StreamPosition,
-) (map[string]gomatrixserverlib.Event, error) {
+) (map[string]gomatrixserverlib.HeaderedEvent, error) {
 	stmt := common.TxStmt(txn, s.selectInviteEventsInRangeStmt)
 	rows, err := stmt.QueryContext(ctx, targetUserID, startPos, endPos)
 	if err != nil {
 		return nil, err
 	}
 	defer common.CloseAndLogIfError(ctx, rows, "selectInviteEventsInRange: rows.close() failed")
-	result := map[string]gomatrixserverlib.Event{}
+	result := map[string]gomatrixserverlib.HeaderedEvent{}
 	for rows.Next() {
 		var (
 			roomID    string
@@ -126,8 +133,8 @@ func (s *inviteEventsStatements) selectInviteEventsInRange(
 			return nil, err
 		}
 
-		event, err := gomatrixserverlib.NewEventFromTrustedJSON(eventJSON, false)
-		if err != nil {
+		var event gomatrixserverlib.HeaderedEvent
+		if err := json.Unmarshal(eventJSON, &event); err != nil {
 			return nil, err
 		}
 
diff --git a/syncapi/storage/postgres/output_room_events_table.go b/syncapi/storage/postgres/output_room_events_table.go
index e7cf1935..5f9a1d0c 100644
--- a/syncapi/storage/postgres/output_room_events_table.go
+++ b/syncapi/storage/postgres/output_room_events_table.go
@@ -44,8 +44,9 @@ CREATE TABLE IF NOT EXISTS syncapi_output_room_events (
   event_id TEXT NOT NULL CONSTRAINT syncapi_event_id_idx UNIQUE,
   -- The 'room_id' key for the event.
   room_id TEXT NOT NULL,
-  -- The JSON for the event. Stored as TEXT because this should be valid UTF-8.
-  event_json TEXT NOT NULL,
+  -- The headered JSON for the event, containing potentially additional metadata such as
+  -- the room version. Stored as TEXT because this should be valid UTF-8.
+  headered_event_json TEXT NOT NULL,
   -- The event type e.g 'm.room.member'.
   type TEXT NOT NULL,
   -- The 'sender' property of the event.
@@ -70,26 +71,26 @@ CREATE TABLE IF NOT EXISTS syncapi_output_room_events (
 
 const insertEventSQL = "" +
 	"INSERT INTO syncapi_output_room_events (" +
-	"room_id, event_id, event_json, type, sender, contains_url, add_state_ids, remove_state_ids, session_id, transaction_id, exclude_from_sync" +
+	"room_id, event_id, headered_event_json, type, sender, contains_url, add_state_ids, remove_state_ids, session_id, transaction_id, exclude_from_sync" +
 	") VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) " +
 	"ON CONFLICT ON CONSTRAINT syncapi_event_id_idx DO UPDATE SET exclude_from_sync = $11 " +
 	"RETURNING id"
 
 const selectEventsSQL = "" +
-	"SELECT id, event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events WHERE event_id = ANY($1)"
+	"SELECT id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events WHERE event_id = ANY($1)"
 
 const selectRecentEventsSQL = "" +
-	"SELECT id, event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
+	"SELECT id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
 	" WHERE room_id = $1 AND id > $2 AND id <= $3" +
 	" ORDER BY id DESC LIMIT $4"
 
 const selectRecentEventsForSyncSQL = "" +
-	"SELECT id, event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
+	"SELECT id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
 	" WHERE room_id = $1 AND id > $2 AND id <= $3 AND exclude_from_sync = FALSE" +
 	" ORDER BY id DESC LIMIT $4"
 
 const selectEarlyEventsSQL = "" +
-	"SELECT id, event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
+	"SELECT id, headered_event_json, session_id, exclude_from_sync, transaction_id FROM syncapi_output_room_events" +
 	" WHERE room_id = $1 AND id > $2 AND id <= $3" +
 	" ORDER BY id ASC LIMIT $4"
 
@@ -98,7 +99,7 @@ const selectMaxEventIDSQL = "" +
 
 // 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, event_json, exclude_from_sync, add_state_ids, remove_state_ids" +
+	"SELECT id, headered_event_json, exclude_from_sync, add_state_ids, remove_state_ids" +
 	" FROM syncapi_output_room_events" +
 	" WHERE (id > $1 AND id <= $2) AND (add_state_ids IS NOT NULL OR remove_state_ids IS NOT NULL)" +
 	" AND ( $3::text[] IS NULL OR     sender  = ANY($3)  )" +
@@ -203,8 +204,8 @@ func (s *outputRoomEventsStatements) selectStateInRange(
 		}
 
 		// TODO: Handle redacted events
-		ev, err := gomatrixserverlib.NewEventFromTrustedJSON(eventBytes, false)
-		if err != nil {
+		var ev gomatrixserverlib.HeaderedEvent
+		if err := json.Unmarshal(eventBytes, &ev); err != nil {
 			return nil, nil, err
 		}
 		needSet := stateNeeded[ev.RoomID()]
@@ -220,7 +221,7 @@ func (s *outputRoomEventsStatements) selectStateInRange(
 		stateNeeded[ev.RoomID()] = needSet
 
 		eventIDToEvent[ev.EventID()] = types.StreamEvent{
-			Event:           ev,
+			HeaderedEvent:   ev,
 			StreamPosition:  streamPos,
 			ExcludeFromSync: excludeFromSync,
 		}
@@ -248,7 +249,7 @@ func (s *outputRoomEventsStatements) selectMaxEventID(
 // of the inserted event.
 func (s *outputRoomEventsStatements) insertEvent(
 	ctx context.Context, txn *sql.Tx,
-	event *gomatrixserverlib.Event, addState, removeState []string,
+	event *gomatrixserverlib.HeaderedEvent, addState, removeState []string,
 	transactionID *api.TransactionID, excludeFromSync bool,
 ) (streamPos types.StreamPosition, err error) {
 	var txnID *string
@@ -266,12 +267,18 @@ func (s *outputRoomEventsStatements) insertEvent(
 		_, containsURL = content["url"]
 	}
 
+	var headeredJSON []byte
+	headeredJSON, err = json.Marshal(event)
+	if err != nil {
+		return
+	}
+
 	stmt := common.TxStmt(txn, s.insertEventStmt)
 	err = stmt.QueryRowContext(
 		ctx,
 		event.RoomID(),
 		event.EventID(),
-		event.JSON(),
+		headeredJSON,
 		event.Type(),
 		event.Sender(),
 		containsURL,
@@ -373,8 +380,8 @@ func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
 			return nil, err
 		}
 		// TODO: Handle redacted events
-		ev, err := gomatrixserverlib.NewEventFromTrustedJSON(eventBytes, false)
-		if err != nil {
+		var ev gomatrixserverlib.HeaderedEvent
+		if err := json.Unmarshal(eventBytes, &ev); err != nil {
 			return nil, err
 		}
 
@@ -386,7 +393,7 @@ func rowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
 		}
 
 		result = append(result, types.StreamEvent{
-			Event:           ev,
+			HeaderedEvent:   ev,
 			StreamPosition:  streamPos,
 			TransactionID:   transactionID,
 			ExcludeFromSync: excludeFromSync,
diff --git a/syncapi/storage/postgres/output_room_events_topology_table.go b/syncapi/storage/postgres/output_room_events_topology_table.go
index b114fd43..280d4ec3 100644
--- a/syncapi/storage/postgres/output_room_events_topology_table.go
+++ b/syncapi/storage/postgres/output_room_events_topology_table.go
@@ -104,7 +104,7 @@ func (s *outputRoomEventsTopologyStatements) prepare(db *sql.DB) (err error) {
 // insertEventInTopology inserts the given event in the room's topology, based
 // on the event's depth.
 func (s *outputRoomEventsTopologyStatements) insertEventInTopology(
-	ctx context.Context, event *gomatrixserverlib.Event,
+	ctx context.Context, event *gomatrixserverlib.HeaderedEvent,
 ) (err error) {
 	_, err = s.insertEventInTopologyStmt.ExecContext(
 		ctx, event.EventID(), event.Depth(), event.RoomID(),
diff --git a/syncapi/storage/postgres/syncserver.go b/syncapi/storage/postgres/syncserver.go
index abb5b4a4..ccf1c565 100644
--- a/syncapi/storage/postgres/syncserver.go
+++ b/syncapi/storage/postgres/syncserver.go
@@ -37,7 +37,7 @@ import (
 
 type stateDelta struct {
 	roomID      string
-	stateEvents []gomatrixserverlib.Event
+	stateEvents []gomatrixserverlib.HeaderedEvent
 	membership  string
 	// The PDU stream position of the latest membership event for this user, if applicable.
 	// Can be 0 if there is no membership event in this delta.
@@ -100,7 +100,7 @@ func (d *SyncServerDatasource) AllJoinedUsersInRooms(ctx context.Context) (map[s
 // If an event is not found in the database then it will be omitted from the list.
 // Returns an error if there was a problem talking with the database.
 // Does not include any transaction IDs in the returned events.
-func (d *SyncServerDatasource) Events(ctx context.Context, eventIDs []string) ([]gomatrixserverlib.Event, error) {
+func (d *SyncServerDatasource) Events(ctx context.Context, eventIDs []string) ([]gomatrixserverlib.HeaderedEvent, error) {
 	streamEvents, err := d.events.selectEvents(ctx, nil, eventIDs)
 	if err != nil {
 		return nil, err
@@ -111,7 +111,7 @@ func (d *SyncServerDatasource) Events(ctx context.Context, eventIDs []string) ([
 	return d.StreamEventsToEvents(nil, streamEvents), nil
 }
 
-func (d *SyncServerDatasource) handleBackwardExtremities(ctx context.Context, ev *gomatrixserverlib.Event) error {
+func (d *SyncServerDatasource) handleBackwardExtremities(ctx context.Context, ev *gomatrixserverlib.HeaderedEvent) error {
 	// If the event is already known as a backward extremity, don't consider
 	// it as such anymore now that we have it.
 	isBackwardExtremity, err := d.backwardExtremities.isBackwardExtremity(ctx, ev.RoomID(), ev.EventID())
@@ -155,8 +155,8 @@ func (d *SyncServerDatasource) handleBackwardExtremities(ctx context.Context, ev
 // Returns an error if there was a problem inserting this event.
 func (d *SyncServerDatasource) WriteEvent(
 	ctx context.Context,
-	ev *gomatrixserverlib.Event,
-	addStateEvents []gomatrixserverlib.Event,
+	ev *gomatrixserverlib.HeaderedEvent,
+	addStateEvents []gomatrixserverlib.HeaderedEvent,
 	addStateEventIDs, removeStateEventIDs []string,
 	transactionID *api.TransactionID, excludeFromSync bool,
 ) (pduPosition types.StreamPosition, returnErr error) {
@@ -192,7 +192,7 @@ func (d *SyncServerDatasource) WriteEvent(
 func (d *SyncServerDatasource) updateRoomState(
 	ctx context.Context, txn *sql.Tx,
 	removedEventIDs []string,
-	addedEvents []gomatrixserverlib.Event,
+	addedEvents []gomatrixserverlib.HeaderedEvent,
 	pduPosition types.StreamPosition,
 ) error {
 	// remove first, then add, as we do not ever delete state, but do replace state which is a remove followed by an add.
@@ -228,7 +228,7 @@ func (d *SyncServerDatasource) updateRoomState(
 // If there was an issue during the retrieval, returns an error
 func (d *SyncServerDatasource) GetStateEvent(
 	ctx context.Context, roomID, evType, stateKey string,
-) (*gomatrixserverlib.Event, error) {
+) (*gomatrixserverlib.HeaderedEvent, error) {
 	return d.roomstate.selectStateEvent(ctx, roomID, evType, stateKey)
 }
 
@@ -237,7 +237,7 @@ func (d *SyncServerDatasource) GetStateEvent(
 // Returns an error if there was an issue with the retrieval.
 func (d *SyncServerDatasource) GetStateEventsForRoom(
 	ctx context.Context, roomID string, stateFilter *gomatrixserverlib.StateFilter,
-) (stateEvents []gomatrixserverlib.Event, err error) {
+) (stateEvents []gomatrixserverlib.HeaderedEvent, err error) {
 	err = common.WithTransaction(d.db, func(txn *sql.Tx) error {
 		stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID, stateFilter)
 		return err
@@ -599,7 +599,7 @@ func (d *SyncServerDatasource) getResponseWithPDUsForCompleteSync(
 
 	// Build up a /sync response. Add joined rooms.
 	for _, roomID := range joinedRoomIDs {
-		var stateEvents []gomatrixserverlib.Event
+		var stateEvents []gomatrixserverlib.HeaderedEvent
 		stateEvents, err = d.roomstate.selectCurrentState(ctx, txn, roomID, &stateFilter)
 		if err != nil {
 			return
@@ -633,9 +633,9 @@ func (d *SyncServerDatasource) getResponseWithPDUsForCompleteSync(
 		jr.Timeline.PrevBatch = types.NewPaginationTokenFromTypeAndPosition(
 			types.PaginationTokenTypeTopology, backwardTopologyPos, 0,
 		).String()
-		jr.Timeline.Events = gomatrixserverlib.ToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
+		jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
 		jr.Timeline.Limited = true
-		jr.State.Events = gomatrixserverlib.ToClientEvents(stateEvents, gomatrixserverlib.FormatSync)
+		jr.State.Events = gomatrixserverlib.HeaderedToClientEvents(stateEvents, gomatrixserverlib.FormatSync)
 		res.Rooms.Join[roomID] = *jr
 	}
 
@@ -707,7 +707,7 @@ func (d *SyncServerDatasource) UpsertAccountData(
 // If the invite was successfully stored this returns the stream ID it was stored at.
 // Returns an error if there was a problem communicating with the database.
 func (d *SyncServerDatasource) AddInviteEvent(
-	ctx context.Context, inviteEvent gomatrixserverlib.Event,
+	ctx context.Context, inviteEvent gomatrixserverlib.HeaderedEvent,
 ) (types.StreamPosition, error) {
 	return d.invites.insertInviteEvent(ctx, inviteEvent)
 }
@@ -758,7 +758,7 @@ func (d *SyncServerDatasource) addInvitesToResponse(
 	for roomID, inviteEvent := range invites {
 		ir := types.NewInviteResponse()
 		ir.InviteState.Events = gomatrixserverlib.ToClientEvents(
-			[]gomatrixserverlib.Event{inviteEvent}, gomatrixserverlib.FormatSync,
+			[]gomatrixserverlib.Event{inviteEvent.Event}, gomatrixserverlib.FormatSync,
 		)
 		// TODO: add the invite state from the invite event.
 		res.Rooms.Invite[roomID] = *ir
@@ -821,9 +821,9 @@ func (d *SyncServerDatasource) addRoomDeltaToResponse(
 		jr.Timeline.PrevBatch = types.NewPaginationTokenFromTypeAndPosition(
 			types.PaginationTokenTypeTopology, backwardTopologyPos, 0,
 		).String()
-		jr.Timeline.Events = gomatrixserverlib.ToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
+		jr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
 		jr.Timeline.Limited = false // TODO: if len(events) >= numRecents + 1 and then set limited:true
-		jr.State.Events = gomatrixserverlib.ToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync)
+		jr.State.Events = gomatrixserverlib.HeaderedToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync)
 		res.Rooms.Join[delta.roomID] = *jr
 	case gomatrixserverlib.Leave:
 		fallthrough // transitions to leave are the same as ban
@@ -834,9 +834,9 @@ func (d *SyncServerDatasource) addRoomDeltaToResponse(
 		lr.Timeline.PrevBatch = types.NewPaginationTokenFromTypeAndPosition(
 			types.PaginationTokenTypeTopology, backwardTopologyPos, 0,
 		).String()
-		lr.Timeline.Events = gomatrixserverlib.ToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
+		lr.Timeline.Events = gomatrixserverlib.HeaderedToClientEvents(recentEvents, gomatrixserverlib.FormatSync)
 		lr.Timeline.Limited = false // TODO: if len(events) >= numRecents + 1 and then set limited:true
-		lr.State.Events = gomatrixserverlib.ToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync)
+		lr.State.Events = gomatrixserverlib.HeaderedToClientEvents(delta.stateEvents, gomatrixserverlib.FormatSync)
 		res.Rooms.Leave[delta.roomID] = *lr
 	}
 
@@ -1074,7 +1074,7 @@ func (d *SyncServerDatasource) currentStateStreamEventsForRoom(
 	}
 	s := make([]types.StreamEvent, len(allState))
 	for i := 0; i < len(s); i++ {
-		s[i] = types.StreamEvent{Event: allState[i], StreamPosition: 0}
+		s[i] = types.StreamEvent{HeaderedEvent: allState[i], StreamPosition: 0}
 	}
 	return s, nil
 }
@@ -1082,10 +1082,10 @@ func (d *SyncServerDatasource) currentStateStreamEventsForRoom(
 // StreamEventsToEvents converts streamEvent to Event. If device is non-nil and
 // matches the streamevent.transactionID device then the transaction ID gets
 // added to the unsigned section of the output event.
-func (d *SyncServerDatasource) StreamEventsToEvents(device *authtypes.Device, in []types.StreamEvent) []gomatrixserverlib.Event {
-	out := make([]gomatrixserverlib.Event, len(in))
+func (d *SyncServerDatasource) StreamEventsToEvents(device *authtypes.Device, in []types.StreamEvent) []gomatrixserverlib.HeaderedEvent {
+	out := make([]gomatrixserverlib.HeaderedEvent, len(in))
 	for i := 0; i < len(in); i++ {
-		out[i] = in[i].Event
+		out[i] = in[i].HeaderedEvent
 		if device != nil && in[i].TransactionID != nil {
 			if device.UserID == in[i].Sender() && device.SessionID == in[i].TransactionID.SessionID {
 				err := out[i].SetUnsignedField(
@@ -1105,7 +1105,7 @@ func (d *SyncServerDatasource) StreamEventsToEvents(device *authtypes.Device, in
 // There may be some overlap where events in stateEvents are already in recentEvents, so filter
 // them out so we don't include them twice in the /sync response. They should be in recentEvents
 // only, so clients get to the correct state once they have rolled forward.
-func removeDuplicates(stateEvents, recentEvents []gomatrixserverlib.Event) []gomatrixserverlib.Event {
+func removeDuplicates(stateEvents, recentEvents []gomatrixserverlib.HeaderedEvent) []gomatrixserverlib.HeaderedEvent {
 	for _, recentEv := range recentEvents {
 		if recentEv.StateKey() == nil {
 			continue // not a state event