Published an OpportunityMarksUpdated event when an opportunity's marks change over the API

internal/api/api.go

@@ -3,6 +3,7 @@ package api
 import (
 	"github.com/go-chi/chi/v5"
 
+	"git.avsdev.uk/Go/job-search/internal/events"
 	"git.avsdev.uk/Go/job-search/internal/store"
 )
 
@@ -11,7 +12,8 @@ import (
 // server applies the API's middleware (CSRF, NoStore, RequireJSON, rate
 // limits); the API only registers handlers.
 type API struct {
-	store *store.Store // nil: no opportunity routes
+	store  *store.Store // nil: no opportunity routes
+	events *events.Bus  // nil: nothing is published
 }
 
 // Option configures the API.
@@ -25,6 +27,14 @@ func WithStore(opportunityStore *store.Store) Option {
 	}
 }
 
+// WithEvents makes the API publish its changes to bus, such as an
+// events.OpportunityMarksUpdated event when an opportunity's marks change.
+func WithEvents(bus *events.Bus) Option {
+	return func(a *API) {
+		a.events = bus
+	}
+}
+
 // New returns the API.
 func New(opts ...Option) *API {
 	api := &API{}
@@ -40,6 +50,6 @@ func New(opts ...Option) *API {
 func (a *API) Routes(router chi.Router) {
 	docsRoutes(router)
 	if a.store != nil {
-		newOpportunityRoutes(a.store).Routes(router)
+		newOpportunityRoutes(a.store, a.events).Routes(router)
 	}
 }

internal/api/opportunities.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-chi/chi/v5"
 
+	"git.avsdev.uk/Go/job-search/internal/events"
 	"git.avsdev.uk/Go/job-search/internal/httpx"
 	"git.avsdev.uk/Go/job-search/internal/store"
 	"git.avsdev.uk/Go/job-search/internal/types"
@@ -33,7 +34,8 @@ type OpportunityPage struct {
 // opportunityRoutes wires HTTP handlers to the opportunity store.
 // Opportunities are read-only; only their marks can be changed.
 type opportunityRoutes struct {
-	store *store.Store
+	store  *store.Store
+	events *events.Bus // nil: nothing is published
 
 	// marksMu makes changing one mark (read the marks, change one, write
 	// them all back) atomic, so concurrent changes to different marks of an
@@ -42,10 +44,13 @@ type opportunityRoutes struct {
 }
 
 // newOpportunityRoutes returns the opportunity handlers for
-// opportunityStore.
-func newOpportunityRoutes(opportunityStore *store.Store) *opportunityRoutes {
+// opportunityStore, publishing mark changes to bus if it isn't nil.
+func newOpportunityRoutes(
+	opportunityStore *store.Store, bus *events.Bus,
+) *opportunityRoutes {
 	return &opportunityRoutes{
-		store: opportunityStore,
+		store:  opportunityStore,
+		events: bus,
 	}
 }
 
@@ -179,7 +184,9 @@ func (h *opportunityRoutes) clearMark(w http.ResponseWriter, r *http.Request) {
 
 // changeMark sets the mark named in the URL to value, keeping the
 // opportunity's other marks, and responds with all of its marks. Setting won
-// while lost is set, or lost while won is, is refused with 409 Conflict.
+// while lost is set, or lost while won is, is refused with 409 Conflict. A
+// change is published as an events.OpportunityMarksUpdated event; one that
+// changes nothing is neither stored nor published.
 func (h *opportunityRoutes) changeMark(
 	w http.ResponseWriter, r *http.Request, value bool,
 ) {
@@ -196,6 +203,7 @@ func (h *opportunityRoutes) changeMark(
 
 	marks := opportunity.Marks
 	marks.UUID = uuid
+	before := marks
 	if !setMark(&marks, name, value) {
 		httpx.WriteProblem(
 			w,
@@ -205,6 +213,10 @@ func (h *opportunityRoutes) changeMark(
 		)
 		return
 	}
+	if marks == before {
+		httpx.WriteJSON(w, http.StatusOK, marks)
+		return
+	}
 
 	err = h.store.SetMarks(r.Context(), marks)
 	if err != nil {
@@ -213,6 +225,15 @@ func (h *opportunityRoutes) changeMark(
 	}
 
 	httpx.WriteJSON(w, http.StatusOK, marks)
+
+	// Still under marksMu, so events are published in the order the
+	// changes were made.
+	if h.events != nil {
+		h.events.Publish(events.Event{
+			Type:    events.OpportunityMarksUpdated,
+			Payload: marks,
+		})
+	}
 }
 
 // findMarks returns the marks of types.OpportunityMarks: every bool field,

internal/api/opportunities_test.go

@@ -14,13 +14,15 @@ import (
 
 	"github.com/go-chi/chi/v5"
 
+	"git.avsdev.uk/Go/job-search/internal/events"
 	"git.avsdev.uk/Go/job-search/internal/store"
 	"git.avsdev.uk/Go/job-search/internal/types"
 )
 
-// opportunitiesServer serves the opportunity routes over a store holding
-// count opportunities, ocds-1 the oldest, and returns its URL.
-func opportunitiesServer(t *testing.T, count int) string {
+// opportunitiesServer serves the API, given opts as well as its store, over
+// a store holding count opportunities, ocds-1 the oldest, and returns its
+// URL.
+func opportunitiesServer(t *testing.T, count int, opts ...Option) string {
 	t.Helper()
 	opportunityStore, err := store.Open(
 		context.Background(),
@@ -57,7 +59,7 @@ func opportunitiesServer(t *testing.T, count int) string {
 	}
 
 	router := chi.NewRouter()
-	New(WithStore(opportunityStore)).Routes(router)
+	New(append([]Option{WithStore(opportunityStore)}, opts...)...).Routes(router)
 	server := httptest.NewServer(router)
 	t.Cleanup(server.Close)
 	return server.URL
@@ -302,3 +304,55 @@ func TestFindMarks(t *testing.T) {
 		t.Error("setMark accepted a name that isn't a mark")
 	}
 }
+
+// A change to a mark is published with all the marks; a change that changes
+// nothing, or is refused, isn't.
+func TestChangeMarkPublishes(t *testing.T) {
+	var bus events.Bus
+	var mu sync.Mutex
+	var published []types.OpportunityMarks
+	bus.Subscribe(func(event events.Event) {
+		if event.Type != events.OpportunityMarksUpdated {
+			t.Errorf("event type %q", event.Type)
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		published = append(published, event.Payload.(types.OpportunityMarks))
+	})
+	url := opportunitiesServer(t, 1, WithEvents(&bus))
+	marks := url + "/opportunities/ocds-1/marks/"
+
+	request(t, "PUT", marks+"interested", nil)    // published
+	request(t, "PUT", marks+"interested", nil)    // already set: not published
+	request(t, "DELETE", marks+"skipped", nil)    // not set: not published
+	request(t, "PUT", marks+"shiny", nil)         // no such mark: refused
+	request(t, "PUT", marks+"won", nil)           // published
+	request(t, "PUT", marks+"lost", nil)          // won and lost: refused
+	request(t, "DELETE", marks+"interested", nil) // published
+
+	mu.Lock()
+	defer mu.Unlock()
+	want := []types.OpportunityMarks{
+		{
+			UUID:       "ocds-1",
+			Interested: true,
+		},
+		{
+			UUID:       "ocds-1",
+			Interested: true,
+			Won:        true,
+		},
+		{
+			UUID: "ocds-1",
+			Won:  true,
+		},
+	}
+	if len(published) != len(want) {
+		t.Fatalf("published %+v; want %+v", published, want)
+	}
+	for i := range want {
+		if published[i] != want[i] {
+			t.Errorf("event %d: %+v; want %+v", i, published[i], want[i])
+		}
+	}
+}

internal/cli/serve/command.go

@@ -136,6 +136,11 @@ set, the API serves the stored opportunities under /api/opportunities.`,
 				)
 			}
 
+			// Events from the sources and the API, such as new opportunities and
+			// changed marks, are pushed to every WebSocket client.
+			var bus events.Bus
+			options = append(options, server.WithEvents(&bus))
+
 			appServer, err := server.New(&conf.Server, options...)
 			if err != nil {
 				return err
@@ -146,12 +151,6 @@ set, the API serves the stored opportunities under /api/opportunities.`,
 			runCtx, cancelRun := context.WithCancel(ctx)
 			defer cancelRun()
 
-			// 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...)

internal/events/events.go

@@ -4,9 +4,17 @@
 // 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"
+// Event types: an Event's Type, which says what its Payload is.
+const (
+	// OpportunitiesUpdated is published when a save stores new
+	// opportunities or new releases of stored ones. Its payload is an
+	// OpportunitiesUpdatedPayload.
+	OpportunitiesUpdated = "OpportunitiesUpdated"
+	// OpportunityMarksUpdated is published when an opportunity's marks are
+	// changed over the API. Its payload is the opportunity's
+	// types.OpportunityMarks after the change, which include its UUID.
+	OpportunityMarksUpdated = "OpportunityMarksUpdated"
+)
 
 // ---------- OpportunitiesUpdated ----------
 

internal/server/server.go

@@ -19,6 +19,7 @@ import (
 
 	"git.avsdev.uk/Go/job-search/internal/api"
 	"git.avsdev.uk/Go/job-search/internal/config"
+	"git.avsdev.uk/Go/job-search/internal/events"
 	"git.avsdev.uk/Go/job-search/internal/logging"
 	"git.avsdev.uk/Go/job-search/internal/middleware"
 	"git.avsdev.uk/Go/job-search/internal/store"
@@ -72,6 +73,16 @@ func WithStore(store *store.Store) Option {
 	}
 }
 
+// WithEvents connects the server to bus: every event published to it is
+// pushed to the server's WebSocket clients (as websocket.TypeEvent
+// messages), and the server's API publishes its changes to it (see
+// api.WithEvents).
+func WithEvents(bus *events.Bus) Option {
+	return func(s *Server) {
+		s.events = bus
+	}
+}
+
 // WithMiddleware appends middleware to a group. This is the generic hook
 // (auth, metrics, tracing, custom header rules, ...).
 func WithMiddleware(group Group, middlewares ...Middleware) Option {
@@ -120,6 +131,9 @@ type Server struct {
 	hub                  *websocket.Hub
 	// store is the opportunities' database, or nil for none.
 	store *store.Store
+	// events is where events are pushed from and the API publishes to, or
+	// nil for none.
+	events *events.Bus
 	// api is the REST API mounted under /api, built from the options.
 	api     *api.API
 	started time.Time
@@ -142,7 +156,10 @@ func New(conf *config.JSCServer, opts ...Option) (*Server, error) {
 	for _, option := range opts {
 		option(server)
 	}
-	server.api = api.New(api.WithStore(server.store))
+	server.api = api.New(
+		api.WithStore(server.store),
+		api.WithEvents(server.events),
+	)
 
 	var err error
 	server.trusted, err = util.ParseNetPrefixes(conf.Proxy.TrustedProxies)
@@ -190,6 +207,13 @@ func New(conf *config.JSCServer, opts ...Option) (*Server, error) {
 		server.handler = logging.AccessLog(server.accessLog, true, server.handler)
 	}
 
+	// Subscribed last, so a server that failed to build is never pushed to.
+	if server.events != nil {
+		server.events.Subscribe(func(event events.Event) {
+			server.hub.PushEvent(event.Type, event.Payload)
+		})
+	}
+
 	return server, nil
 }
 
@@ -210,12 +234,6 @@ 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/server/server_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/spf13/viper"
 
 	"git.avsdev.uk/Go/job-search/internal/config"
+	"git.avsdev.uk/Go/job-search/internal/events"
 	"git.avsdev.uk/Go/job-search/internal/middleware"
 	"git.avsdev.uk/Go/job-search/internal/store"
 	"git.avsdev.uk/Go/job-search/internal/types"
@@ -336,10 +337,11 @@ func (b *syncBuffer) String() string {
 	return b.buffer.String()
 }
 
-// With WithOpportunityStore the opportunities are served, and changing a mark
-// needs the CSRF token; without it there are no opportunity routes.
-func TestOpportunityRoutes(t *testing.T) {
-	opportunityStore, err := store.Open(
+// opportunityStore returns a store, in a temporary file, holding one
+// opportunity, ocds-1.
+func opportunityStore(t *testing.T) *store.Store {
+	t.Helper()
+	st, err := store.Open(
 		context.Background(),
 		filepath.Join(t.TempDir(), "server.db"),
 		store.WithAutoMigrate(),
@@ -348,9 +350,9 @@ func TestOpportunityRoutes(t *testing.T) {
 		t.Fatal(err)
 	}
 	t.Cleanup(func() {
-		opportunityStore.Close()
+		st.Close()
 	})
-	_, err = opportunityStore.SaveOpportunity(
+	_, err = st.SaveOpportunity(
 		context.Background(),
 		&types.Opportunity{
 			UUID:     "ocds-1",
@@ -368,7 +370,13 @@ func TestOpportunityRoutes(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	ts := newTestServer(t, nil, WithStore(opportunityStore))
+	return st
+}
+
+// With WithStore the opportunities are served, and changing a mark needs
+// the CSRF token; without it there are no opportunity routes.
+func TestOpportunityRoutes(t *testing.T) {
+	ts := newTestServer(t, nil, WithStore(opportunityStore(t)))
 	c := clientWithJar(t)
 	do := apiClient(t, ts, c, getToken(t, c, ts.URL))
 
@@ -393,3 +401,54 @@ func TestOpportunityRoutes(t *testing.T) {
 		t.Errorf("no store: %d; want 404", res.StatusCode)
 	}
 }
+
+// With WithEvents, events published to the bus, and the API's own, such as
+// a changed mark, are pushed to WebSocket clients.
+func TestEventsPushed(t *testing.T) {
+	var bus events.Bus
+	ts := newTestServer(
+		t, nil, WithStore(opportunityStore(t)), WithEvents(&bus),
+	)
+	c := clientWithJar(t)
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	conn, _, err := websocket.Dial(
+		ctx,
+		"ws"+strings.TrimPrefix(ts.URL, "http")+"/ws",
+		&websocket.DialOptions{
+			HTTPClient: c,
+		},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.CloseNow()
+	read := func() ws.Envelope {
+		var e ws.Envelope
+		if err := wsjson.Read(ctx, conn, &e); err != nil {
+			t.Fatal(err)
+		}
+		return e
+	}
+	read() // welcome
+
+	bus.Publish(events.Event{
+		Type:    events.OpportunitiesUpdated,
+		Payload: map[string]int{"n": 1},
+	})
+	e := read()
+	if e.Type != ws.TypeEvent || e.EventType != events.OpportunitiesUpdated {
+		t.Fatalf("want the published event, got %+v", e)
+	}
+
+	do := apiClient(t, ts, c, getToken(t, c, ts.URL))
+	mark := "/api/opportunities/ocds-1/marks/interested"
+	if res := do("PUT", mark, "", true); res.StatusCode != 200 {
+		t.Fatalf("mark: %d", res.StatusCode)
+	}
+	e = read()
+	if e.EventType != events.OpportunityMarksUpdated ||
+		!strings.Contains(string(e.Payload), `"interested":true`) {
+		t.Fatalf("want the marks event, got %+v", e)
+	}
+}