diff --git a/internal/events/events.go b/internal/events/events.go index 41da862..665fe90 100644 --- a/internal/events/events.go +++ b/internal/events/events.go @@ -4,6 +4,8 @@ // and their payloads. package events +import "git.avsdev.uk/Go/job-search/internal/types" + // Event types: an Event's Type, which says what its Payload is. const ( // OpportunitiesUpdated is published when a save stores new @@ -14,6 +16,10 @@ const ( // changed over the API. Its payload is the opportunity's // types.OpportunityMarks after the change, which include its UUID. 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 ---------- @@ -33,3 +39,14 @@ type OpportunitiesSummary struct { New int `json:"new"` 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"` +} diff --git a/internal/events/events_test.go b/internal/events/events_test.go index 01fe04c..2d8578e 100644 --- a/internal/events/events_test.go +++ b/internal/events/events_test.go @@ -3,6 +3,8 @@ package events import ( "encoding/json" "testing" + + "git.avsdev.uk/Go/job-search/internal/types" ) // 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) } } + +// 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) + } +} diff --git a/internal/fetcher/fetcher.go b/internal/fetcher/fetcher.go index ece9863..5504cc4 100644 --- a/internal/fetcher/fetcher.go +++ b/internal/fetcher/fetcher.go @@ -39,8 +39,10 @@ func WithStore(opportunityStore *store.Store) Option { } // WithEvents makes the workers publish an events.OpportunitiesUpdated event -// to bus whenever a save stores new opportunities or new releases. It needs -// WithStore: without a store nothing is saved, so nothing is published. +// to bus whenever a save stores new opportunities or new releases, and an +// 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 { return func(f *Fetcher) { f.events = bus @@ -235,11 +237,45 @@ func (f *Fetcher) save( "new_releases", result.NewReleases, ) + f.publishClassificationChanges(result) f.publishUpdates(opportunities, result.OpportunityResults) 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 // 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 diff --git a/internal/fetcher/fetcher_test.go b/internal/fetcher/fetcher_test.go index f24ce39..1e94164 100644 --- a/internal/fetcher/fetcher_test.go +++ b/internal/fetcher/fetcher_test.go @@ -737,3 +737,74 @@ func TestSaveWithoutEvents(t *testing.T) { 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, + ) + } +} diff --git a/internal/store/store.go b/internal/store/store.go index 6a24dfd..57ff1f2 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -61,6 +61,14 @@ type SaveMultipleResult struct { UpdatedOpportunities int // OpportunityResults is a slice of the results of each opportunity saved. 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. @@ -297,7 +305,9 @@ func (s *Store) SaveOpportunity( // SetMarks to change them). // // 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( ctx context.Context, opportunities []*types.Opportunity, ) (SaveMultipleResult, error) { @@ -313,6 +323,11 @@ func (s *Store) SaveOpportunities( } defer tx.Rollback() //nolint:errcheck // no-op after Commit + before, err := classificationsInUse(ctx, tx) + if err != nil { + return SaveMultipleResult{}, err + } + total := SaveMultipleResult{ 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 { return SaveMultipleResult{}, err } @@ -338,6 +361,51 @@ func (s *Store) SaveOpportunities( 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 // one prepared opportunity inside tx. isNew reports that the opportunity was // not already stored. diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 5c57d9e..60face4 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -944,3 +944,69 @@ func TestListClassifications(t *testing.T) { 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, + ) + } +}