Added change batches, so clients can catch up on changed opportunities

Every save that stores new opportunities or new releases records them in a
change batch (migration 00008: opportunity_change_batches and
opportunity_change_batch_opportunities; the opportunities table is
unchanged), as does every change of marks. A batch is named by a random
SHA-256 hash, so it can't be guessed, and ordered by its created_at, which
is kept strictly increasing even if the clock goes back.

GET /api/opportunities pages now carry the newest batch, read before the
page, and GET /api/opportunities?changed_after=<batch> returns every
opportunity changed since that batch, unpaged, with the newest batch it
includes. Offset and limit can't be combined with it, and an unknown batch
is 410 Gone, telling the client to reload the whole list.

OpportunitiesUpdated now carries its save's batch, and a "large" flag: over
50 opportunities, new and updated are left empty and only the summary
counts them, so a big catch-up no longer sends tens of thousands of UUIDs.

internal/api/opportunities.go

@@ -24,11 +24,17 @@ var markIndex, markNames = findMarks()
 
 // OpportunityPage is a page of opportunities, as GET /opportunities returns
 // it. HasMore reports that another page follows, at Offset+Limit.
+//
+// Batch is the hash of the newest change batch when the page was read, or
+// "" if none has been recorded: given as changed_after, it gets the
+// opportunities changed since. A page of changed opportunities is all of
+// them, so Limit is their number and HasMore is false.
 type OpportunityPage struct {
 	Opportunities []types.Opportunity `json:"opportunities"`
 	Offset        int                 `json:"offset"`
 	Limit         int                 `json:"limit"`
 	HasMore       bool                `json:"has_more"`
+	Batch         string              `json:"batch"`
 }
 
 // opportunityRoutes wires HTTP handlers to the opportunity store.
