feat: instant job update notifications using server-sent events

This commit introduces a new feature that enables clients to receive
instant notifications for job status updates.

Details:

- A new endpoint `/jobs/{id}/status/subscribe` has been added, which
  implements instant notifications through Server-Sent Events (SSE).
- By subscribing to this endpoint, clients can efficiently keep track of
  job status changes without the need to repeatedly poll the server.
- A client reference implementation has been added to wfxctl.

Signed-off-by: Michael Adler <[email protected]>

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Instant job update notifications using server-sent events (see #11)
+
 ### Fixed
 
 - Send HTTP status code 404 when attempting to access the file server while it is disabled

api/errors.go

@@ -41,3 +41,9 @@ var WorkflowInvalid = model.Error{
 	Logref:  "18f57adc70dd79c7fb4f1246be8a6e04",
 	Message: "Workflow validation failed",
 }
+
+var JobTerminalState = model.Error{
+	Code:    "wfx.jobTerminalState",
+	Logref:  "916f0a913a3e4a52a96bd271e029c201",
+	Message: "The request was invalid because the job is in a terminal state",
+}

api/job_status_test.go

@@ -12,11 +12,15 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/siemens/wfx/generated/model"
 	"github.com/siemens/wfx/internal/handler/job"
+	"github.com/siemens/wfx/internal/handler/job/status"
 	"github.com/siemens/wfx/internal/handler/workflow"
+	"github.com/siemens/wfx/persistence"
 	"github.com/siemens/wfx/workflow/dau"
 	"github.com/steinfletcher/apitest"
 	jsonpath "github.com/steinfletcher/apitest-jsonpath"
@@ -118,3 +122,120 @@ func TestJobStatusUpdate(t *testing.T) {
 		Assert(jsonpath.Contains(`$.state`, "DOWNLOADING")).
 		End()
 }
+
+func TestJobStatusSubscribe(t *testing.T) {
+	db := newInMemoryDB(t)
+	wf := dau.DirectWorkflow()
+	_, err := workflow.CreateWorkflow(context.Background(), db, wf)
+	require.NoError(t, err)
+
+	north, south := createNorthAndSouth(t, db)
+
+	handlers := []http.Handler{north, south}
+	for i, name := range allAPIs {
+		handler := handlers[i]
+		t.Run(name, func(t *testing.T) {
+			jobReq := model.JobRequest{
+				ClientID: "foo",
+				Workflow: wf.Name,
+			}
+			job, err := job.CreateJob(context.Background(), db, &jobReq)
+			require.NoError(t, err)
+			jobPath := fmt.Sprintf("/api/wfx/v1/jobs/%s/status/subscribe", job.ID)
+
+			job, err = db.UpdateJob(context.Background(), job, persistence.JobUpdate{
+				Status: &model.JobStatus{State: "ACTIVATING"},
+			})
+			require.NoError(t, err)
+
+			var wg sync.WaitGroup
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				for {
+					if status.SubscriberCount() > 0 {
+						break
+					}
+					time.Sleep(time.Millisecond)
+				}
+				// give it some extra time, just to be safe
+				time.Sleep(500 * time.Millisecond)
+				// update job to terminal state
+				_, err = status.Update(context.Background(), db, job.ID, &model.JobStatus{State: "ACTIVATED"}, model.EligibleEnumCLIENT)
+				require.NoError(t, err)
+			}()
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			apitest.New().
+				Handler(handler).
+				Get(jobPath).WithContext(ctx).
+				Expect(t).
+				Status(http.StatusOK).
+				Header("Content-Type", "text/event-stream").
+				Body(`data: {"state":"ACTIVATING"}
+
+data: {"state":"ACTIVATED"}
+
+`).
+				End()
+
+			wg.Wait()
+			status.ShutdownSubscribers()
+		})
+	}
+}
+
+func TestJobStatusSubscribe_NotFound(t *testing.T) {
+	db := newInMemoryDB(t)
+	north, south := createNorthAndSouth(t, db)
+
+	handlers := []http.Handler{north, south}
+	for i, name := range allAPIs {
+		handler := handlers[i]
+		t.Run(name, func(t *testing.T) {
+			apitest.New().
+				Handler(handler).
+				Get("/api/wfx/v1/jobs/42/status/subscribe").
+				Expect(t).
+				Status(http.StatusNotFound).
+				Header("Content-Type", "application/json").
+				Assert(jsonpath.Equal(`$.errors[0].code`, "wfx.jobNotFound")).
+				End()
+		})
+	}
+}
+
+func TestJobStatusSubscribe_TerminalState(t *testing.T) {
+	db := newInMemoryDB(t)
+
+	wf := dau.DirectWorkflow()
+	_, _ = workflow.CreateWorkflow(context.Background(), db, wf)
+
+	tmpJob := model.Job{
+		ClientID: "foo",
+		Workflow: wf,
+		Status:   &model.JobStatus{State: "ACTIVATED"},
+	}
+
+	job, err := db.CreateJob(context.Background(), &tmpJob)
+	require.NoError(t, err)
+
+	jobPath := fmt.Sprintf("/api/wfx/v1/jobs/%s/status/subscribe", job.ID)
+
+	north, south := createNorthAndSouth(t, db)
+	handlers := []http.Handler{north, south}
+	for i, name := range allAPIs {
+		handler := handlers[i]
+		t.Run(name, func(t *testing.T) {
+			apitest.New().
+				Handler(handler).
+				Get(jobPath).
+				Expect(t).
+				Status(http.StatusBadRequest).
+				Header("Content-Type", "application/json").
+				Assert(jsonpath.Equal(`$.errors[0].code`, "wfx.jobTerminalState")).
+				End()
+		})
+	}
+}

