diff --git a/internal/cli/serve/command.go b/internal/cli/serve/command.go index 570f128..ad812d6 100644 --- a/internal/cli/serve/command.go +++ b/internal/cli/serve/command.go @@ -16,6 +16,7 @@ import ( "github.com/spf13/cobra" "git.avsdev.uk/Go/job-search/internal/cli" + "git.avsdev.uk/Go/job-search/internal/events" "git.avsdev.uk/Go/job-search/internal/fetcher" "git.avsdev.uk/Go/job-search/internal/logging" "git.avsdev.uk/Go/job-search/internal/server" @@ -30,9 +31,10 @@ func NewCommand(opts *cli.Options) *cobra.Command { Long: `Serve the REST API under /api and the WebSocket endpoint under /ws. It also runs a worker for every enabled source, as fetcher service does, -saving what they fetch to database.file if it is set. A source that stops -with an error stops the server too, and the server stopping stops the -sources.`, +saving what they fetch to database.file if it is set, and pushing an +OpportunitiesUpdated event to every WebSocket client when a save stores new +opportunities or new releases. A source that stops with an error stops the +server too, and the server stopping stops the sources.`, PersistentPreRun: func(cmd *cobra.Command, args []string) { slog.Debug("command persistent pre-run", "command", "serve") }, @@ -136,6 +138,14 @@ sources.`, ) } + // Events from the sources, such as new opportunities, are pushed to + // every WebSocket client. + var bus events.Bus + bus.Subscribe(func(event events.Event) { + appServer.PushEvent(event.Type, event.Payload) + }) + fetcherOptions = append(fetcherOptions, fetcher.WithEvents(&bus)) + sourceFetcher := fetcher.New(&conf.Sources, fetcherOptions...) fetcherDone := make(chan error, 1) go func() { diff --git a/internal/events/bus.go b/internal/events/bus.go new file mode 100644 index 0000000..b018b3a --- /dev/null +++ b/internal/events/bus.go @@ -0,0 +1,40 @@ +package events + +import "sync" + +// Bus passes each published Event to every subscriber. Its zero value is +// ready to use, and it is safe for concurrent use. +type Bus struct { + mu sync.RWMutex + handlers []func(Event) +} + +// Event is something that happened, for subscribers to react to. Type names +// it (e.g. OpportunitiesUpdated) and says what Payload is. +type Event struct { + Type string + Payload any +} + +// Subscribe makes handler receive every event published from now on. +// Subscribe before anything publishes, so no event is missed. +func (b *Bus) Subscribe(handler func(Event)) { + b.mu.Lock() + defer b.mu.Unlock() + b.handlers = append(b.handlers, handler) +} + +// Publish passes event to every subscriber in turn, in the order they +// subscribed, and returns once they all have. Handlers are called on the +// publisher's goroutine, so they mustn't block; one with slow work to do +// should queue it. Publish may be called from several goroutines at once. +func (b *Bus) Publish(event Event) { + b.mu.RLock() + handlers := b.handlers + b.mu.RUnlock() + + // Called outside the lock, so a handler may itself subscribe or publish. + for _, handler := range handlers { + handler(event) + } +} diff --git a/internal/events/bus_test.go b/internal/events/bus_test.go new file mode 100644 index 0000000..0bb43b1 --- /dev/null +++ b/internal/events/bus_test.go @@ -0,0 +1,90 @@ +package events + +import ( + "sync" + "sync/atomic" + "testing" +) + +// Every subscriber gets every event, in the order they subscribed. +func TestPublish(t *testing.T) { + var bus Bus + var got []string + for _, name := range []string{"first", "second"} { + bus.Subscribe(func(e Event) { + got = append(got, name+":"+e.Type) + }) + } + + bus.Publish(Event{ + Type: "a", + }) + bus.Publish(Event{ + Type: "b", + }) + + want := []string{"first:a", "second:a", "first:b", "second:b"} + if len(got) != len(want) { + t.Fatalf("got %v; want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("got %v; want %v", got, want) + } + } +} + +// Publishing with no subscribers does nothing. +func TestPublishNoSubscribers(t *testing.T) { + var bus Bus + bus.Publish(Event{ + Type: OpportunitiesUpdated, + }) +} + +// A handler may subscribe or publish without deadlocking. +func TestHandlerReenters(t *testing.T) { + var bus Bus + var inner atomic.Int32 + bus.Subscribe(func(e Event) { + if e.Type == "outer" { + bus.Subscribe(func(Event) { + inner.Add(1) + }) + bus.Publish(Event{ + Type: "inner", + }) + } + }) + + bus.Publish(Event{ + Type: "outer", + }) + if inner.Load() != 1 { + t.Errorf("inner handler ran %d times; want 1", inner.Load()) + } +} + +// Publish and Subscribe are safe from several goroutines at once (run with +// -race). +func TestConcurrent(t *testing.T) { + var bus Bus + var count atomic.Int32 + var wg sync.WaitGroup + for range 10 { + wg.Add(2) + go func() { + defer wg.Done() + bus.Subscribe(func(Event) { + count.Add(1) + }) + }() + go func() { + defer wg.Done() + bus.Publish(Event{ + Type: "x", + }) + }() + } + wg.Wait() +} diff --git a/internal/events/events.go b/internal/events/events.go new file mode 100644 index 0000000..672df9d --- /dev/null +++ b/internal/events/events.go @@ -0,0 +1,27 @@ +// Package events passes events between the parts of JobSearch that cause +// them, such as the fetcher, and the parts that react to them, such as the +// WebSocket hub, so neither depends on the other. It holds the event types +// and their payloads. +package events + +// OpportunitiesUpdated is published when a save stores new opportunities or +// new releases of stored ones. Its payload is an OpportunitiesUpdatedPayload. +const OpportunitiesUpdated = "OpportunitiesUpdated" + +// ---------- OpportunitiesUpdated ---------- + +// 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. +type OpportunitiesUpdatedPayload struct { + New []string `json:"new"` + Updated []string `json:"updated"` + Summary OpportunitiesSummary `json:"summary"` +} + +// OpportunitiesSummary counts the opportunities in an +// OpportunitiesUpdatedPayload. +type OpportunitiesSummary struct { + New int `json:"new"` + Updated int `json:"updated"` +} diff --git a/internal/events/events_test.go b/internal/events/events_test.go new file mode 100644 index 0000000..01fe04c --- /dev/null +++ b/internal/events/events_test.go @@ -0,0 +1,25 @@ +package events + +import ( + "encoding/json" + "testing" +) + +// The payload marshals with the keys clients read, and empty lists as []. +func TestOpportunitiesUpdatedPayloadJSON(t *testing.T) { + payload := OpportunitiesUpdatedPayload{ + New: []string{"ocds-1"}, + Updated: []string{}, + Summary: OpportunitiesSummary{ + New: 1, + }, + } + encoded, err := json.Marshal(payload) + if err != nil { + t.Fatal(err) + } + want := `{"new":["ocds-1"],"updated":[],"summary":{"new":1,"updated":0}}` + 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 e370388..ece9863 100644 --- a/internal/fetcher/fetcher.go +++ b/internal/fetcher/fetcher.go @@ -9,6 +9,7 @@ import ( "time" "git.avsdev.uk/Go/job-search/internal/config" + "git.avsdev.uk/Go/job-search/internal/events" "git.avsdev.uk/Go/job-search/internal/store" "git.avsdev.uk/Go/job-search/internal/types" "git.avsdev.uk/Go/job-search/internal/util" @@ -37,6 +38,15 @@ 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. +func WithEvents(bus *events.Bus) Option { + return func(f *Fetcher) { + f.events = bus + } +} + // saveFunc saves the opportunities a worker fetched; see Fetcher.save. type saveFunc func( ctx context.Context, @@ -66,6 +76,7 @@ type Fetcher struct { conf *config.JSCSources log *slog.Logger store *store.Store // nil: nothing is persisted + events *events.Bus // nil: nothing is published started time.Time // backgroundCtx is the context the workers run under. backgroundCancel @@ -199,9 +210,10 @@ func (f *Fetcher) startFrom( return latest, nil } -// save saves opportunities to the store given WithStore, and logs what -// changed. Without a store it does nothing. A failure fails the worker's run, -// so its checkpoint isn't moved on and the next run fetches them again. +// save saves opportunities to the store given WithStore, logs what changed, +// and publishes it (see WithEvents). Without a store it does nothing. A +// failure fails the worker's run, so its checkpoint isn't moved on and the +// next run fetches them again. func (f *Fetcher) save( ctx context.Context, logger *slog.Logger, @@ -223,5 +235,45 @@ func (f *Fetcher) save( "new_releases", result.NewReleases, ) + f.publishUpdates(opportunities, result.OpportunityResults) + return nil } + +// 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 +// whose row was only rewritten, such as one fetched again from the newest +// stored release, isn't counted as updated. +func (f *Fetcher) publishUpdates( + opportunities []*types.Opportunity, results []store.SaveResult, +) { + if f.events == nil { + return + } + + payload := events.OpportunitiesUpdatedPayload{ + New: []string{}, + Updated: []string{}, + } + for i, result := range results { + switch { + case result.NewOpportunity: + payload.New = append(payload.New, opportunities[i].UUID) + case result.NewReleases > 0: + payload.Updated = append(payload.Updated, opportunities[i].UUID) + } + } + payload.Summary = events.OpportunitiesSummary{ + New: len(payload.New), + Updated: len(payload.Updated), + } + if payload.Summary.New == 0 && payload.Summary.Updated == 0 { + return + } + + f.events.Publish(events.Event{ + Type: events.OpportunitiesUpdated, + Payload: payload, + }) +} diff --git a/internal/fetcher/fetcher_test.go b/internal/fetcher/fetcher_test.go index 1e7112d..f24ce39 100644 --- a/internal/fetcher/fetcher_test.go +++ b/internal/fetcher/fetcher_test.go @@ -9,12 +9,14 @@ import ( "net/http" "net/http/httptest" "os" + "reflect" "strings" "sync/atomic" "testing" "time" "git.avsdev.uk/Go/job-search/internal/config" + "git.avsdev.uk/Go/job-search/internal/events" "git.avsdev.uk/Go/job-search/internal/store" "git.avsdev.uk/Go/job-search/internal/types" "git.avsdev.uk/Go/job-search/internal/util" @@ -659,3 +661,79 @@ func checkAboutADayAgo(t *testing.T, updatedFrom string) { t.Errorf("updatedFrom = %s; want about a day ago", updatedFrom) } } + +// fetched returns the opportunities converted from body, a release package. +func fetched(t *testing.T, body string) []*types.Opportunity { + t.Helper() + srv, _ := ftsServer(t, http.StatusOK, body) + opportunities, err := fetchers["FindATender"]( + context.Background(), srv.URL, time.Now().Add(-time.Hour), + ) + if err != nil { + t.Fatal(err) + } + return opportunities +} + +// A save publishes the new opportunities and those with new releases, and +// nothing when an opportunity is only fetched again. +func TestSavePublishesUpdates(t *testing.T) { + var bus events.Bus + var published []events.OpportunitiesUpdatedPayload + bus.Subscribe(func(event events.Event) { + if event.Type != events.OpportunitiesUpdated { + t.Errorf("event type = %q", event.Type) + } + payload := event.Payload.(events.OpportunitiesUpdatedPayload) + published = append(published, payload) + }) + f := New( + &config.JSCSources{}, + WithStore(openTestStore(t)), + WithEvents(&bus), + ) + save := func(body string) { + t.Helper() + err := f.save(context.Background(), slog.Default(), fetched(t, body)) + if err != nil { + t.Fatal(err) + } + } + newRelease := strings.NewReplacer( + `"id": "r1"`, `"id": "r2"`, + "2026-10-01T12:00:00Z", "2026-10-02T12:00:00Z", + ).Replace(onePackage) + + save(onePackage) // new + save(onePackage) // fetched again: nothing published + save(newRelease) // a new release of a stored opportunity + + want := []events.OpportunitiesUpdatedPayload{ + { + New: []string{"ocds-1"}, + Updated: []string{}, + Summary: events.OpportunitiesSummary{ + New: 1, + }, + }, + { + New: []string{}, + Updated: []string{"ocds-1"}, + Summary: events.OpportunitiesSummary{ + Updated: 1, + }, + }, + } + if !reflect.DeepEqual(published, want) { + t.Errorf("published %+v; want %+v", published, want) + } +} + +// Without WithEvents, a save publishes nothing and still succeeds. +func TestSaveWithoutEvents(t *testing.T) { + f := New(&config.JSCSources{}, WithStore(openTestStore(t))) + err := f.save(context.Background(), slog.Default(), fetched(t, onePackage)) + if err != nil { + t.Fatal(err) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 45ed89d..ae0f534 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -194,6 +194,12 @@ func (s *Server) logger() *slog.Logger { // Handler exposes the root handler (useful for httptest). func (s *Server) Handler() http.Handler { return s.handler } +// PushEvent sends an event to every client of the server's WebSocket hub +// (see websocket.Hub.PushEvent). +func (s *Server) PushEvent(eventType string, payload any) { + s.hub.PushEvent(eventType, payload) +} + // Run listens and serves until ctx is cancelled, then shuts down gracefully. // With TLS, it also serves an HTTP to HTTPS redirect if tls.redirect_http_port // is set. diff --git a/internal/store/store.go b/internal/store/store.go index eb8f717..b7b57b1 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -65,6 +65,8 @@ type SaveMultipleResult struct { // SaveResult reports what SaveOpportunity changed. type SaveResult struct { + // NewOpportunity is true if the opportunity was not already stored. + NewOpportunity bool // OpportunityUpdated is true if the opportunity row and its classifications // were written. It is false when every release in the save was older than the // one the stored opportunity was last updated from. @@ -344,7 +346,6 @@ func saveOpportunity( ) (SaveResult, bool, error) { var ( result SaveResult - isNew bool stored time.Time ) err := tx.GetContext( @@ -352,7 +353,7 @@ func saveOpportunity( ) switch { case errors.Is(err, sql.ErrNoRows): - isNew = true + result.NewOpportunity = true result.OpportunityUpdated = true case err != nil: return SaveResult{}, false, fmt.Errorf( @@ -380,7 +381,7 @@ func saveOpportunity( } } - return result, isNew, nil + return result, result.NewOpportunity, nil } // insertReleaseWithTags stores release and its tags unless it is already diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 99483a4..831918c 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -871,3 +871,34 @@ func TestLatestReleaseDate(t *testing.T) { t.Errorf("unknown source: %v, %v; want the zero time", latest, err) } } + +// NewOpportunity is set only for an opportunity that wasn't already stored, +// by both SaveOpportunity and SaveOpportunities. +func TestSaveResultNewOpportunity(t *testing.T) { + s := openStore(t) + ctx := context.Background() + + res := mustSave(t, s, opportunity("ocds-1", "First", 1, "r1")) + if !res.NewOpportunity { + t.Errorf("first save: NewOpportunity = false") + } + res = mustSave(t, s, opportunity("ocds-1", "First", 2, "r2")) + if res.NewOpportunity { + t.Errorf("second save: NewOpportunity = true") + } + + total, err := s.SaveOpportunities(ctx, []*types.Opportunity{ + opportunity("ocds-1", "First", 3, "r3"), + opportunity("ocds-2", "Second", 3, "r1"), + }) + if err != nil { + t.Fatal(err) + } + results := total.OpportunityResults + if results[0].NewOpportunity || !results[1].NewOpportunity { + t.Errorf( + "SaveOpportunities results = %+v; want only the second new", + results, + ) + } +} diff --git a/internal/websocket/http_test.go b/internal/websocket/http_test.go index 5de4555..59c1b6c 100644 --- a/internal/websocket/http_test.go +++ b/internal/websocket/http_test.go @@ -143,6 +143,38 @@ func TestPushReachesBothTransports(t *testing.T) { } } +// PushEvent sends a TypeEvent message naming the event, to both transports. +func TestPushEvent(t *testing.T) { + h, addr := tcpHub(t, nil) + url := wsHub(t, h) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + conn := dialWS(t, ctx, url) + c := dialTCP(t, addr) + c.read() // welcome + waitCount(t, h, 2) + + h.PushEvent("OpportunitiesUpdated", map[string]int{"new": 1}) + var e Envelope + if err := wsjson.Read(ctx, conn, &e); err != nil { + t.Fatal(err) + } + received := map[string]Envelope{ + "websocket": e, + "tcp": c.read(), + } + for transport, got := range received { + if got.Type != TypeEvent || got.EventType != "OpportunitiesUpdated" || + string(got.Payload) != `{"new":1}` { + t.Errorf( + "%s: got %+v; want the OpportunitiesUpdated event", + transport, got, + ) + } + } +} + // A payload that can't be marshalled is not sent. func TestPushBadPayload(t *testing.T) { h, addr := tcpHub(t, nil) diff --git a/internal/websocket/hub.go b/internal/websocket/hub.go index 9e83a6a..d43ba7c 100644 --- a/internal/websocket/hub.go +++ b/internal/websocket/hub.go @@ -78,21 +78,46 @@ func (h *Hub) logger() *slog.Logger { // whose queue is full is closed as a slow consumer. A payload that can't be // marshalled is logged and nothing is sent. func (h *Hub) Push(msgType string, payload any) { - var rawPayload json.RawMessage + h.push( + Envelope{ + Type: msgType, + }, + payload, + ) +} + +// PushEvent sends a TypeEvent message to every client, as Push does, with +// eventType as the envelope's EventType. +func (h *Hub) PushEvent(eventType string, payload any) { + h.push( + Envelope{ + Type: TypeEvent, + EventType: eventType, + }, + payload, + ) +} + +// push sends envelope, with payload marshalled into it, to every client (see +// Push). +func (h *Hub) push(envelope Envelope, payload any) { if payload != nil { var err error - if rawPayload, err = json.Marshal(payload); err != nil { - h.logger().Error("marshal push payload", "type", msgType, "error", err) + envelope.Payload, err = json.Marshal(payload) + if err != nil { + h.logger().Error( + "marshal push payload", + "type", envelope.Type, + "event_type", envelope.EventType, + "error", err, + ) return } } - msg, err := json.Marshal(Envelope{ - Type: msgType, - Payload: rawPayload, - }) + msg, err := json.Marshal(envelope) if err != nil { - h.logger().Error("marshal push", "type", msgType, "error", err) + h.logger().Error("marshal push", "type", envelope.Type, "error", err) return } diff --git a/internal/websocket/messages.go b/internal/websocket/messages.go index 72d67f1..bac1eb3 100644 --- a/internal/websocket/messages.go +++ b/internal/websocket/messages.go @@ -19,23 +19,27 @@ // {"type":"echo","id":"2","payload":{...}} // {"type":"error","id":"3","error":"..."} // {"type":"item.created","payload":{...}} +// {"type":"event","eventType":"OpportunitiesUpdated","payload":{...}} // -// Any other type from a client gets an "unknown message type" error. The last -// message is one sent by Hub.Push: its type is the pushed message type. +// Any other type from a client gets an "unknown message type" error. The +// last two are sent by Hub.Push, with the pushed message type, and by +// Hub.PushEvent, with TypeEvent and the event's type (see package events). package websocket import "encoding/json" // Envelope is the wire format for every message in both directions. +// EventType is set only on a TypeEvent message, naming the event. type Envelope struct { - Type string `json:"type"` - ID string `json:"id,omitempty"` - Payload json.RawMessage `json:"payload,omitempty"` - Error string `json:"error,omitempty"` + Type string `json:"type"` + EventType string `json:"eventType,omitempty"` + ID string `json:"id,omitempty"` + Payload json.RawMessage `json:"payload,omitempty"` + Error string `json:"error,omitempty"` } -// Envelope types. TypeEvent, TypeRequest and TypeAck are reserved: nothing -// sends or handles them yet. +// Envelope types. TypeRequest and TypeAck are reserved: nothing sends or +// handles them yet. const ( // TypeWelcome is sent to a client when it connects, with its client ID. TypeWelcome = "welcome" @@ -45,7 +49,8 @@ const ( TypePong = "pong" // TypeEcho is sent back to the client with the same ID and payload. TypeEcho = "echo" - // TypeEvent is reserved for events pushed to clients. + // TypeEvent is an event pushed to clients by Hub.PushEvent, named by the + // envelope's EventType. TypeEvent = "event" // TypeRequest is reserved for requests from clients. TypeRequest = "request"