Consistency

message/store.go

@@ -299,6 +299,8 @@ func (c *commonStore) Message(id string) (*model.Message, error) {
 
 // UpdateMessageTime updates the time column for a message by ID. This is only used for testing.
 func (c *commonStore) UpdateMessageTime(messageID string, timestamp int64) error {
+	c.maybeLock()
+	defer c.maybeUnlock()
 	_, err := c.db.Exec(c.queries.updateMessageTime, timestamp, messageID)
 	return err
 }

webpush/store.go

@@ -157,7 +157,7 @@ func (s *commonStore) RemoveExpiredSubscriptions(expireAfter time.Duration) erro
 }
 
 // SetSubscriptionUpdatedAt updates the updated_at timestamp for a subscription by endpoint. This is
-// exported for testing purposes and is not part of the Store interface.
+// exported for testing purposes.
 func (s *commonStore) SetSubscriptionUpdatedAt(endpoint string, updatedAt int64) error {
 	_, err := s.db.Exec(s.queries.updateSubscriptionUpdatedAt, updatedAt, endpoint)
 	return err

webpush/store_sqlite.go

@@ -9,7 +9,6 @@ import (
 
 const (
 	sqliteCreateWebPushSubscriptionsTableQuery = `
-		BEGIN;
 		CREATE TABLE IF NOT EXISTS subscription (
 			id TEXT PRIMARY KEY,
 			endpoint TEXT NOT NULL,
@@ -32,8 +31,7 @@ const (
 		CREATE TABLE IF NOT EXISTS schemaVersion (
 			id INT PRIMARY KEY,
 			version INT NOT NULL
-		);			
-		COMMIT;
+		);
 	`
 	sqliteBuiltinStartupQueries = `
 		PRAGMA foreign_keys = ON;
@@ -122,13 +120,18 @@ func setupSQLite(db *sql.DB) error {
 }
 
 func setupNewSQLite(db *sql.DB) error {
-	if _, err := db.Exec(sqliteCreateWebPushSubscriptionsTableQuery); err != nil {
+	tx, err := db.Begin()
+	if err != nil {
 		return err
 	}
-	if _, err := db.Exec(sqliteInsertWebPushSchemaVersionQuery, sqliteCurrentWebPushSchemaVersion); err != nil {
+	defer tx.Rollback()
+	if _, err := tx.Exec(sqliteCreateWebPushSubscriptionsTableQuery); err != nil {
 		return err
 	}
-	return nil
+	if _, err := tx.Exec(sqliteInsertWebPushSchemaVersionQuery, sqliteCurrentWebPushSchemaVersion); err != nil {
+		return err
+	}
+	return tx.Commit()
 }
 
 func runSQLiteStartupQueries(db *sql.DB, startupQueries string) error {