diff --git a/internal/cli/store.go b/internal/cli/store.go index d8f441e..2c3a817 100644 --- a/internal/cli/store.go +++ b/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, diff --git a/internal/config/README.md b/internal/config/README.md index e70a66c..be66e05 100644 --- a/internal/config/README.md +++ b/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 diff --git a/internal/config/conf_database.go b/internal/config/conf_database.go index c644797..35f3847 100644 --- a/internal/config/conf_database.go +++ b/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"` + File string `cfg:"file"` + ChangeBatchRetention time.Duration `cfg:"change_batch_retention" default:"720h"` } // ---------- Section methods (see Section) ---------- diff --git a/internal/config/conf_sections_test.go b/internal/config/conf_sections_test.go index 8b6d198..3ef8e35 100644 --- a/internal/config/conf_sections_test.go +++ b/internal/config/conf_sections_test.go @@ -595,18 +595,24 @@ func TestLoadFromViperInvalidValues(t *testing.T) { func TestDatabaseSection(t *testing.T) { c := JSCDatabase{ - File: "./jobsearch.db", + 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", + "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) { diff --git a/internal/store/README.md b/internal/store/README.md index 3c4ef31..4b07ea2 100644 --- a/internal/store/README.md +++ b/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. diff --git a/internal/store/batches_test.go b/internal/store/batches_test.go index d6bc0bf..6ed005e 100644 --- a/internal/store/batches_test.go +++ b/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, + ) + } + } +} diff --git a/internal/store/common.go b/internal/store/common.go index bfafe35..f3516f2 100644 --- a/internal/store/common.go +++ b/internal/store/common.go @@ -3,13 +3,15 @@ package store import ( "database/sql" "fmt" + "time" ) // Option configures Open. type Option func(*options) type options struct { - autoMigrate bool + 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. diff --git a/internal/store/queries.go b/internal/store/queries.go index 9992b75..f83c5e9 100644 --- a/internal/store/queries.go +++ b/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 = ?` diff --git a/internal/store/queries_test.go b/internal/store/queries_test.go index 45ff336..6f3cdd5 100644 --- a/internal/store/queries_test.go +++ b/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}, diff --git a/internal/store/store.go b/internal/store/store.go index b56ba6d..46e4bfb 100644 --- a/internal/store/store.go +++ b/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. @@ -243,7 +246,8 @@ func Open(ctx context.Context, path string, opts ...Option) (*Store, error) { } return &Store{ - db: sqlx.NewDb(sqldb, "sqlite3"), + 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 }