api/northbound.go

@@ -25,6 +25,8 @@ import (
 	"github.com/siemens/wfx/internal/handler/job/tags"
 	"github.com/siemens/wfx/internal/handler/workflow"
 	"github.com/siemens/wfx/middleware/logging"
+	"github.com/siemens/wfx/middleware/responder/sse"
+	"github.com/siemens/wfx/middleware/responder/util"
 	"github.com/siemens/wfx/persistence"
 )
 
@@ -263,5 +265,19 @@ func NewNorthboundAPI(storage persistence.Storage) (*operations.WorkflowExecutor
 			return northbound.NewDeleteJobsIDTagsOK().WithPayload(tags)
 		})
 
+	serverAPI.NorthboundGetJobsIDStatusSubscribeHandler = northbound.GetJobsIDStatusSubscribeHandlerFunc(
+		func(params northbound.GetJobsIDStatusSubscribeParams) middleware.Responder {
+			ctx := params.HTTPRequest.Context()
+			eventChan, err := status.AddSubscriber(ctx, storage, params.ID)
+			if ftag.Get(err) == ftag.NotFound {
+				return util.ForceJSONResponse(http.StatusNotFound,
+					&model.ErrorResponse{Errors: []*model.Error{&JobNotFound}})
+			} else if ftag.Get(err) == ftag.InvalidArgument {
+				return util.ForceJSONResponse(http.StatusBadRequest,
+					&model.ErrorResponse{Errors: []*model.Error{&JobTerminalState}})
+			}
+			return sse.Responder(ctx, eventChan)
+		})
+
 	return serverAPI, nil
 }

api/southbound.go

@@ -25,6 +25,8 @@ import (
 	"github.com/siemens/wfx/internal/handler/job/tags"
 	"github.com/siemens/wfx/internal/handler/workflow"
 	"github.com/siemens/wfx/middleware/logging"
+	"github.com/siemens/wfx/middleware/responder/sse"
+	"github.com/siemens/wfx/middleware/responder/util"
 	"github.com/siemens/wfx/persistence"
 )
 
@@ -176,5 +178,19 @@ func NewSouthboundAPI(storage persistence.Storage) (*operations.WorkflowExecutor
 			return southbound.NewGetJobsIDTagsOK().WithPayload(tags)
 		})
 
