Pull docker containers events for each App

Co-authored-by: lucarin91 <lucarin@protonmail.com>
Co-authored-by: Your Name <youremail@example.com>
Co-authored-by: Davide <davideneri18@gmail.com>

Author: 4 people

cmd/gendoc/docs.go

@@ -520,6 +520,33 @@ Contains a JSON object with the details of an error.
 				{StatusCode: http.StatusInternalServerError, Reference: "#/components/responses/InternalServerError"},
 			},
 		},
+		{
+			OperationId: "getAppsEvents",
+			Method:      http.MethodGet,
+			Path:        "/v1/apps/events",
+			CustomSuccessResponse: &CustomResponseDef{
+				ContentType:   "text/event-stream",
+				DataStructure: orchestrator.LogMessage{},
+			},
+			Description: `A stream of Server-Sent Events (SSE) that notifies the apps status.
+The client will receive events formatted as follows:
+
+**Event 'app'**:
+Contains a JSON object with an informational message.
+'event: app'
+'data: {"id":"dXNlcjpleGFtcG","name":"example-app-for-status-events","description":"My app description","icon":"💻","status":"running","example":false,"default":false}'
+
+**Event 'error'**:
+Contains a JSON object with the details of an error.
+'event: error'
+'data: {"code":"INTERNAL_SERVER_ERROR","message":"An error occurred during operation"}'
+`,
+			Summary: "Get application events",
+			Tags:    []Tag{ApplicationTag},
+			PossibleErrors: []ErrorResponse{
+				{StatusCode: http.StatusInternalServerError, Reference: "#/components/responses/InternalServerError"},
+			},
+		},
 		{
 			OperationId: "getAppEvents",
 			Method:      http.MethodGet,
@@ -532,7 +559,7 @@ Contains a JSON object with the details of an error.
 				DataStructure: orchestrator.LogMessage{},
 			},
 			Description: "Returns events for a specific app ",
