Added database.change_batch_retention, deleting old change batches

Each save that records a batch also deletes, in the same
transaction, the batches older than database.change_batch_retention
(30 days by default; 0 or less keeps them for ever), with their
records of opportunities. A client still holding a deleted batch
gets 410 Gone from ?changed_after= and reloads the whole list, as
for any unknown batch.

internal/cli/store.go

@@ -38,7 +38,11 @@ func OpenConfiguredStore(
 		return nil, fmt.Errorf("database %s: %w", file, err)
 	}
 
-	opportunityStore, err := store.Open(ctx, file)
+	opportunityStore, err := store.Open(
+		ctx,
+		file,
+		store.WithChangeBatchRetention(conf.Database.ChangeBatchRetention),
+	)
 	if errors.Is(err, store.ErrNeedsMigration) {
 		return nil, fmt.Errorf(
 			"database %s: %w: run `job-search migrate`", file, err,

internal/config/README.md

@@ -58,6 +58,7 @@ priority over `--quiet`.
 | Key | Type | Default | Notes |
 |---|---|---|---|
 | `database.file` | string | | Path to the SQLite database file. Empty: commands that need the database fail |
+| `database.change_batch_retention` | duration | `720h` | How long a batch of changed opportunities is kept for clients to catch up from (`?changed_after=`); older ones are deleted as new ones are recorded. A client whose batch has gone reloads the whole list. 0 or less: kept for ever |
 
 Database is currently restricted to SQLite v3. At this point, no other engine is
 supported.
@@ -219,6 +220,7 @@ logging:
     file: ""
 database:
     file: ./jobsearch.db
+    change_batch_retention: 720h
 sources:
     find_a_tender:
         enabled: true

internal/config/conf_database.go

@@ -1,15 +1,21 @@
 package config
 
-import "reflect"
+import (
+	"reflect"
+	"time"
+)
 
 // JSCKDatabase is the database section's configuration key.
 const JSCKDatabase string = "database"
 
 // JSCDatabase configures the SQLite database. File is the path to
 // the database file; empty means none, and commands that need the database
-// fail with a message saying so.
+// fail with a message saying so. ChangeBatchRetention is how long a batch of
+// changed opportunities is kept for clients to catch up from (see
+// store.WithChangeBatchRetention); 0 or less means they are kept for ever.
 type JSCDatabase struct {
 	File                 string        `cfg:"file"`
+	ChangeBatchRetention time.Duration `cfg:"change_batch_retention" default:"720h"`
 }
 
 // ---------- Section methods (see Section) ----------

internal/config/conf_sections_test.go

@@ -596,17 +596,23 @@ func TestLoadFromViperInvalidValues(t *testing.T) {
 func TestDatabaseSection(t *testing.T) {
 	c := JSCDatabase{
 		File:                 "./jobsearch.db",
+		ChangeBatchRetention: 48 * time.Hour,
 	}
 
-	if got, want := c.Keys(), []string{"file"}; !reflect.DeepEqual(got, want) {
-		t.Errorf("Keys = %v; want %v", got, want)
+	keys := []string{"file", "change_batch_retention"}
+	if got := c.Keys(); !reflect.DeepEqual(got, keys) {
+		t.Errorf("Keys = %v; want %v", got, keys)
 	}
 	if got, want := c.Map(), map[string]any{
 		"file":                   "./jobsearch.db",
+		"change_batch_retention": "48h0m0s",
 	}; !reflect.DeepEqual(got, want) {
 		t.Errorf("Map = %v; want %v", got, want)
 	}
-	want := map[string]any{"database.file": "./jobsearch.db"}
+	want := map[string]any{
+		"database.file":                   "./jobsearch.db",
+		"database.change_batch_retention": "48h0m0s",
+	}
 	if got := c.Flatten(); !reflect.DeepEqual(got, want) {
 		t.Errorf("Flatten = %v; want %v", got, want)
 	}
@@ -636,6 +642,19 @@ func TestDatabaseSection(t *testing.T) {
 		conf.Database.File != "./other.db" {
 		t.Errorf("SetValue(database.file) = %v, File = %q", err, conf.Database.File)
 	}
+
+	// 30 days by default.
+	retention := defaultConfig(t).Database.ChangeBatchRetention
+	if retention != 720*time.Hour {
+		t.Errorf("default change_batch_retention = %v; want 720h", retention)
+	}
+	err := conf.SetValue("database.change_batch_retention", "0s")
+	if err != nil || conf.Database.ChangeBatchRetention != 0 {
+		t.Errorf(
+			"SetValue(change_batch_retention) = %v, retention = %v",
+			err, conf.Database.ChangeBatchRetention,
+		)
+	}
 }
 
 func TestViperValue(t *testing.T) {

internal/store/README.md

@@ -97,6 +97,7 @@ A soft convention for the SQL in `queries.go`, followed so the queries look alik
 ```go
 s, err := store.Open(ctx, "jobsearch.db") // fails with ErrNeedsMigration if not migrated
 s, err := store.Open(ctx, path, store.WithAutoMigrate()) // tests: migrate on open
+s, err := store.Open(ctx, path, store.WithChangeBatchRetention(30*24*time.Hour))
 res, err := s.SaveOpportunity(ctx, &opportunity) // SaveResult{OpportunityUpdated, NewReleases}
 res, err := s.SaveOpportunities(ctx, opportunities) // all in one transaction; SaveMultipleResult
 err = s.SetMarks(ctx, types.OpportunityMarks{UUID: uuid, Interested: true}) // ErrNotFound
@@ -119,14 +120,14 @@ ok, err := s.ReleaseExists(ctx, uuid, releaseID)
 - **Minimal writing.** Releases and tags are write-once. A release already stored (same `uuid` and `release_id`) is skipped, and `NewReleases` counts only the new ones. `ReleaseExists` lets a fetcher skip converting a release it already has.
 - **Opportunity row updates.** The opportunity row and its classifications are replaced only if the save's newest release date is at least the stored `latest_release_date`. An older release arriving late is still stored, but never overwrites newer opportunity data; `OpportunityUpdated` reports which happened.
 - **Marks are kept.** An opportunity's marks are stored when it is first saved; saving it again never changes them, so fetched data (which carries no marks) can't reset what the user has set.
+- **Change batches.** The opportunities a save adds or gives new releases are recorded, in the same transaction, in a change batch named by a random hash; `SaveOpportunities` reports it as `ChangeBatch` ("" if nothing changed). A batch's `created_at` is kept later than every earlier batch's, even if the clock goes back. Recording one deletes the batches older than the store's retention (`WithChangeBatchRetention`; without it, or with 0 or less, they are kept for ever).
 
-**Marks.** `SetMarks` replaces an opportunity's marks, returning `ErrNotFound` if the opportunity isn't stored. `won` and `lost` exclude each other: `SetMarks`, `SaveOpportunity` and `SaveOpportunities` reject marks with both set with `types.ErrWonAndLost` (from `OpportunityMarks.Validate`), and the database's `CHECK` constraint backs this up.
+**Marks.** `SetMarks` replaces an opportunity's marks, returning `ErrNotFound` if the opportunity isn't stored. `won` and `lost` exclude each other: `SetMarks`, `SaveOpportunity` and `SaveOpportunities` reject marks with both set with `types.ErrWonAndLost` (from `OpportunityMarks.Validate`), and the database's `CHECK` constraint backs this up. A change of marks is recorded in a change batch of its own, so a client catching up sees it.
 
 
 **Listing.** `ListOpportunities` filters by status, classification code, title text, a range on `LatestReleaseDate` and each mark (`MarkSet` or `MarkUnset`; an opportunity with no marks stored has every mark unset), newest first, 50 per page by default. It loads classifications and marks but not releases (which can be large); those are loaded by `ListReleases` or `GetOpportunity`.
 
-
-**Changes.** Each save that adds opportunities or gives them new releases records them in a change batch, in the same transaction, and `SaveOpportunities` reports the batch's hash (`ChangeBatch`). `SetMarks` records the opportunity in a batch of its own. A client that read the list after batch B (`LatestChangeBatch`, read before listing) gets everything changed since from `ListChangedOpportunities(ctx, B)`, with the newest batch to ask from next time; an unknown hash is `ErrUnknownBatch`. Batches are ordered by `created_at`, which `recordChangeBatch` keeps strictly increasing even if the clock goes back, and only batches up to the newest when the call starts are read, so one recorded meanwhile is left for next time rather than missed. Batches are never deleted yet.
+`LatestChangeBatch` returns the hash of the newest change batch ("" if there is none). Read before listing, it is the batch to give `ListChangedOpportunities`, which returns the opportunities in every later batch, once each, unpaged and ordered as `ListOpportunities` orders them, with the hash of the newest of those batches to give it next time. Batches recorded while it runs are left for next time rather than missed. A hash it doesn't know, such as a deleted batch's, is `ErrUnknownBatch`.
 
 
 **Title search.** The words are matched against the title using FTS5: every word must appear, each also matches as a prefix, case and accents are ignored, and "migrating" finds "migration", etc. Input is sanitised, so FTS5 operators and quotes typed by a user are searched as plain words, never interpreted.

internal/store/batches_test.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -53,3 +54,71 @@ func TestChangeBatchAfterClockWentBack(t *testing.T) {
 		t.Errorf("changed after future: %v, %q", got, hash)
 	}
 }
+
+// Recording a batch deletes the batches older than the retention, with their
+// records of opportunities; with no retention, none are deleted.
+func TestChangeBatchRetention(t *testing.T) {
+	for _, retention := range []time.Duration{time.Hour, 0} {
+		db := openTestDB(t)
+		s := &Store{
+			db:                   db,
+			changeBatchRetention: retention,
+		}
+		ctx := context.Background()
+
+		_, err := s.SaveOpportunity(
+			ctx,
+			&types.Opportunity{
+				UUID:              "ocds-1",
+				LatestReleaseDate: time.Now(),
+			},
+		)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		// Backdate every batch so far by two hours.
+		db.MustExec(
+			`UPDATE opportunity_change_batches SET created_at = ?`,
+			time.Now().UTC().Add(-2*time.Hour),
+		)
+		old, err := s.LatestChangeBatch(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		err = s.SetMarks(ctx, types.OpportunityMarks{
+			UUID:    "ocds-1",
+			Skipped: true,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		var batches, records int
+		db.Get(&batches, `SELECT COUNT(*) FROM opportunity_change_batches`)
+		db.Get(
+			&records,
+			`SELECT COUNT(*) FROM opportunity_change_batch_opportunities`,
+		)
+		_, _, err = s.ListChangedOpportunities(ctx, old)
+
+		if retention > 0 {
+			if batches != 1 || records != 1 ||
+				!errors.Is(err, ErrUnknownBatch) {
+				t.Errorf(
+					"retention %v: %d batches, %d records, err %v; "+
+						"want the old batch gone",
+					retention, batches, records, err,
+				)
+			}
+			continue
+		}
+		if batches != 2 || records != 2 || err != nil {
+			t.Errorf(
+				"no retention: %d batches, %d records, err %v; want both kept",
+				batches, records, err,
+			)
+		}
+	}
+}

internal/store/common.go

@@ -3,6 +3,7 @@ package store
 import (
 	"database/sql"
 	"fmt"
+	"time"
 )
 
 // Option configures Open.
@@ -10,6 +11,7 @@ type Option func(*options)
 
 type options struct {
 	autoMigrate          bool
+	changeBatchRetention time.Duration
 }
 
 // OpenDB opens the SQLite database at path with the store's connection
@@ -29,6 +31,16 @@ func WithAutoMigrate() Option {
 	}
 }
 
+// WithChangeBatchRetention makes the store delete change batches older than
+// retention whenever it records a new one, so a client can catch up from a
+// batch for that long (see ListChangedOpportunities). Without it, or with 0
+// or less, batches are kept for ever.
+func WithChangeBatchRetention(retention time.Duration) Option {
+	return func(opts *options) {
+		opts.changeBatchRetention = retention
+	}
+}
+
 // dsn builds the mattn/go-sqlite3 connection string.
 //
 //   - _foreign_keys: SQLite ignores REFERENCES clauses unless this is on.

internal/store/queries.go

@@ -294,6 +294,11 @@ FROM opportunity_change_batches
 ORDER BY created_at DESC
 LIMIT 1`
 
+// deleteChangeBatchesBefore deletes the batches created before a time, and
+// (by cascade) the records of their opportunities.
+const deleteChangeBatchesBefore = `
+DELETE FROM opportunity_change_batches WHERE created_at < ?`
+
 // selectChangeBatchByHash reads the batch with a hash into a changeBatch.
 const selectChangeBatchByHash = `
 SELECT * FROM opportunity_change_batches WHERE hash = ?`

internal/store/queries_test.go

@@ -117,6 +117,9 @@ var checks = map[string]queryCheck{
 	"insertChangeBatch":       {sql: insertChangeBatch},
 	"selectLatestChangeBatch": {sql: selectLatestChangeBatch},
 	"selectChangeBatchByHash": {sql: selectChangeBatchByHash},
+	"deleteChangeBatchesBefore": {
+		sql: deleteChangeBatchesBefore,
+	},
 
 	// opportunity_change_batch_opportunities
 	"insertChangeBatchOpportunity": {sql: insertChangeBatchOpportunity},

internal/store/store.go

@@ -51,6 +51,9 @@ const DefaultListLimit = 50
 // concurrent use.
 type Store struct {
 	db *sqlx.DB
+	// changeBatchRetention is how long change batches are kept; 0 or less
+	// means for ever (see WithChangeBatchRetention).
+	changeBatchRetention time.Duration
 }
 
 // SaveMultipleResult reports what SaveOpportunities changed.
@@ -244,6 +247,7 @@ func Open(ctx context.Context, path string, opts ...Option) (*Store, error) {
 
 	return &Store{
 		db:                   sqlx.NewDb(sqldb, "sqlite3"),
+		changeBatchRetention: settings.changeBatchRetention,
 	}, nil
 }
 
@@ -307,7 +311,7 @@ func (s *Store) SaveOpportunity(
 		return SaveResult{}, err
 	}
 
-	_, err = recordChangeBatch(
+	_, err = s.recordChangeBatch(
 		ctx,
 		tx,
 		changedUUIDs([]*types.Opportunity{opportunity}, []SaveResult{result}),
@@ -387,7 +391,7 @@ func (s *Store) SaveOpportunities(
 	total.AddedClassifications = added
 	total.RemovedClassificationIDs = removed
 
-	total.ChangeBatch, err = recordChangeBatch(
+	total.ChangeBatch, err = s.recordChangeBatch(
 		ctx, tx, changedUUIDs(opportunities, total.OpportunityResults),
 	)
 	if err != nil {
@@ -465,14 +469,15 @@ func changedUUIDs(
 
 // recordChangeBatch records a change batch of the opportunities uuids inside
 // tx and returns its hash, or "" without recording anything if uuids is
-// empty.
+// empty. Batches older than the store's retention are deleted (see
+// WithChangeBatchRetention).
 //
 // The batch's created_at is the current time, or just after the newest
 // batch's if that isn't later (the clock went back, or two batches came
 // within its resolution), so batches are ordered by created_at. tx holds the
 // write lock from its start (see dsn), so no other batch can be recorded in
 // between.
-func recordChangeBatch(
+func (s *Store) recordChangeBatch(
 	ctx context.Context, tx *sqlx.Tx, uuids []string,
 ) (string, error) {
 	if len(uuids) == 0 {
@@ -506,6 +511,14 @@ func recordChangeBatch(
 		}
 	}
 
+	if s.changeBatchRetention > 0 {
+		cutoff := createdAt.Add(-s.changeBatchRetention)
+		_, err = tx.ExecContext(ctx, deleteChangeBatchesBefore, cutoff)
+		if err != nil {
+			return "", fmt.Errorf("deleting old change batches: %w", err)
+		}
+	}
+
 	return hash, nil
 }
 
@@ -681,7 +694,7 @@ func (s *Store) SetMarks(
 		return err
 	}
 
-	_, err = recordChangeBatch(ctx, tx, []string{marks.UUID})
+	_, err = s.recordChangeBatch(ctx, tx, []string{marks.UUID})
 	if err != nil {
 		return err
 	}