Convert events/event_json tables to share code (#1062)

* Convert event_json table

* Convert the events table

1 files changed, 23 insertions, 20 deletions

diff --git a/roomserver/storage/postgres/events_table.go b/roomserver/storage/postgres/events_table.go
index c28fa8e6..9c464946 100644
--- a/roomserver/storage/postgres/events_table.go
+++ b/roomserver/storage/postgres/events_table.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/lib/pq"
 	"github.com/matrix-org/dendrite/internal"
+	"github.com/matrix-org/dendrite/roomserver/storage/tables"
 	"github.com/matrix-org/dendrite/roomserver/types"
 	"github.com/matrix-org/gomatrixserverlib"
 )
@@ -136,13 +137,14 @@ type eventStatements struct {
 	selectRoomNIDForEventNIDStmt           *sql.Stmt
 }
 
-func (s *eventStatements) prepare(db *sql.DB) (err error) {
-	_, err = db.Exec(eventsSchema)
+func NewPostgresEventsTable(db *sql.DB) (tables.Events, error) {
+	s := &eventStatements{}
+	_, err := db.Exec(eventsSchema)
 	if err != nil {
-		return
+		return nil, err
 	}
 
-	return statementList{
+	return s, statementList{
 		{&s.insertEventStmt, insertEventSQL},
 		{&s.selectEventStmt, selectEventSQL},
 		{&s.bulkSelectStateEventByIDStmt, bulkSelectStateEventByIDSQL},
@@ -160,8 +162,9 @@ func (s *eventStatements) prepare(db *sql.DB) (err error) {
 	}.prepare(db)
 }
 
-func (s *eventStatements) insertEvent(
+func (s *eventStatements) InsertEvent(
 	ctx context.Context,
+	txn *sql.Tx,
 	roomNID types.RoomNID,
 	eventTypeNID types.EventTypeNID,
 	eventStateKeyNID types.EventStateKeyNID,
@@ -179,8 +182,8 @@ func (s *eventStatements) insertEvent(
 	return types.EventNID(eventNID), types.StateSnapshotNID(stateNID), err
 }
 
-func (s *eventStatements) selectEvent(
-	ctx context.Context, eventID string,
+func (s *eventStatements) SelectEvent(
+	ctx context.Context, txn *sql.Tx, eventID string,
 ) (types.EventNID, types.StateSnapshotNID, error) {
 	var eventNID int64
 	var stateNID int64
@@ -190,7 +193,7 @@ func (s *eventStatements) selectEvent(
 
 // bulkSelectStateEventByID lookups a list of state events by event ID.
 // If any of the requested events are missing from the database it returns a types.MissingEventError
-func (s *eventStatements) bulkSelectStateEventByID(
+func (s *eventStatements) BulkSelectStateEventByID(
 	ctx context.Context, eventIDs []string,
 ) ([]types.StateEntry, error) {
 	rows, err := s.bulkSelectStateEventByIDStmt.QueryContext(ctx, pq.StringArray(eventIDs))
@@ -233,7 +236,7 @@ func (s *eventStatements) bulkSelectStateEventByID(
 // bulkSelectStateAtEventByID lookups the state at a list of events by event ID.
 // If any of the requested events are missing from the database it returns a types.MissingEventError.
 // If we do not have the state for any of the requested events it returns a types.MissingEventError.
-func (s *eventStatements) bulkSelectStateAtEventByID(
+func (s *eventStatements) BulkSelectStateAtEventByID(
 	ctx context.Context, eventIDs []string,
 ) ([]types.StateAtEvent, error) {
 	rows, err := s.bulkSelectStateAtEventByIDStmt.QueryContext(ctx, pq.StringArray(eventIDs))
@@ -270,14 +273,14 @@ func (s *eventStatements) bulkSelectStateAtEventByID(
 	return results, nil
 }
 
-func (s *eventStatements) updateEventState(
+func (s *eventStatements) UpdateEventState(
 	ctx context.Context, eventNID types.EventNID, stateNID types.StateSnapshotNID,
 ) error {
 	_, err := s.updateEventStateStmt.ExecContext(ctx, int64(eventNID), int64(stateNID))
 	return err
 }
 
-func (s *eventStatements) selectEventSentToOutput(
+func (s *eventStatements) SelectEventSentToOutput(
 	ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
 ) (sentToOutput bool, err error) {
 	stmt := internal.TxStmt(txn, s.selectEventSentToOutputStmt)
@@ -285,13 +288,13 @@ func (s *eventStatements) selectEventSentToOutput(
 	return
 }
 
-func (s *eventStatements) updateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error {
+func (s *eventStatements) UpdateEventSentToOutput(ctx context.Context, txn *sql.Tx, eventNID types.EventNID) error {
 	stmt := internal.TxStmt(txn, s.updateEventSentToOutputStmt)
 	_, err := stmt.ExecContext(ctx, int64(eventNID))
 	return err
 }
 
-func (s *eventStatements) selectEventID(
+func (s *eventStatements) SelectEventID(
 	ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
 ) (eventID string, err error) {
 	stmt := internal.TxStmt(txn, s.selectEventIDStmt)
@@ -299,7 +302,7 @@ func (s *eventStatements) selectEventID(
 	return
 }
 
-func (s *eventStatements) bulkSelectStateAtEventAndReference(
+func (s *eventStatements) BulkSelectStateAtEventAndReference(
 	ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID,
 ) ([]types.StateAtEventAndReference, error) {
 	stmt := internal.TxStmt(txn, s.bulkSelectStateAtEventAndReferenceStmt)
@@ -341,8 +344,8 @@ func (s *eventStatements) bulkSelectStateAtEventAndReference(
 	return results, nil
 }
 
-func (s *eventStatements) bulkSelectEventReference(
-	ctx context.Context, eventNIDs []types.EventNID,
+func (s *eventStatements) BulkSelectEventReference(
+	ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID,
 ) ([]gomatrixserverlib.EventReference, error) {
 	rows, err := s.bulkSelectEventReferenceStmt.QueryContext(ctx, eventNIDsAsArray(eventNIDs))
 	if err != nil {
@@ -367,7 +370,7 @@ func (s *eventStatements) bulkSelectEventReference(
 }
 
 // bulkSelectEventID returns a map from numeric event ID to string event ID.
-func (s *eventStatements) bulkSelectEventID(ctx context.Context, eventNIDs []types.EventNID) (map[types.EventNID]string, error) {
+func (s *eventStatements) BulkSelectEventID(ctx context.Context, eventNIDs []types.EventNID) (map[types.EventNID]string, error) {
 	rows, err := s.bulkSelectEventIDStmt.QueryContext(ctx, eventNIDsAsArray(eventNIDs))
 	if err != nil {
 		return nil, err
@@ -394,7 +397,7 @@ func (s *eventStatements) bulkSelectEventID(ctx context.Context, eventNIDs []typ
 
 // bulkSelectEventNIDs returns a map from string event ID to numeric event ID.
 // If an event ID is not in the database then it is omitted from the map.
-func (s *eventStatements) bulkSelectEventNID(ctx context.Context, eventIDs []string) (map[string]types.EventNID, error) {
+func (s *eventStatements) BulkSelectEventNID(ctx context.Context, eventIDs []string) (map[string]types.EventNID, error) {
 	rows, err := s.bulkSelectEventNIDStmt.QueryContext(ctx, pq.StringArray(eventIDs))
 	if err != nil {
 		return nil, err
@@ -412,7 +415,7 @@ func (s *eventStatements) bulkSelectEventNID(ctx context.Context, eventIDs []str
 	return results, rows.Err()
 }
 
-func (s *eventStatements) selectMaxEventDepth(ctx context.Context, eventNIDs []types.EventNID) (int64, error) {
+func (s *eventStatements) SelectMaxEventDepth(ctx context.Context, txn *sql.Tx, eventNIDs []types.EventNID) (int64, error) {
 	var result int64
 	stmt := s.selectMaxEventDepthStmt
 	err := stmt.QueryRowContext(ctx, eventNIDsAsArray(eventNIDs)).Scan(&result)
@@ -422,7 +425,7 @@ func (s *eventStatements) selectMaxEventDepth(ctx context.Context, eventNIDs []t
 	return result, nil
 }
 
-func (s *eventStatements) selectRoomNIDForEventNID(
+func (s *eventStatements) SelectRoomNIDForEventNID(
 	ctx context.Context, txn *sql.Tx, eventNID types.EventNID,
 ) (roomNID types.RoomNID, err error) {
 	selectStmt := internal.TxStmt(txn, s.selectRoomNIDForEventNIDStmt)