From 4656ebf9de516483b678c790ab734f4cac5dd31f Mon Sep 17 00:00:00 2001
From: Ashley Nelson <fant@shley.email>
Date: Sun, 2 Oct 2022 20:03:18 +0000
Subject: [PATCH] Add function to convert current room state to StreamEvents

The current room state table records were previously reusing a function
for converting output room event rows into StreamEvents. This
required altering the query to produce additional columns required by
output room events, which is error prone. This commit creates a function
explicitly for the current room state table so the query can be changed
to only select fields present in the table.
---
 .../postgres/current_room_state_table.go      | 38 +++++++++++++++----
 .../sqlite3/current_room_state_table.go       | 38 +++++++++++++++----
 2 files changed, 62 insertions(+), 14 deletions(-)

diff --git a/syncapi/storage/postgres/current_room_state_table.go b/syncapi/storage/postgres/current_room_state_table.go
index 4ffd29610..2ccf0be1a 100644
--- a/syncapi/storage/postgres/current_room_state_table.go
+++ b/syncapi/storage/postgres/current_room_state_table.go
@@ -104,12 +104,7 @@ const selectStateEventSQL = "" +
 	"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
-	// the rowsToStreamEvents expects there to be exactly seven 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 event_id, added_at, headered_event_json, 0 AS session_id, false AS exclude_from_sync, '' AS transaction_id, history_visibility" +
-	" FROM syncapi_current_room_state WHERE event_id = ANY($1)"
+	"SELECT event_id, added_at, headered_event_json, history_visibility FROM syncapi_current_room_state WHERE event_id = ANY($1)"
 
 const selectSharedUsersSQL = "" +
 	"SELECT state_key FROM syncapi_current_room_state WHERE room_id = ANY(" +
@@ -365,7 +360,36 @@ func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
 		return nil, err
 	}
 	defer internal.CloseAndLogIfError(ctx, rows, "selectEventsWithEventIDs: rows.close() failed")
-	return rowsToStreamEvents(rows)
+	return currentRoomStateRowsToStreamEvents(rows)
+}
+
+func currentRoomStateRowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
+	var events []types.StreamEvent
+	for rows.Next() {
+		var (
+			eventID           string
+			streamPos         types.StreamPosition
+			eventBytes        []byte
+			historyVisibility gomatrixserverlib.HistoryVisibility
+		)
+		if err := rows.Scan(&eventID, &streamPos, &eventBytes, &historyVisibility); err != nil {
+			return nil, err
+		}
+		// TODO: Handle redacted events
+		var ev gomatrixserverlib.HeaderedEvent
+		if err := ev.UnmarshalJSONWithEventID(eventBytes, eventID); err != nil {
+			return nil, err
+		}
+
+		ev.Visibility = historyVisibility
+
+		events = append(events, types.StreamEvent{
+			HeaderedEvent:  &ev,
+			StreamPosition: streamPos,
+		})
+	}
+
+	return events, nil
 }
 
 func rowsToEvents(rows *sql.Rows) ([]*gomatrixserverlib.HeaderedEvent, error) {
diff --git a/syncapi/storage/sqlite3/current_room_state_table.go b/syncapi/storage/sqlite3/current_room_state_table.go
index c4019fed2..ff45e786e 100644
--- a/syncapi/storage/sqlite3/current_room_state_table.go
+++ b/syncapi/storage/sqlite3/current_room_state_table.go
@@ -88,12 +88,7 @@ const selectStateEventSQL = "" +
 	"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
-	// the rowsToStreamEvents expects there to be exactly seven 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 event_id, added_at, headered_event_json, 0 AS session_id, false AS exclude_from_sync, '' AS transaction_id, history_visibility" +
-	" FROM syncapi_current_room_state WHERE event_id IN ($1)"
+	"SELECT event_id, added_at, headered_event_json, history_visibility FROM syncapi_current_room_state WHERE event_id IN ($1)"
 
 const selectSharedUsersSQL = "" +
 	"SELECT state_key FROM syncapi_current_room_state WHERE room_id IN(" +
@@ -378,7 +373,7 @@ func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
 			return nil, err
 		}
 		start = start + n
-		events, err := rowsToStreamEvents(rows)
+		events, err := currentRoomStateRowsToStreamEvents(rows)
 		internal.CloseAndLogIfError(ctx, rows, "selectEventsWithEventIDs: rows.close() failed")
 		if err != nil {
 			return nil, err
@@ -388,6 +383,35 @@ func (s *currentRoomStateStatements) SelectEventsWithEventIDs(
 	return res, nil
 }
 
+func currentRoomStateRowsToStreamEvents(rows *sql.Rows) ([]types.StreamEvent, error) {
+	var events []types.StreamEvent
+	for rows.Next() {
+		var (
+			eventID           string
+			streamPos         types.StreamPosition
+			eventBytes        []byte
+			historyVisibility gomatrixserverlib.HistoryVisibility
+		)
+		if err := rows.Scan(&eventID, &streamPos, &eventBytes, &historyVisibility); err != nil {
+			return nil, err
+		}
+		// TODO: Handle redacted events
+		var ev gomatrixserverlib.HeaderedEvent
+		if err := ev.UnmarshalJSONWithEventID(eventBytes, eventID); err != nil {
+			return nil, err
+		}
+
+		ev.Visibility = historyVisibility
+
+		events = append(events, types.StreamEvent{
+			HeaderedEvent:  &ev,
+			StreamPosition: streamPos,
+		})
+	}
+
+	return events, nil
+}
+
 func rowsToEvents(rows *sql.Rows) ([]*gomatrixserverlib.HeaderedEvent, error) {
 	result := []*gomatrixserverlib.HeaderedEvent{}
 	for rows.Next() {