feat: Use transactions when storing feeds to local database (#157)

This updates the store interface with a BeginBatch() method that causes
subsequent calls to UpsertItem() to happen inside a transaction (until
EndBatch() is called).

We then modify fetchAllFeeds to use the updated API to wrap its outer loop
in a transaction.

In a small test (6 feeds, about 2500 items) this reduced the time required
to run `refresh` from 1 minutes to 1 second. That's a 98% improvement!

Closes: #156

Author: larsks

internal/commands/commands.go

@@ -151,6 +151,12 @@ func (c Commands) fetchAllFeeds() ([]store.Item, []ErrorItem, error) {
 		close(ch)
 	}()
 
+	err := c.store.BeginBatch()
+	if err != nil {
+		return items, errorItems, fmt.Errorf("fetchAllFeeds: failed to begin batch: %w", err)
+	}
+	defer c.store.EndBatch()
+
 	for result := range ch {
 		if result.err != nil {
 			errorItems = append(errorItems, ErrorItem{FeedURL: result.url, Err: result.err})

internal/store/store.go

@@ -34,6 +34,8 @@ func (i Item) Read() bool {
 
 type Store interface {
 	UpsertItem(item Item) error
+	BeginBatch() error
+	EndBatch() error
 	GetAllItems(ordering string) ([]Item, error)
 	GetItemByID(ID int) (Item, error)
 	GetAllFeedURLs() ([]string, error)
@@ -45,8 +47,9 @@ type Store interface {
 }
 
 type SQLiteStore struct {
-	path string
-	db   *sql.DB
+	path  string
+	db    *sql.DB
+	batch *sql.Tx
 }
 
 func NewSQLiteStore(basePath string, dbName string) (*SQLiteStore, error) {
@@ -148,8 +151,46 @@ func runMigrations(db *sql.DB) (err error) {
 	return err
 }
 
-func (sls SQLiteStore) UpsertItem(item Item) error {
-	stmt, err := sls.db.Prepare(`select count(id), id from items where feedurl = ? and title = ?;`)
+// Begin a transaction. UpsertItem will use this transaction until
+// client code calls EndBatch().
+func (sls *SQLiteStore) BeginBatch() error {
+	tx, err := sls.db.Begin()
+	if err != nil {
+		return fmt.Errorf("BeginBatch: %w", err)
+	}
+	sls.batch = tx
+	return nil
+}
+
+// Commit a transaction. This commits any changes made since BeginBatch()
+// and reset sls.batch to nil so that subsequent calls to UpsertItem()
+// will not use a transaction.
+func (sls *SQLiteStore) EndBatch() error {
+	if sls.batch == nil {
+		return nil
+	}
+	err := sls.batch.Commit()
+	sls.batch = nil
+	if err != nil {
+		return fmt.Errorf("EndBatch: %w", err)
+	}
+	return nil
+}
+
+func (sls *SQLiteStore) UpsertItem(item Item) error {
+	if sls.batch != nil {
+		return sls.upsertItem(sls.batch, item)
+	}
+	return sls.upsertItem(sls.db, item)
+}
+
+// This interface is implemented by both sql.DB and by sql.Tx
+type statementPreparer interface {
+	Prepare(query string) (*sql.Stmt, error)
+}
+
+func (sls *SQLiteStore) upsertItem(db statementPreparer, item Item) error {
+	stmt, err := db.Prepare(`select count(id), id from items where feedurl = ? and title = ?;`)
 	if err != nil {
 		return fmt.Errorf("sqlite.go: could not prepare query: %w", err)
 	}
@@ -162,7 +203,7 @@ func (sls SQLiteStore) UpsertItem(item Item) error {
 	}
 
 	if count == 0 {
-		stmt, err = sls.db.Prepare(`insert into items (feedurl, link, title, content, author, publishedat, createdat, updatedat) values (?, ?, ?, ?, ?, ?, ?, ?)`)
+		stmt, err = db.Prepare(`insert into items (feedurl, link, title, content, author, publishedat, createdat, updatedat) values (?, ?, ?, ?, ?, ?, ?, ?)`)
 		if err != nil {
 			return fmt.Errorf("sqlite.go: could not prepare query: %w", err)
 		}
@@ -172,7 +213,7 @@ func (sls SQLiteStore) UpsertItem(item Item) error {
 			return fmt.Errorf("sqlite.go: Upsert failed: %w", err)
 		}
 	} else {
-		stmt, err = sls.db.Prepare(`update items set content = ?, updatedat = ? where id = ?`)
+		stmt, err = db.Prepare(`update items set content = ?, updatedat = ? where id = ?`)
 		if err != nil {
 			return fmt.Errorf("sqlite.go: could not prepare query: %w", err)
 		}