Added a ClassificationsUpdated event

SaveOpportunities now reports the classification codes a save brings into
use (AddedClassifications, in full) and takes out of use
(RemovedClassificationIDs), by comparing the codes in use before and after
the save inside its transaction. The fetcher publishes them as a
ClassificationsUpdated event, {"added":[codes],"removed":[ids]}, never null,
before the save's OpportunitiesUpdated, so clients can keep their list of
classifications without asking the API again.
This commit is contained in:
bwilliams committed 2026-10-16 15:24:37 +01:00
1 parent aa06b94870
commit 51d6470649
6 files changed
+287 -3

No files matched your search

+17
View File
@@ -4,6 +4,8 @@
// and their payloads. // and their payloads.
package events package events
import "git.avsdev.uk/Go/job-search/internal/types"
// Event types: an Event's Type, which says what its Payload is. // Event types: an Event's Type, which says what its Payload is.
const ( const (
// OpportunitiesUpdated is published when a save stores new // OpportunitiesUpdated is published when a save stores new
@@ -14,6 +16,10 @@ const (
// changed over the API. Its payload is the opportunity's // changed over the API. Its payload is the opportunity's
// types.OpportunityMarks after the change, which include its UUID. // types.OpportunityMarks after the change, which include its UUID.
OpportunityMarksUpdated = "OpportunityMarksUpdated" OpportunityMarksUpdated = "OpportunityMarksUpdated"
// ClassificationsUpdated is published when a save brings classification
// codes into use (some opportunity has them) or takes them out of use.
// Its payload is a ClassificationsUpdatedPayload.
ClassificationsUpdated = "ClassificationsUpdated"
) )
// ---------- OpportunitiesUpdated ---------- // ---------- OpportunitiesUpdated ----------
@@ -33,3 +39,14 @@ type OpportunitiesSummary struct {
New int `json:"new"` New int `json:"new"`
Updated int `json:"updated"` Updated int `json:"updated"`
} }
// ---------- ClassificationsUpdated ----------
// ClassificationsUpdatedPayload is how the classification codes in use
// changed, so a client's list of them can be kept up to date without asking
// for it again: Added are the codes now in use, in full, and Removed the ids
// of those no longer in use. Neither is ever null.
type ClassificationsUpdatedPayload struct {
Added []types.ClassificationCode `json:"added"`
Removed []int64 `json:"removed"`
}
+26
View File
@@ -3,6 +3,8 @@ package events
import ( import (
"encoding/json" "encoding/json"
"testing" "testing"
"git.avsdev.uk/Go/job-search/internal/types"
) )
// The payload marshals with the keys clients read, and empty lists as []. // The payload marshals with the keys clients read, and empty lists as [].
@@ -23,3 +25,27 @@ func TestOpportunitiesUpdatedPayloadJSON(t *testing.T) {
t.Errorf("JSON = %s; want %s", encoded, want) t.Errorf("JSON = %s; want %s", encoded, want)
} }
} }
// The classifications payload marshals with the keys clients read.
func TestClassificationsUpdatedPayloadJSON(t *testing.T) {
payload := ClassificationsUpdatedPayload{
Added: []types.ClassificationCode{
{
ID: 7,
IsCPV: true,
Code: "72000000",
Classification: "IT services",
},
},
Removed: []int64{3},
}
encoded, err := json.Marshal(payload)
if err != nil {
t.Fatal(err)
}
want := `{"added":[{"id":7,"is_cpv":true,"code":"72000000",` +
`"classification":"IT services"}],"removed":[3]}`
if string(encoded) != want {
t.Errorf("JSON = %s; want %s", encoded, want)
}
}
+38 -2
View File
@@ -39,8 +39,10 @@ func WithStore(opportunityStore *store.Store) Option {
} }
// WithEvents makes the workers publish an events.OpportunitiesUpdated event // WithEvents makes the workers publish an events.OpportunitiesUpdated event
// to bus whenever a save stores new opportunities or new releases. It needs // to bus whenever a save stores new opportunities or new releases, and an
// WithStore: without a store nothing is saved, so nothing is published. // events.ClassificationsUpdated event when it changes the classification
// codes in use. It needs WithStore: without a store nothing is saved, so
// nothing is published.
func WithEvents(bus *events.Bus) Option { func WithEvents(bus *events.Bus) Option {
return func(f *Fetcher) { return func(f *Fetcher) {
f.events = bus f.events = bus
@@ -235,11 +237,45 @@ func (f *Fetcher) save(
"new_releases", result.NewReleases, "new_releases", result.NewReleases,
) )
f.publishClassificationChanges(result)
f.publishUpdates(opportunities, result.OpportunityResults) f.publishUpdates(opportunities, result.OpportunityResults)
return nil return nil
} }
// publishClassificationChanges publishes an events.ClassificationsUpdated
// event for the classification codes a save, with result, brought into or
// out of use, if it changed any and a bus was given WithEvents. It's
// published before the save's events.OpportunitiesUpdated, so a client
// knows a new code before an opportunity that has it.
func (f *Fetcher) publishClassificationChanges(
result store.SaveMultipleResult,
) {
if f.events == nil {
return
}
if len(result.AddedClassifications) == 0 &&
len(result.RemovedClassificationIDs) == 0 {
return
}
payload := events.ClassificationsUpdatedPayload{
Added: result.AddedClassifications,
Removed: result.RemovedClassificationIDs,
}
if payload.Added == nil {
payload.Added = []types.ClassificationCode{}
}
if payload.Removed == nil {
payload.Removed = []int64{}
}
f.events.Publish(events.Event{
Type: events.ClassificationsUpdated,
Payload: payload,
})
}
// publishUpdates publishes an events.OpportunitiesUpdated event for the // publishUpdates publishes an events.OpportunitiesUpdated event for the
// opportunities that results, from saving them, show were new or gained new // opportunities that results, from saving them, show were new or gained new
// releases, if there were any and a bus was given WithEvents. An opportunity // releases, if there were any and a bus was given WithEvents. An opportunity
+71
View File
@@ -737,3 +737,74 @@ func TestSaveWithoutEvents(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
} }
// withClassification returns onePackage with its release changed to id and
// date, and its tender classified by code.
func withClassification(id, date, code string) string {
return strings.NewReplacer(
`"id": "r1"`, `"id": "`+id+`"`,
"2026-10-01T12:00:00Z", date,
`"tender": {`,
`"tender": {"classification": {"scheme": "CPV", "id": "`+code+
`", "description": "Code `+code+`"}, `,
).Replace(onePackage)
}
// A save publishes the classification codes it brings into use, in full, and
// the ids of those it takes out of use, before OpportunitiesUpdated; nothing
// when the codes in use don't change.
func TestSavePublishesClassificationChanges(t *testing.T) {
var bus events.Bus
var published []events.Event
bus.Subscribe(func(event events.Event) {
published = append(published, event)
})
f := New(
&config.JSCSources{},
WithStore(openTestStore(t)),
WithEvents(&bus),
)
save := func(body string) []events.ClassificationsUpdatedPayload {
t.Helper()
published = nil
err := f.save(context.Background(), slog.Default(), fetched(t, body))
if err != nil {
t.Fatal(err)
}
var changes []events.ClassificationsUpdatedPayload
for i, event := range published {
if event.Type != events.ClassificationsUpdated {
continue
}
if i != 0 {
t.Errorf("ClassificationsUpdated was event %d; want it first", i)
}
changes = append(
changes, event.Payload.(events.ClassificationsUpdatedPayload),
)
}
return changes
}
changes := save(withClassification("r1", "2026-10-01T12:00:00Z", "72000000"))
if len(changes) != 1 || len(changes[0].Added) != 1 ||
changes[0].Added[0].Code != "72000000" || len(changes[0].Removed) != 0 {
t.Fatalf("first save published %+v; want 72000000 added", changes)
}
oldID := changes[0].Added[0].ID
changes = save(withClassification("r1", "2026-10-01T12:00:00Z", "72000000"))
if len(changes) != 0 {
t.Errorf("saving again published %+v; want nothing", changes)
}
changes = save(withClassification("r2", "2026-10-02T12:00:00Z", "03000000"))
if len(changes) != 1 || len(changes[0].Added) != 1 ||
changes[0].Added[0].Code != "03000000" ||
len(changes[0].Removed) != 1 || changes[0].Removed[0] != oldID {
t.Errorf(
"replacing the code published %+v; want 03000000 added, %d removed",
changes, oldID,
)
}
}
+69 -1
View File
@@ -61,6 +61,14 @@ type SaveMultipleResult struct {
UpdatedOpportunities int UpdatedOpportunities int
// OpportunityResults is a slice of the results of each opportunity saved. // OpportunityResults is a slice of the results of each opportunity saved.
OpportunityResults []SaveResult OpportunityResults []SaveResult
// AddedClassifications are the classification codes that no stored
// opportunity had before the save and at least one has after it, ordered
// by code; nil if there are none.
AddedClassifications []types.ClassificationCode
// RemovedClassificationIDs are the ids of the classification codes that
// a stored opportunity had before the save and none has after it, in
// code order; nil if there are none.
RemovedClassificationIDs []int64
} }
// SaveResult reports what SaveOpportunity changed. // SaveResult reports what SaveOpportunity changed.
@@ -297,7 +305,9 @@ func (s *Store) SaveOpportunity(
// SetMarks to change them). // SetMarks to change them).
// //
// Each opportunity is normalised in place (see prepareOpportunity), and when // Each opportunity is normalised in place (see prepareOpportunity), and when
// it is written its classifications' IDs are filled in. // 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).
func (s *Store) SaveOpportunities( func (s *Store) SaveOpportunities(
ctx context.Context, opportunities []*types.Opportunity, ctx context.Context, opportunities []*types.Opportunity,
) (SaveMultipleResult, error) { ) (SaveMultipleResult, error) {
@@ -313,6 +323,11 @@ func (s *Store) SaveOpportunities(
} }
defer tx.Rollback() //nolint:errcheck // no-op after Commit defer tx.Rollback() //nolint:errcheck // no-op after Commit
before, err := classificationsInUse(ctx, tx)
if err != nil {
return SaveMultipleResult{}, err
}
total := SaveMultipleResult{ total := SaveMultipleResult{
OpportunityResults: make([]SaveResult, len(opportunities)), OpportunityResults: make([]SaveResult, len(opportunities)),
} }
@@ -331,6 +346,14 @@ func (s *Store) SaveOpportunities(
} }
} }
after, err := classificationsInUse(ctx, tx)
if err != nil {
return SaveMultipleResult{}, err
}
added, removed := classificationChanges(before, after)
total.AddedClassifications = added
total.RemovedClassificationIDs = removed
if err := tx.Commit(); err != nil { if err := tx.Commit(); err != nil {
return SaveMultipleResult{}, err return SaveMultipleResult{}, err
} }
@@ -338,6 +361,51 @@ func (s *Store) SaveOpportunities(
return total, nil return total, nil
} }
// classificationsInUse returns the classification codes at least one
// opportunity has, ordered by code, as tx sees them.
func classificationsInUse(
ctx context.Context, tx *sqlx.Tx,
) ([]types.ClassificationCode, error) {
var classifications []types.ClassificationCode
err := tx.SelectContext(ctx, &classifications, selectClassificationsInUse)
if err != nil {
return nil, fmt.Errorf("listing classifications: %w", err)
}
return classifications, nil
}
// classificationChanges compares the classification codes in use before and
// after a save: it returns the codes in after but not before, and the ids of
// those in before but not after, each in the order they're given.
func classificationChanges(
before, after []types.ClassificationCode,
) ([]types.ClassificationCode, []int64) {
wasInUse := make(map[int64]bool, len(before))
for _, classification := range before {
wasInUse[classification.ID] = true
}
isInUse := make(map[int64]bool, len(after))
for _, classification := range after {
isInUse[classification.ID] = true
}
var added []types.ClassificationCode
for _, classification := range after {
if !wasInUse[classification.ID] {
added = append(added, classification)
}
}
var removed []int64
for _, classification := range before {
if !isInUse[classification.ID] {
removed = append(removed, classification.ID)
}
}
return added, removed
}
// saveOpportunity does the work of SaveOpportunity and SaveOpportunities for // saveOpportunity does the work of SaveOpportunity and SaveOpportunities for
// one prepared opportunity inside tx. isNew reports that the opportunity was // one prepared opportunity inside tx. isNew reports that the opportunity was
// not already stored. // not already stored.
+66
View File
@@ -944,3 +944,69 @@ func TestListClassifications(t *testing.T) {
t.Errorf("after the change: codes = %s; want 03000000", got) t.Errorf("after the change: codes = %s; want 03000000", got)
} }
} }
// SaveOpportunities reports the classification codes a save brings into use
// and those it takes out of use, and none when the set doesn't change.
func TestSaveOpportunitiesClassificationChanges(t *testing.T) {
s := openStore(t)
ctx := context.Background()
save := func(opportunity *types.Opportunity) store.SaveMultipleResult {
t.Helper()
total, err := s.SaveOpportunities(ctx, []*types.Opportunity{opportunity})
if err != nil {
t.Fatal(err)
}
return total
}
codesOf := func(classifications []types.ClassificationCode) string {
var codes []string
for _, classification := range classifications {
codes = append(codes, classification.Code)
}
return strings.Join(codes, ",")
}
total := save(opportunity("ocds-1", "First", 1, "r1"))
if codesOf(total.AddedClassifications) != "03000000,72000000" ||
total.RemovedClassificationIDs != nil {
t.Fatalf(
"first save: added %v, removed %v; want both codes added",
total.AddedClassifications, total.RemovedClassificationIDs,
)
}
removedID := total.AddedClassifications[1].ID // 72000000
total = save(opportunity("ocds-2", "Second", 1, "r1"))
if total.AddedClassifications != nil ||
total.RemovedClassificationIDs != nil {
t.Errorf(
"codes already in use: added %v, removed %v; want none",
total.AddedClassifications, total.RemovedClassificationIDs,
)
}
// ocds-1 drops 72000000, which ocds-2 still has: still in use.
newer := opportunity("ocds-1", "Newer", 2, "r2")
newer.Classifications = newer.Classifications[1:]
total = save(newer)
if total.AddedClassifications != nil ||
total.RemovedClassificationIDs != nil {
t.Errorf(
"code still in use: added %v, removed %v; want none",
total.AddedClassifications, total.RemovedClassificationIDs,
)
}
// ocds-2 drops it too: it goes out of use.
newer = opportunity("ocds-2", "Newer", 2, "r2")
newer.Classifications = newer.Classifications[1:]
total = save(newer)
if total.AddedClassifications != nil ||
len(total.RemovedClassificationIDs) != 1 ||
total.RemovedClassificationIDs[0] != removedID {
t.Errorf(
"code out of use: added %v, removed %v; want removed [%d]",
total.AddedClassifications, total.RemovedClassificationIDs, removedID,
)
}
}