diff --git a/internal/api/api.go b/internal/api/api.go index d5b4eeb..581f3c5 100644 --- a/internal/api/api.go +++ b/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) } } diff --git a/internal/api/opportunities.go b/internal/api/opportunities.go index 3b31cd8..0418227 100644 --- a/internal/api/opportunities.go +++ b/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, diff --git a/internal/api/opportunities_test.go b/internal/api/opportunities_test.go index f519c95..6a2da56 100644 --- a/internal/api/opportunities_test.go +++ b/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]) + } + } +} diff --git a/internal/cli/serve/command.go b/internal/cli/serve/command.go index c583c46..1f5d7c8 100644 --- a/internal/cli/serve/command.go +++ b/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...) diff --git a/internal/events/events.go b/internal/events/events.go index 672df9d..41da862 100644 --- a/internal/events/events.go +++ b/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 ---------- diff --git a/internal/server/server.go b/internal/server/server.go index 262e847..3eb735b 100644 --- a/internal/server/server.go +++ b/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. diff --git a/internal/server/server_test.go b/internal/server/server_test.go index 26c3dff..6b5c25e 100644 --- a/internal/server/server_test.go +++ b/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) + } +}