Added an events bus, and pushed OpportunitiesUpdated to WebSocket clients when the fetcher stores new opportunities

internal/events decouples the parts of JobSearch that cause events from
the parts that react to them. bus.go has Bus: Subscribe, and a synchronous
Publish that calls each handler in turn on the publisher's goroutine, so
handlers mustn't block (the hub doesn't: it queues per client). events.go
has the event types and their payloads, starting with OpportunitiesUpdated.

With fetcher.WithEvents, a save that stores new opportunities, or new
releases of stored ones, publishes OpportunitiesUpdated with their UUIDs and
counts. An opportunity whose row is only rewritten, as when it is fetched
again from the newest stored release, isn't counted as updated, and nothing
is published when nothing changed. store.SaveResult now reports
NewOpportunity for each opportunity saved.

job-search serve subscribes its WebSocket hub to the bus, so every client
gets {"type":"event","eventType":"OpportunitiesUpdated","payload":{"new":
[...],"updated":[...],"summary":{"new":n,"updated":n}}}. The Envelope gains
an eventType field, and Hub.PushEvent and Server.PushEvent send events.

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() {

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)
+	}
+}

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()
+}

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"`
+}

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)
+	}
+}

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,
+	})
+}

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)
+	}
+}

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.

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

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,
+		)
+	}
+}

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)

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
 	}
 

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"