+	serverAPI.SouthboundGetJobsIDStatusSubscribeHandler = southbound.GetJobsIDStatusSubscribeHandlerFunc(
+		func(params southbound.GetJobsIDStatusSubscribeParams) middleware.Responder {
+			ctx := params.HTTPRequest.Context()
+			eventChan, err := status.AddSubscriber(ctx, storage, params.ID)
+			if ftag.Get(err) == ftag.NotFound {
+				return util.ForceJSONResponse(http.StatusNotFound,
+					&model.ErrorResponse{Errors: []*model.Error{&JobNotFound}})
+			} else if ftag.Get(err) == ftag.InvalidArgument {
+				return util.ForceJSONResponse(http.StatusBadRequest,
+					&model.ErrorResponse{Errors: []*model.Error{&JobTerminalState}})
+			}
+			return sse.Responder(ctx, eventChan)
+		})
+
 	return serverAPI, nil
 }

api/workflow_test.go

@@ -213,8 +213,11 @@ func createNorthAndSouth(t *testing.T, db persistence.Storage) (http.Handler, ht
 }
 
 func persistJob(t *testing.T, db persistence.Storage) *model.Job {
-	wf, err := workflow.CreateWorkflow(context.Background(), db, dau.DirectWorkflow())
-	require.NoError(t, err)
+	wf := dau.DirectWorkflow()
+	if found, _ := workflow.GetWorkflow(context.Background(), db, wf.Name); found == nil {
+		_, err := workflow.CreateWorkflow(context.Background(), db, wf)
+		require.NoError(t, err)
+	}
 
 	jobReq := model.JobRequest{
 		ClientID: "foo",

cmd/wfx/cmd/root/cmd.go

@@ -30,6 +30,7 @@ import (
 	"github.com/rs/zerolog/log"
 	"github.com/siemens/wfx/cmd/wfx/metadata"
 	"github.com/siemens/wfx/internal/config"
+	"github.com/siemens/wfx/internal/handler/job/status"
 	"github.com/siemens/wfx/persistence"
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
@@ -255,7 +256,10 @@ Examples of tasks are installation of firmware or other types of commands issued
 			}
 		}
 
-		// Create a context with a timeout to allow outstanding requests to complete
+		// shut down (disconnect) subscribers otherwise we cannot stop the web server due to open connections
+		status.ShutdownSubscribers()
+
+		// create a context with a timeout to allow outstanding requests to complete
 		var timeout time.Duration
 		k.Read(func(k *koanf.Koanf) {
 			timeout = k.Duration(gracefulTimeoutFlag)

cmd/wfxctl/cmd/job/job.go

@@ -18,6 +18,7 @@ import (
 	"github.com/siemens/wfx/cmd/wfxctl/cmd/job/getstatus"
 	"github.com/siemens/wfx/cmd/wfxctl/cmd/job/gettags"
 	"github.com/siemens/wfx/cmd/wfxctl/cmd/job/query"
+	"github.com/siemens/wfx/cmd/wfxctl/cmd/job/subscribestatus"
 	"github.com/siemens/wfx/cmd/wfxctl/cmd/job/updatedefinition"
 	"github.com/siemens/wfx/cmd/wfxctl/cmd/job/updatestatus"
 	"github.com/spf13/cobra"
@@ -43,4 +44,5 @@ func init() {
 	Command.AddCommand(addtags.Command)
 	Command.AddCommand(deltags.Command)
 	Command.AddCommand(gettags.Command)
+	Command.AddCommand(subscribestatus.Command)
 }

cmd/wfxctl/cmd/job/subscribestatus/main_test.go

@@ -0,0 +1,19 @@
+package subscribestatus
+
+/*
+ * SPDX-FileCopyrightText: 2023 Siemens AG
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Author: Michael Adler <[email protected]>
+ */
+
+import (
+	"testing"
+
+	"go.uber.org/goleak"
+)
+
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m)
+}