Use TransactionWriter in other component SQLite (#1209)

* Use TransactionWriter on other component SQLites

* Fix sync API tests

* Fix panic in media API

* Fix a couple of transactions

* Fix wrong query, add some logging output

* Add debug logging into StoreEvent

* Adjust InsertRoomNID

* Update logging

2 files changed, 32 insertions, 12 deletions

diff --git a/appservice/storage/sqlite3/appservice_events_table.go b/appservice/storage/sqlite3/appservice_events_table.go
index 479f2213..da31f235 100644
--- a/appservice/storage/sqlite3/appservice_events_table.go
+++ b/appservice/storage/sqlite3/appservice_events_table.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"time"
 
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 	"github.com/matrix-org/gomatrixserverlib"
 	log "github.com/sirupsen/logrus"
 )
@@ -65,6 +66,8 @@ const (
 )
 
 type eventsStatements struct {
+	db                                     *sql.DB
+	writer                                 *sqlutil.TransactionWriter
 	selectEventsByApplicationServiceIDStmt *sql.Stmt
 	countEventsByApplicationServiceIDStmt  *sql.Stmt
 	insertEventStmt                        *sql.Stmt
@@ -73,6 +76,8 @@ type eventsStatements struct {
 }
 
 func (s *eventsStatements) prepare(db *sql.DB) (err error) {
+	s.db = db
+	s.writer = sqlutil.NewTransactionWriter()
 	_, err = db.Exec(appserviceEventsSchema)
 	if err != nil {
 		return
@@ -217,13 +222,15 @@ func (s *eventsStatements) insertEvent(
 		return err
 	}
 
-	_, err = s.insertEventStmt.ExecContext(
-		ctx,
-		appServiceID,
-		eventJSON,
-		-1, // No transaction ID yet
-	)
-	return
+	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+		_, err := s.insertEventStmt.ExecContext(
+			ctx,
+			appServiceID,
+			eventJSON,
+			-1, // No transaction ID yet
+		)
+		return err
+	})
 }
 
 // updateTxnIDForEvents sets the transactionID for a collection of events. Done
@@ -234,8 +241,10 @@ func (s *eventsStatements) updateTxnIDForEvents(
 	appserviceID string,
 	maxID, txnID int,
 ) (err error) {
-	_, err = s.updateTxnIDForEventsStmt.ExecContext(ctx, txnID, appserviceID, maxID)
-	return
+	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+		_, err := s.updateTxnIDForEventsStmt.ExecContext(ctx, txnID, appserviceID, maxID)
+		return err
+	})
 }
 
 // deleteEventsBeforeAndIncludingID removes events matching given IDs from the database.
@@ -244,6 +253,8 @@ func (s *eventsStatements) deleteEventsBeforeAndIncludingID(
 	appserviceID string,
 	eventTableID int,
 ) (err error) {
-	_, err = s.deleteEventsBeforeAndIncludingIDStmt.ExecContext(ctx, appserviceID, eventTableID)
-	return
+	return s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+		_, err := s.deleteEventsBeforeAndIncludingIDStmt.ExecContext(ctx, appserviceID, eventTableID)
+		return err
+	})
 }
diff --git a/appservice/storage/sqlite3/txn_id_counter_table.go b/appservice/storage/sqlite3/txn_id_counter_table.go
index b1ee6076..501ab5aa 100644
--- a/appservice/storage/sqlite3/txn_id_counter_table.go
+++ b/appservice/storage/sqlite3/txn_id_counter_table.go
@@ -18,6 +18,8 @@ package sqlite3
 import (
 	"context"
 	"database/sql"
+
+	"github.com/matrix-org/dendrite/internal/sqlutil"
 )
 
 const txnIDSchema = `
@@ -35,10 +37,14 @@ const selectTxnIDSQL = `
 `
 
 type txnStatements struct {
+	db              *sql.DB
+	writer          *sqlutil.TransactionWriter
 	selectTxnIDStmt *sql.Stmt
 }
 
 func (s *txnStatements) prepare(db *sql.DB) (err error) {
+	s.db = db
+	s.writer = sqlutil.NewTransactionWriter()
 	_, err = db.Exec(txnIDSchema)
 	if err != nil {
 		return
@@ -55,6 +61,9 @@ func (s *txnStatements) prepare(db *sql.DB) (err error) {
 func (s *txnStatements) selectTxnID(
 	ctx context.Context,
 ) (txnID int, err error) {
-	err = s.selectTxnIDStmt.QueryRowContext(ctx).Scan(&txnID)
+	err = s.writer.Do(s.db, nil, func(txn *sql.Tx) error {
+		err := s.selectTxnIDStmt.QueryRowContext(ctx).Scan(&txnID)
+		return err
+	})
 	return
 }