@@ -65,22 +71,38 @@ func (h *opportunityRoutes) Routes(router chi.Router) {
 // list serves GET /opportunities: a page of opportunities, newest release
 // first, with their classifications and marks but not their releases. The
 // page is set by the offset and limit (default 50, at most 100) query
-// parameters.
+// parameters. With changed_after, it serves the opportunities changed since
+// that batch instead (see listChanged).
 //
 //	@Summary		List opportunities
 //	@Description	A page of opportunities, newest release first, with their
 //	@Description	classifications and marks but not their releases. has_more
-//	@Description	says whether another page follows, at offset+limit.
+//	@Description	says whether another page follows, at offset+limit. batch
+//	@Description	is the newest change batch when the page was read.
+//	@Description	With changed_after (a batch from a page or an
+//	@Description	OpportunitiesUpdated event), the page is instead every
+//	@Description	opportunity changed since that batch (new, given new
+//	@Description	releases or with changed marks), and batch is the newest
+//	@Description	batch they include. offset and limit can't be given with
+//	@Description	changed_after. An unknown batch is 410 Gone: reload the
+//	@Description	whole list.
 //	@Tags			opportunities
 //	@Produce		json
-//	@Param			offset	query		int	false	"Opportunities to skip"	minimum(0)	default(0)
+//	@Param			offset			query		int		false	"Opportunities to skip"	minimum(0)	default(0)
-//	@Param			limit	query		int	false	"Page size"				minimum(1)	maximum(100)	default(50)
+//	@Param			limit			query		int		false	"Page size"				minimum(1)	maximum(100)	default(50)
-//	@Success		200		{object}	OpportunityPage
+//	@Param			changed_after	query		string	false	"List the opportunities changed after this batch instead"
-//	@Failure		400		{object}	httpx.Problem	"Invalid offset or limit"
+//	@Success		200				{object}	OpportunityPage
-//	@Failure		500		{object}	httpx.Problem
+//	@Failure		400				{object}	httpx.Problem	"Invalid offset or limit, or either with changed_after"
+//	@Failure		410				{object}	httpx.Problem	"Unknown batch"
+//	@Failure		500				{object}	httpx.Problem
 //	@Router			/opportunities [get]
 func (h *opportunityRoutes) list(w http.ResponseWriter, r *http.Request) {
 	query := r.URL.Query()
+	if query.Has("changed_after") {
+		h.listChanged(w, r)
+		return
+	}
+
 	offset, offsetErr := intParam(query.Get("offset"), 0, 0, 1<<30)
 	limit, limitErr := intParam(query.Get("limit"), 50, 1, 100)
 	err := errors.Join(offsetErr, limitErr)
@@ -89,6 +111,14 @@ func (h *opportunityRoutes) list(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// Read first, so nothing changed while the page is read is missed by
+	// asking for the changes after it.
+	batch, err := h.store.LatestChangeBatch(r.Context())
+	if err != nil {
+		httpx.WriteProblem(w, r, http.StatusInternalServerError, "")
+		return
+	}
+
 	// One more than the page, to tell whether another page follows.
 	opportunities, err := h.store.ListOpportunities(
 		r.Context(),
@@ -107,6 +137,7 @@ func (h *opportunityRoutes) list(w http.ResponseWriter, r *http.Request) {
 		Offset:        offset,
 		Limit:         limit,
 		HasMore:       len(opportunities) > limit,
+		Batch:         batch,
 	}
 	if page.HasMore {
 		page.Opportunities = opportunities[:limit]
@@ -115,6 +146,48 @@ func (h *opportunityRoutes) list(w http.ResponseWriter, r *http.Request) {
 	httpx.WriteJSON(w, http.StatusOK, page)
 }
 
+// listChanged serves GET /opportunities?changed_after=batch: every
+// opportunity changed after the change batch, as one page, with the newest
+// batch it includes. An unknown batch, such as one from another database, is
+// 410 Gone, as the client can't catch up from it.
+func (h *opportunityRoutes) listChanged(
+	w http.ResponseWriter, r *http.Request,
+) {
+	query := r.URL.Query()
+	if query.Has("offset") || query.Has("limit") {
+		httpx.WriteProblem(
+			w,
+			r,
+			http.StatusBadRequest,
+			"offset and limit can't be used with changed_after",
+		)
+		return
+	}
+
+	opportunities, batch, err := h.store.ListChangedOpportunities(
+		r.Context(), query.Get("changed_after"),
+	)
+	if errors.Is(err, store.ErrUnknownBatch) {
+		httpx.WriteProblem(
+			w,
+			r,
+			http.StatusGone,
+			"unknown batch: reload the whole list",
+		)
+		return
+	}
+	if err != nil {
+		httpx.WriteProblem(w, r, http.StatusInternalServerError, "")
+		return
+	}
+
+	httpx.WriteJSON(w, http.StatusOK, OpportunityPage{
+		Opportunities: opportunities,
+		Limit:         len(opportunities),
+		Batch:         batch,
+	})
+}
+
 // get serves GET /opportunities/{uuid}: one opportunity with its
 // classifications, marks and releases.
 //

internal/api/opportunities_test.go

@@ -172,6 +172,75 @@ func TestListOpportunitiesBadPaging(t *testing.T) {
 	}
 }
 
+// A page names the newest change batch, and changed_after lists what
+// changed after a batch: here, a change of marks.
+func TestListChangedOpportunities(t *testing.T) {
+	url := opportunitiesServer(t, 3)
+
+	var page OpportunityPage
+	request(t, "GET", url+"/opportunities", &page)
+	if page.Batch == "" {
+		t.Fatal("the page names no batch")
+	}
+
+	var changed OpportunityPage
+	status := request(
+		t, "GET", url+"/opportunities?changed_after="+page.Batch, &changed,
+	)
+	if status != 200 || len(changed.Opportunities) != 0 ||
+		changed.Opportunities == nil || changed.Batch != page.Batch {
+		t.Fatalf(
+			"nothing changed: status %d, %+v; want [] and the same batch",
+			status, changed,
+		)
+	}
+
+	request(t, "PUT", url+"/opportunities/ocds-2/marks/skipped", nil)
+	status = request(
+		t, "GET", url+"/opportunities?changed_after="+page.Batch, &changed,
+	)
+	if status != 200 || strings.Join(uuids(changed), ",") != "ocds-2" ||
+		!changed.Opportunities[0].Marks.Skipped {
+		t.Fatalf("after a mark: status %d, %+v; want ocds-2", status, changed)
+	}
+	if changed.Batch == page.Batch || changed.Limit != 1 || changed.HasMore {
+		t.Errorf(
+			"after a mark: batch %q, limit %d, has_more %v; "+
+				"want a newer batch, 1, false",
+			changed.Batch, changed.Limit, changed.HasMore,
+		)
+	}
+
+	var latest OpportunityPage
+	request(t, "GET", url+"/opportunities?limit=1", &latest)
+	if latest.Batch != changed.Batch {
+		t.Errorf("page batch %q; want %q", latest.Batch, changed.Batch)
+	}
+}
+
+// An unknown batch is 410 Gone, and paging can't be used with changed_after.
+func TestListChangedOpportunitiesRefused(t *testing.T) {
+	url := opportunitiesServer(t, 1)
+	var page OpportunityPage
+	request(t, "GET", url+"/opportunities", &page)
+
+	tests := []struct {
+		query string
+		want  int
+	}{
+		{"?changed_after=unknown", http.StatusGone},
+		{"?changed_after=", http.StatusGone},
+		{"?changed_after=" + page.Batch + "&limit=5", http.StatusBadRequest},
+		{"?offset=0&changed_after=" + page.Batch, http.StatusBadRequest},
+	}
+	for _, tt := range tests {
+		status := request(t, "GET", url+"/opportunities"+tt.query, nil)
+		if status != tt.want {
+			t.Errorf("%s: status %d; want %d", tt.query, status, tt.want)
+		}
+	}
+}
+
 // One opportunity comes with its releases; a missing one is 404.
 func TestGetOpportunity(t *testing.T) {
 	url := opportunitiesServer(t, 2)

internal/events/events.go

@@ -24,13 +24,27 @@ const (
 
 // ---------- OpportunitiesUpdated ----------
 
+// MaxListedOpportunities is the most opportunities an
+// OpportunitiesUpdatedPayload lists by UUID; more make it Large.
+const MaxListedOpportunities = 50
+
 // OpportunitiesUpdatedPayload lists the opportunities a save changed by UUID:
 // New were stored for the first time, and Updated were already stored and
 // gained at least one new release. Summary counts them.
+//
+// When there are more than MaxListedOpportunities, Large is true and New and
+// Updated are empty, to keep the event small; Summary still counts them.
+// Neither is ever null.
+//
+// Batch is the hash of the store's change batch recording them. A client
+// gets them, with any later changes, from GET /api/opportunities with
+// changed_after set to the last batch it has seen.
 type OpportunitiesUpdatedPayload struct {
 	New     []string             `json:"new"`
 	Updated []string             `json:"updated"`
+	Large   bool                 `json:"large"`
 	Summary OpportunitiesSummary `json:"summary"`
+	Batch   string               `json:"batch"`
 }
 
 // OpportunitiesSummary counts the opportunities in an

internal/events/events_test.go

@@ -15,12 +15,14 @@ func TestOpportunitiesUpdatedPayloadJSON(t *testing.T) {
 		Summary: OpportunitiesSummary{
 			New: 1,
 		},
+		Batch: "abc",
 	}
 	encoded, err := json.Marshal(payload)
 	if err != nil {
 		t.Fatal(err)
 	}
-	want := `{"new":["ocds-1"],"updated":[],"summary":{"new":1,"updated":0}}`
+	want := `{"new":["ocds-1"],"updated":[],"large":false,` +
+		`"summary":{"new":1,"updated":0},"batch":"abc"}`
 	if string(encoded) != want {
 		t.Errorf("JSON = %s; want %s", encoded, want)
 	}

internal/fetcher/fetcher.go

@@ -238,7 +238,7 @@ func (f *Fetcher) save(
 	)
 
 	f.publishClassificationChanges(result)
-	f.publishUpdates(opportunities, result.OpportunityResults)
+	f.publishUpdates(opportunities, result)
 
 	return nil
 }
@@ -277,12 +277,14 @@ func (f *Fetcher) publishClassificationChanges(
 }
 
 // publishUpdates publishes an events.OpportunitiesUpdated event for the
-// opportunities that results, from saving them, show were new or gained new
+// opportunities that result, from saving them, shows were new or gained new
 // releases, if there were any and a bus was given WithEvents. An opportunity
 // whose row was only rewritten, such as one fetched again from the newest
-// stored release, isn't counted as updated.
+// stored release, isn't counted as updated. Over
+// events.MaxListedOpportunities, they are only counted (see
+// events.OpportunitiesUpdatedPayload).
 func (f *Fetcher) publishUpdates(
-	opportunities []*types.Opportunity, results []store.SaveResult,
+	opportunities []*types.Opportunity, result store.SaveMultipleResult,
 ) {
 	if f.events == nil {
 		return
@@ -291,12 +293,13 @@ func (f *Fetcher) publishUpdates(
 	payload := events.OpportunitiesUpdatedPayload{
 		New:     []string{},
 		Updated: []string{},
+		Batch:   result.ChangeBatch,
 	}
-	for i, result := range results {
+	for i, saved := range result.OpportunityResults {
 		switch {
-		case result.NewOpportunity:
+		case saved.NewOpportunity:
 			payload.New = append(payload.New, opportunities[i].UUID)
-		case result.NewReleases > 0:
+		case saved.NewReleases > 0:
 			payload.Updated = append(payload.Updated, opportunities[i].UUID)
 		}
 	}
@@ -307,6 +310,11 @@ func (f *Fetcher) publishUpdates(
 	if payload.Summary.New == 0 && payload.Summary.Updated == 0 {
 		return
 	}
+	if len(payload.New)+len(payload.Updated) > events.MaxListedOpportunities {
+		payload.New = []string{}
+		payload.Updated = []string{}
+		payload.Large = true
+	}
 
 	f.events.Publish(events.Event{
 		Type:    events.OpportunitiesUpdated,

internal/fetcher/fetcher_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
@@ -708,6 +709,18 @@ func TestSavePublishesUpdates(t *testing.T) {
 	save(onePackage) // fetched again: nothing published
 	save(newRelease) // a new release of a stored opportunity
 
+	// Each event names its own batch; the hashes are random.
+	if len(published) == 2 {
+		if published[0].Batch == "" ||
+			published[0].Batch == published[1].Batch {
+			t.Errorf(
+				"batches %q and %q; want two different ones",
+				published[0].Batch, published[1].Batch,
+			)
+		}
+		published[0].Batch, published[1].Batch = "", ""
+	}
+
 	want := []events.OpportunitiesUpdatedPayload{
 		{
 			New:     []string{"ocds-1"},
@@ -729,6 +742,61 @@ func TestSavePublishesUpdates(t *testing.T) {
 	}
 }
 
+// More than events.MaxListedOpportunities are only counted.
+func TestPublishUpdatesLarge(t *testing.T) {
+	var bus events.Bus
+	var published []events.OpportunitiesUpdatedPayload
+	bus.Subscribe(func(event events.Event) {
+		payload := event.Payload.(events.OpportunitiesUpdatedPayload)
+		published = append(published, payload)
+	})
+	f := New(&config.JSCSources{}, WithEvents(&bus))
+
+	publish := func(count int) {
+		t.Helper()
+		var opportunities []*types.Opportunity
+		result := store.SaveMultipleResult{
+			ChangeBatch: "batch",
+		}
+		for i := range count {
+			opportunities = append(opportunities, &types.Opportunity{
+				UUID: fmt.Sprintf("ocds-%d", i),
+			})
+			result.OpportunityResults = append(
+				result.OpportunityResults,
+				store.SaveResult{
+					NewOpportunity: i%2 == 0,
+					NewReleases:    1,
+				},
+			)
+		}
+		f.publishUpdates(opportunities, result)
+	}
+
+	publish(events.MaxListedOpportunities)
+	publish(events.MaxListedOpportunities + 1)
+	if len(published) != 2 {
+		t.Fatalf("published %d events; want 2", len(published))
+	}
+
+	listed := published[0]
+	if listed.Large ||
+		len(listed.New)+len(listed.Updated) != events.MaxListedOpportunities {
+		t.Errorf("at the limit: %+v; want every UUID listed", listed)
+	}
+
+	large := published[1]
+	want := events.OpportunitiesSummary{
+		New:     26,
+		Updated: 25,
+	}
+	if !large.Large || len(large.New) != 0 || len(large.Updated) != 0 ||
+		large.New == nil || large.Updated == nil ||
+		large.Summary != want || large.Batch != "batch" {
+		t.Errorf("over the limit: %+v; want only counted", large)
+	}
+}
+
 // Without WithEvents, a save publishes nothing and still succeeds.
 func TestSaveWithoutEvents(t *testing.T) {
 	f := New(&config.JSCSources{}, WithStore(openTestStore(t)))

internal/store/README.md

@@ -39,6 +39,8 @@ internal/store/common.go           common for migrate, store and direct access
 | `classification_codes` | `id` | Shared between opportunities. A row is reused only when every content column matches. The table grows over time to include codes not already seen in fetched opportunities. |
 | `opportunity_classifications` | (`opportunity_uuid`, `classification_id`) | Mapping table; replaced as a set when the opportunity is updated. |
 | `opportunities_fts` | rowid = `opportunities.id` | FTS5 index on `opportunities.title`, kept in sync by triggers. |
+| `opportunity_change_batches` | `id`; `hash` (unique) | Write-once. One row per save that adds opportunities or gives them new releases, and per change of marks. `hash` names the batch to clients (random, so unguessable); `created_at` orders the batches. |
+| `opportunity_change_batch_opportunities` | (`batch_id`, `opportunity_uuid`) | The opportunities in each batch. References both tables, cascading on delete. |
 
 Foreign keys are enforced (`_foreign_keys=on` in the connection string) and cascade on delete from `opportunities`. The connection also enables WAL mode, a 5s busy timeout, and `_txlock=immediate`, so a save takes the write lock up front rather than failing halfway through.
 
@@ -88,7 +90,7 @@ A soft convention for the SQL in `queries.go`, followed so the queries look alik
 - A column list that doesn't fit on one line goes inside its brackets, indented one tab and filled to about 80 characters per line. The `VALUES` list wraps at the same columns, so each line of values matches the line of columns above it.
 - `ON CONFLICT … DO UPDATE SET` puts one `column = excluded.column` per line, indented one tab.
 - A subquery (`EXISTS (…)`, `IN (…)`) is indented one tab inside its brackets, with the closing bracket at the start of a line.
-- Tables are aliased consistently: `t` for `opportunities`, `c` for `classification_codes`, `tc` for `opportunity_classifications`, `m` for `opportunity_marks`.
+- Tables are aliased consistently: `t` for `opportunities`, `c` for `classification_codes`, `tc` for `opportunity_classifications`, `m` for `opportunity_marks`, `b` for `opportunity_change_batches`, `bo` for `opportunity_change_batch_opportunities`.
 
 ## Store API
 
@@ -106,6 +108,8 @@ ts, err := s.ListOpportunities(ctx, store.OpportunityFilter{
 	Limit:      20,
 })
 rs, err := s.ListReleases(ctx, uuid) // oldest first, with tags
+batch, err := s.LatestChangeBatch(ctx) // hash of the newest change batch, or ""
+ts, batch, err = s.ListChangedOpportunities(ctx, batch) // changed since; ErrUnknownBatch
 ok, err := s.ReleaseExists(ctx, uuid, releaseID)
 ```
 
@@ -122,5 +126,8 @@ ok, err := s.ReleaseExists(ctx, uuid, releaseID)
 **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.
+
+
 **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

@@ -0,0 +1,55 @@
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.avsdev.uk/Go/job-search/internal/types"
+)
+
+// A batch recorded after the clock went back is still created after the
+// newest batch, so asking for the batches after that one finds it.
+func TestChangeBatchAfterClockWentBack(t *testing.T) {
+	db := openTestDB(t)
+	s := &Store{db: db}
+	ctx := context.Background()
+
+	_, err := s.SaveOpportunity(
+		ctx,
+		&types.Opportunity{
+			UUID:              "ocds-1",
+			LatestReleaseDate: time.Now(),
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	future := time.Now().UTC().Add(time.Hour)
+	db.MustExec(insertChangeBatch, "future", future)
+
+	err = s.SetMarks(ctx, types.OpportunityMarks{
+		UUID: "ocds-1",
+		Won:  true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var latest changeBatch
+	if err := db.Get(&latest, selectLatestChangeBatch); err != nil {
+		t.Fatal(err)
+	}
+	if !latest.CreatedAt.After(future) {
+		t.Errorf("batch created at %v, not after %v", latest.CreatedAt, future)
+	}
+
+	got, hash, err := s.ListChangedOpportunities(ctx, "future")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 || got[0].UUID != "ocds-1" || hash != latest.Hash {
+		t.Errorf("changed after future: %v, %q", got, hash)
+	}
+}

internal/store/migrations/00008_create_opportunity_change_batches.sql

@@ -0,0 +1,33 @@
+-- +goose Up
+-- +goose StatementBegin
+-- A batch of changed opportunities: those one save stored for the first time
+-- or gave new releases, or the one whose marks were changed. A client asks
+-- for what changed after a batch it has seen by the batch's hash.
+CREATE TABLE IF NOT EXISTS "opportunity_change_batches" (
+	"id" INTEGER PRIMARY KEY,
+	-- A random hash naming the batch to clients, so a batch can't be guessed
+	-- or worked out from another.
+	"hash" TEXT NOT NULL UNIQUE,
+	-- When the batch was recorded, in UTC. Each batch's is later than every
+	-- earlier batch's, so the batches after one are those created after it.
+	"created_at" DATETIME NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS "opportunity_change_batches_created_at"
+	ON "opportunity_change_batches" ("created_at");
+
+-- The opportunities in each batch.
+CREATE TABLE IF NOT EXISTS "opportunity_change_batch_opportunities" (
+	"batch_id" INTEGER NOT NULL
+		REFERENCES "opportunity_change_batches" ("id") ON DELETE CASCADE,
+	"opportunity_uuid" TEXT NOT NULL
+		REFERENCES "opportunities" ("uuid") ON DELETE CASCADE,
+	PRIMARY KEY ("batch_id", "opportunity_uuid")
+) WITHOUT ROWID;
+-- +goose StatementEnd
+
+-- +goose Down
+-- +goose StatementBegin
+DROP TABLE IF EXISTS "opportunity_change_batch_opportunities";
+DROP TABLE IF EXISTS "opportunity_change_batches";
+-- +goose StatementEnd

internal/store/queries.go

@@ -277,3 +277,45 @@ SELECT *
 FROM opportunity_marks
 WHERE uuid IN (?)
 ORDER BY uuid`
+
+// ---------- opportunity_change_batches ----------
+
+// insertChangeBatch records a batch, returning its id. The created_at
+// argument must be later than every stored batch's (see recordChangeBatch).
+const insertChangeBatch = `
+INSERT INTO opportunity_change_batches (hash, created_at)
+VALUES (?, ?)
+RETURNING id`
+
+// selectLatestChangeBatch reads the newest batch into a changeBatch.
+const selectLatestChangeBatch = `
+SELECT *
+FROM opportunity_change_batches
+ORDER BY created_at DESC
+LIMIT 1`
+
+// selectChangeBatchByHash reads the batch with a hash into a changeBatch.
+const selectChangeBatchByHash = `
+SELECT * FROM opportunity_change_batches WHERE hash = ?`
+
+// ---------- opportunity_change_batch_opportunities ----------
+
+// insertChangeBatchOpportunity adds an opportunity to a batch.
+const insertChangeBatchOpportunity = `
+INSERT INTO opportunity_change_batch_opportunities (batch_id, opportunity_uuid)
+VALUES (?, ?)
+ON CONFLICT DO NOTHING`
+
+// selectOpportunitiesChangedBetween reads, into opportunityRows, the
+// opportunities in the batches created after the first argument and at or
+// before the second, once each, ordered as ListOpportunities orders them.
+const selectOpportunitiesChangedBetween = `
+SELECT t.*
+FROM opportunities t
+WHERE t.uuid IN (
+	SELECT bo.opportunity_uuid
+	FROM opportunity_change_batch_opportunities bo
+	JOIN opportunity_change_batches b ON b.id = bo.batch_id
+	WHERE b.created_at > ? AND b.created_at <= ?
+)
+ORDER BY t.latest_release_date DESC, t.uuid`

internal/store/queries_test.go

@@ -112,6 +112,17 @@ var checks = map[string]queryCheck{
 	"insertOpportunityMarks":      {sql: insertOpportunityMarks, named: true},
 	"upsertOpportunityMarks":      {sql: upsertOpportunityMarks, named: true},
 	"selectMarksForOpportunities": {sql: selectMarksForOpportunities, in: true},
+
+	// opportunity_change_batches
+	"insertChangeBatch":       {sql: insertChangeBatch},
+	"selectLatestChangeBatch": {sql: selectLatestChangeBatch},
+	"selectChangeBatchByHash": {sql: selectChangeBatchByHash},
+
+	// opportunity_change_batch_opportunities
+	"insertChangeBatchOpportunity": {sql: insertChangeBatchOpportunity},
+	"selectOpportunitiesChangedBetween": {
+		sql: selectOpportunitiesChangedBetween,
+	},
 }
 
 func withFilter(fragment string) string {

internal/store/store.go

@@ -5,7 +5,10 @@ package store
 
 import (
 	"context"
+	"crypto/rand"
+	"crypto/sha256"
 	"database/sql"
+	"encoding/hex"
 	"errors"
 	"fmt"
 	"strings"
@@ -22,6 +25,9 @@ import (
 var (
 	// ErrNotFound is returned when a requested opportunity does not exist.
 	ErrNotFound = errors.New("store: not found")
+	// ErrUnknownBatch is returned by ListChangedOpportunities for a hash
+	// that names no change batch.
+	ErrUnknownBatch = errors.New("store: unknown change batch")
 	// ErrNeedsMigration is returned by Open when the database has pending
 	// migrations. Apply them with the migrate command, or pass
 	// WithAutoMigrate.
@@ -69,6 +75,10 @@ type SaveMultipleResult struct {
 	// a stored opportunity had before the save and none has after it, in
 	// code order; nil if there are none.
 	RemovedClassificationIDs []int64
+	// ChangeBatch is the hash of the change batch recording the
+	// opportunities the save stored for the first time or gave new releases
+	// (see ListChangedOpportunities); "" if there are none.
+	ChangeBatch string
 }
 
 // SaveResult reports what SaveOpportunity changed.
@@ -254,6 +264,15 @@ type opportunityRow struct {
 	types.Opportunity
 }
 
+// changeBatch is a row of opportunity_change_batches: a batch of changed
+// opportunities, named to clients by its Hash. ID and CreatedAt are internal
+// to the store.
+type changeBatch struct {
+	ID        int64     `db:"id"`
+	Hash      string    `db:"hash"`
+	CreatedAt time.Time `db:"created_at"`
+}
+
 // SaveOpportunity stores opportunity and its releases in one transaction.
 //
 // Releases (and their tags) are write-once: ones already stored are skipped.
@@ -265,6 +284,9 @@ type opportunityRow struct {
 // opportunity.Marks are stored only for a new opportunity; existing marks are
 // kept (use SetMarks to change them).
 //
+// A new opportunity, or one given new releases, is recorded in a change
+// batch of its own (see ListChangedOpportunities).
+//
 // opportunity is normalised in place (see prepareOpportunity), and when it is
 // written its classifications' IDs are filled in.
 func (s *Store) SaveOpportunity(
@@ -285,6 +307,15 @@ func (s *Store) SaveOpportunity(
 		return SaveResult{}, err
 	}
 
+	_, err = recordChangeBatch(
+		ctx,
+		tx,
+		changedUUIDs([]*types.Opportunity{opportunity}, []SaveResult{result}),
+	)
+	if err != nil {
+		return SaveResult{}, err
+	}
+
 	if err := tx.Commit(); err != nil {
 		return SaveResult{}, err
 	}
@@ -307,7 +338,9 @@ func (s *Store) SaveOpportunity(
 // Each opportunity is normalised in place (see prepareOpportunity), and when
 // it is written its classifications' IDs are filled in. The result reports
 // the classification codes the save brought into or out of use (see
-// ListClassifications).
+// ListClassifications), and the hash of the change batch recording the
+// opportunities that were new or given new releases (see
+// ListChangedOpportunities).
 func (s *Store) SaveOpportunities(
 	ctx context.Context, opportunities []*types.Opportunity,
 ) (SaveMultipleResult, error) {
@@ -354,6 +387,13 @@ func (s *Store) SaveOpportunities(
 	total.AddedClassifications = added
 	total.RemovedClassificationIDs = removed
 
+	total.ChangeBatch, err = recordChangeBatch(
+		ctx, tx, changedUUIDs(opportunities, total.OpportunityResults),
+	)
+	if err != nil {
+		return SaveMultipleResult{}, err
+	}
+
 	if err := tx.Commit(); err != nil {
 		return SaveMultipleResult{}, err
 	}
@@ -406,6 +446,82 @@ func classificationChanges(
 	return added, removed
 }
 
+// changedUUIDs returns the UUIDs of the opportunities that results, from
+// saving them, show were new or given new releases. An opportunity whose row
+// was only rewritten, such as one fetched again from its newest stored
+// release, hasn't changed.
+func changedUUIDs(
+	opportunities []*types.Opportunity, results []SaveResult,
+) []string {
+	var uuids []string
+	for i, result := range results {
+		if result.NewOpportunity || result.NewReleases > 0 {
+			uuids = append(uuids, opportunities[i].UUID)
+		}
+	}
+
+	return uuids
+}
+
+// recordChangeBatch records a change batch of the opportunities uuids inside
+// tx and returns its hash, or "" without recording anything if uuids is
+// empty.
+//
+// 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(
+	ctx context.Context, tx *sqlx.Tx, uuids []string,
+) (string, error) {
+	if len(uuids) == 0 {
+		return "", nil
+	}
+
+	var latest changeBatch
+	err := tx.GetContext(ctx, &latest, selectLatestChangeBatch)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		return "", fmt.Errorf("reading the latest change batch: %w", err)
+	}
+
+	createdAt := time.Now().UTC()
+	if !createdAt.After(latest.CreatedAt) {
+		createdAt = latest.CreatedAt.Add(time.Nanosecond)
+	}
+	hash := newBatchHash(createdAt)
+
+	var batchID int64
+	err = tx.GetContext(ctx, &batchID, insertChangeBatch, hash, createdAt)
+	if err != nil {
+		return "", fmt.Errorf("recording a change batch: %w", err)
+	}
+
+	for _, uuid := range uuids {
+		_, err = tx.ExecContext(ctx, insertChangeBatchOpportunity, batchID, uuid)
+		if err != nil {
+			return "", fmt.Errorf(
+				"recording %s in a change batch: %w", uuid, err,
+			)
+		}
+	}
+
+	return hash, nil
+}
+
+// newBatchHash returns the hash of a change batch created at createdAt: the
+// hex SHA-256 hash of random bytes and the time. The random bytes make it
+// unguessable, so a client can only name a batch it was given.
+func newBatchHash(createdAt time.Time) string {
+	random := make([]byte, 32)
+	rand.Read(random) // crypto/rand.Read never returns an error
+
+	hash := sha256.Sum256(
+		append(random, createdAt.Format(time.RFC3339Nano)...),
+	)
+	return hex.EncodeToString(hash[:])
+}
+
 // saveOpportunity does the work of SaveOpportunity and SaveOpportunities for
 // one prepared opportunity inside tx. isNew reports that the opportunity was
 // not already stored.
@@ -539,7 +655,8 @@ func writeOpportunity(
 
 // SetMarks replaces the marks of the opportunity marks.UUID, or returns
 // ErrNotFound. Marks that fail marks.Validate (e.g. types.ErrWonAndLost) are
-// not stored.
+// not stored. The opportunity is recorded in a change batch of its own (see
+// ListChangedOpportunities), so a client catching up sees its new marks.
 func (s *Store) SetMarks(
 	ctx context.Context, marks types.OpportunityMarks,
 ) error {
@@ -550,11 +667,26 @@ func (s *Store) SetMarks(
 		return fmt.Errorf("opportunity %s: %w", marks.UUID, err)
 	}
 
-	_, err := s.db.NamedExecContext(ctx, upsertOpportunityMarks, marks)
+	tx, err := s.db.BeginTxx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback() //nolint:errcheck // no-op after Commit
+
+	_, err = tx.NamedExecContext(ctx, upsertOpportunityMarks, marks)
 	if isForeignKeyError(err) {
 		return ErrNotFound
 	}
-	return err
+	if err != nil {
+		return err
+	}
+
+	_, err = recordChangeBatch(ctx, tx, []string{marks.UUID})
+	if err != nil {
+		return err
+	}
+
+	return tx.Commit()
 }
 
 // isForeignKeyError reports whether err is SQLite rejecting a write that
@@ -606,6 +738,76 @@ func (s *Store) ListOpportunities(
 		return nil, err
 	}
 
+	return s.opportunitiesFromRows(ctx, rows)
+}
+
+// LatestChangeBatch returns the hash of the newest change batch, or "" if
+// none has been recorded. Read before listing opportunities, it is the batch
+// to give ListChangedOpportunities later to get what changed since.
+func (s *Store) LatestChangeBatch(ctx context.Context) (string, error) {
+	var latest changeBatch
+	err := s.db.GetContext(ctx, &latest, selectLatestChangeBatch)
+	if errors.Is(err, sql.ErrNoRows) {
+		return "", nil
+	}
+	if err != nil {
+		return "", fmt.Errorf("reading the latest change batch: %w", err)
+	}
+
+	return latest.Hash, nil
+}
+
+// ListChangedOpportunities returns the opportunities recorded in the change
+// batches created after the batch with the hash after, once each, as
+// ListOpportunities returns them but unpaged, and the hash of the newest of
+// those batches, to give it next time. With no newer batches, it returns no
+// opportunities and after. An unknown hash is ErrUnknownBatch.
+//
+// Batches recorded while it runs are left for next time, so none is missed:
+// only those up to the newest batch when it starts are read.
+func (s *Store) ListChangedOpportunities(
+	ctx context.Context, after string,
+) ([]types.Opportunity, string, error) {
+	var since changeBatch
+	err := s.db.GetContext(ctx, &since, selectChangeBatchByHash, after)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, "", ErrUnknownBatch
+	}
+	if err != nil {
+		return nil, "", fmt.Errorf("reading change batch %s: %w", after, err)
+	}
+
+	var latest changeBatch
+	err = s.db.GetContext(ctx, &latest, selectLatestChangeBatch)
+	if err != nil {
+		return nil, "", fmt.Errorf("reading the latest change batch: %w", err)
+	}
+
+	var rows []opportunityRow
+	err = s.db.SelectContext(
+		ctx,
+		&rows,
+		selectOpportunitiesChangedBetween,
+		since.CreatedAt,
+		latest.CreatedAt,
+	)
+	if err != nil {
+		return nil, "", fmt.Errorf("listing changed opportunities: %w", err)
+	}
+
+	opportunities, err := s.opportunitiesFromRows(ctx, rows)
+	if err != nil {
+		return nil, "", err
+	}
+
+	return opportunities, latest.Hash, nil
+}
+
+// opportunitiesFromRows returns the opportunities in rows, in order, with
+// their classifications and marks. The result is never nil.
+func (s *Store) opportunitiesFromRows(
+	ctx context.Context, rows []opportunityRow,
+) ([]types.Opportunity, error) {
 	opportunities := make([]types.Opportunity, len(rows))
 	for i, row := range rows {
 		opportunities[i] = row.Opportunity

internal/store/store_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"path/filepath"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -121,15 +122,7 @@ func uuids(ts []types.Opportunity) []string {
 }
 
 func equal(a, b []string) bool {
-	if len(a) != len(b) {
+	return slices.Equal(a, b)
-		return false
-	}
-	for i := range a {
-		if a[i] != b[i] {
-			return false
-		}
-	}
-	return true
 }
 
 func TestSaveAndGetRoundTrip(t *testing.T) {
@@ -1010,3 +1003,96 @@ func TestSaveOpportunitiesClassificationChanges(t *testing.T) {
 		)
 	}
 }
+
+// A save records the opportunities it adds or gives new releases in a change
+// batch, as does a change of marks, and ListChangedOpportunities returns
+// those in the batches after a given one.
+func TestChangeBatches(t *testing.T) {
+	s := openStore(t)
+	ctx := context.Background()
+	save := func(opportunities ...*types.Opportunity) string {
+		t.Helper()
+		total, err := s.SaveOpportunities(ctx, opportunities)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return total.ChangeBatch
+	}
+	changed := func(after string) ([]types.Opportunity, string) {
+		t.Helper()
+		opportunities, latest, err := s.ListChangedOpportunities(ctx, after)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return opportunities, latest
+	}
+	latest := func() string {
+		t.Helper()
+		hash, err := s.LatestChangeBatch(ctx)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return hash
+	}
+
+	if hash := latest(); hash != "" {
+		t.Fatalf("empty store: latest batch %q, want none", hash)
+	}
+
+	first := save(
+		opportunity("ocds-1", "One", 1, "r1"),
+		opportunity("ocds-2", "Two", 1, "r1"),
+	)
+	if len(first) != 64 || strings.Trim(first, "0123456789abcdef") != "" {
+		t.Fatalf("batch %q is not a hex SHA-256 hash", first)
+	}
+	if hash := latest(); hash != first {
+		t.Errorf("latest batch %q, want %q", hash, first)
+	}
+
+	// Nothing new: no batch.
+	if hash := save(opportunity("ocds-1", "One", 1, "r1")); hash != "" {
+		t.Errorf("save with nothing new recorded batch %q", hash)
+	}
+
+	second := save(
+		opportunity("ocds-2", "Two", 2, "r1", "r2"),
+		opportunity("ocds-3", "Three", 3, "r1"),
+	)
+	if second == "" || second == first {
+		t.Fatalf("second batch %q, first %q", second, first)
+	}
+
+	got, hash := changed(first)
+	if want := []string{"ocds-3", "ocds-2"}; !equal(uuids(got), want) {
+		t.Errorf("changed after first: %v, want %v", uuids(got), want)
+	}
+	if hash != second {
+		t.Errorf("changed after first: latest %q, want %q", hash, second)
+	}
+	if len(got) == 2 && len(got[1].Classifications) != 2 {
+		t.Errorf("changed opportunity without classifications: %+v", got[1])
+	}
+
+	mustSetMarks(t, s, types.OpportunityMarks{
+		UUID:    "ocds-1",
+		Skipped: true,
+	})
+	got, hash = changed(second)
+	if !equal(uuids(got), []string{"ocds-1"}) || !got[0].Marks.Skipped {
+		t.Errorf("changed after marks: %+v, want ocds-1 skipped", got)
+	}
+	if hash == second || hash != latest() {
+		t.Errorf("changed after marks: latest %q, want a newer batch", hash)
+	}
+
+	got, again := changed(hash)
+	if got == nil || len(got) != 0 || again != hash {
+		t.Errorf("changed after latest: %v, %q; want [], %q", got, again, hash)
+	}
+
+	_, _, err := s.ListChangedOpportunities(ctx, "unknown")
+	if !errors.Is(err, store.ErrUnknownBatch) {
+		t.Errorf("unknown batch: err %v, want ErrUnknownBatch", err)
+	}
+}