-			Summary:     "Get application events ",
+			Summary:     "Get application events",
 			Tags:        []Tag{ApplicationTag},
 			PossibleErrors: []ErrorResponse{
 				{StatusCode: http.StatusInternalServerError, Reference: "#/components/responses/InternalServerError"},

internal/api/docs/openapi.yaml

@@ -409,7 +409,7 @@ paths:
           description: OK
         "500":
           $ref: '#/components/responses/InternalServerError'
-      summary: 'Get application events '
+      summary: Get application events
       tags:
       - Application
   /v1/apps/{id}/logs:
@@ -541,6 +541,28 @@ paths:
       summary: Stop an existing app/example
       tags:
       - Application
+  /v1/apps/events:
+    get:
+      description: "A stream of Server-Sent Events (SSE) that notifies the apps status.\nThe
+        client will receive events formatted as follows:\n\n**Event 'app'**:\nContains
+        a JSON object with an informational message.\n'event: app'\n'data: {\"id\":\"dXNlcjpleGFtcG\",\"name\":\"example-app-for-status-events\",\"description\":\"My
+        app description\",\"icon\":\"\U0001F4BB\",\"status\":\"running\",\"example\":false,\"default\":false}'\n\n**Event
+        'error'**:\nContains a JSON object with the details of an error.\n'event:
+        error'\n'data: {\"code\":\"INTERNAL_SERVER_ERROR\",\"message\":\"An error
+        occurred during operation\"}'\n"
+      operationId: getAppsEvents
+      responses:
+        "200":
+          content:
+            text/event-stream:
+              schema:
+                type: string
+          description: OK
+        "500":
+          $ref: '#/components/responses/InternalServerError'
+      summary: Get application events
+      tags:
+      - Application
   /v1/bricks:
     get:
       description: Returns all the existing bricks. Bricks that are ready to use are

internal/api/handlers/app_status.go

@@ -3,7 +3,6 @@ package handlers
 import (
 	"log/slog"
 	"net/http"
-	"time"
 
 	"github.com/docker/cli/cli/command"
 
@@ -28,30 +27,12 @@ func HandlerAppStatus(
 		}
 		defer sseStream.Close()
 
-		// TODO: maybe we should limit the size of this cache?
-		stateCache := make(map[string]orchestrator.Status)
-
-		for {
-			apps, err := orchestrator.AppStatus(r.Context(), cfg, dockerCli.Client(), idProvider)
+		for appStatus, err := range orchestrator.AppStatusEvents(r.Context(), cfg, dockerCli, idProvider) {
 			if err != nil {
-				slog.Error("Unable to get apps status", slog.String("error", err.Error()))
-				render.EncodeResponse(w, http.StatusInternalServerError, models.ErrorResponse{Details: "unable to get apps status"})
-				return
-			}
-			for _, a := range apps {
-				status, exist := stateCache[a.ID.String()]
-				if !exist || status != a.Status {
-					sseStream.Send(render.SSEEvent{Type: "app", Data: a})
-				}
-
-				stateCache[a.ID.String()] = a.Status
-			}
-
-			select {
-			case <-r.Context().Done():
-				return
-			case <-time.After(1 * time.Second):
+				sseStream.SendError(render.SSEErrorData{Code: render.InternalServiceErr, Message: err.Error()})
+				continue
 			}
+			sseStream.Send(render.SSEEvent{Type: "app", Data: appStatus})
 		}
 	}
 }

internal/e2e/client/client.gen.go

@@ -1,9 +1,12 @@
 package daemon
 
 import (
+	"bufio"
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
+	"log"
 	"net/http"
 	"strconv"
 	"strings"
@@ -852,6 +855,66 @@ func TestAppPorts(t *testing.T) {
 	})
 }
 
+func TestGetAppsStatusEvents(t *testing.T) {
+
+	httpClient := GetHttpclient(t)
+	appName := "example-app-for-status-events"
+
+	t.Run("StreamAppEvents_Success", func(t *testing.T) {
+		eventsResp, err := httpClient.GetAppsEvents(t.Context())
+		require.NoError(t, err)
+		require.Equal(t, http.StatusOK, eventsResp.StatusCode)
+		defer eventsResp.Body.Close()
+		scanner := bufio.NewScanner(eventsResp.Body)
+		go func() {
+			createResp, err := httpClient.CreateAppWithResponse(
+				t.Context(),
+				&client.CreateAppParams{SkipSketch: f.Ptr(true)},
+				client.CreateAppRequest{
+					Icon:        f.Ptr("💻"),
+					Name:        appName,
+					Description: f.Ptr("My app description"),
+				},
+			)
+			require.NoError(t, err)
+			require.Equal(t, http.StatusCreated, createResp.StatusCode())
+			appResponse, err := httpClient.StartAppWithResponse(t.Context(), *createResp.JSON201.Id)
+			require.NoError(t, err)
+			require.Equal(t, http.StatusOK, appResponse.StatusCode())
+		}()
+		var lastStatuses []string
+		for scanner.Scan() {
+			line := scanner.Text()
+			if line == "" {
+				continue
+			}
+			eventData := strings.TrimPrefix(line, "data: ")
+			if strings.Contains(eventData, `"name":"`+appName+`"`) {
+				var event map[string]interface{}
+				err := json.Unmarshal([]byte(eventData), &event)
+				require.NoError(t, err)
+				status, ok := event["status"].(string)
+				require.True(t, ok, "status field missing or not string")
+				lastStatuses = append(lastStatuses, status)
+				if len(lastStatuses) > 3 {
+					lastStatuses = lastStatuses[1:]
+				}
+				if len(lastStatuses) == 3 &&
+					lastStatuses[0] == "stopped" &&
+					lastStatuses[1] == "running" &&
+					lastStatuses[2] == "stopped" {
+					fmt.Println("Desired sequence received, terminating test")
+					return
+				}
+			}
+		}
+		if err := scanner.Err(); err != nil {
+			log.Fatal(fmt.Errorf("error reading event stream: %w", err))
+		}
+
+	})
+}
+
 func TestAppList(t *testing.T) {
 	httpClient := GetHttpclient(t)
 	t.Run("AppListEmpty_success", func(t *testing.T) {
@@ -880,7 +943,6 @@ func TestAppList(t *testing.T) {
 		require.Equal(t, http.StatusOK, resp.StatusCode())
 		require.NotNil(t, resp.JSON200)
 		require.Equal(t, len(*resp.JSON200.Apps), expectedAppNumber, "The apps list should contain "+strconv.Itoa(expectedAppNumber)+" elements")
-
 	})
 
 	t.Run("AppListDefault_success", func(t *testing.T) {

internal/e2e/daemon/app_test.go

@@ -8,5 +8,4 @@ const (
 	noExistingApp                  = "dXNlcjp0ZXN0LWFwcAw"
 	malformedAppId                 = "this-is-definitely-not-base64"
 	noExisitingExample             = "ZXhhbXBsZXM6anVzdGJsaW5f"
-	existingExampleAppId           = "ZXhhbXBsZXM6YWlyLXF1YWxpdHktbW9uaXRvcmluZw"
 )