refact(orchestrator): make the apt upgrade as a package that handle the events

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Luca Rinaldi <lucarin@protonmail.com>

Author: 3 people

cmd/arduino-app-cli/daemon.go

@@ -8,7 +8,7 @@ import (
 	"time"
 
 	"github.com/arduino/arduino-app-cli/internal/api"
-	"github.com/arduino/arduino-app-cli/internal/api/handlers"
+	"github.com/arduino/arduino-app-cli/internal/apt"
 	"github.com/arduino/arduino-app-cli/internal/orchestrator"
 	"github.com/arduino/arduino-app-cli/pkg/httprecover"
 
@@ -44,7 +44,7 @@ func newDaemonCmd(docker *dockerClient.Client) *cobra.Command {
 func httpHandler(ctx context.Context, dockerClient *dockerClient.Client, daemonPort string) {
 	slog.Info("Starting HTTP server", slog.String("address", ":"+daemonPort))
 
-	apiSrv := api.NewHTTPRouter(dockerClient, Version, handlers.NewUpdateEventsBroker())
+	apiSrv := api.NewHTTPRouter(dockerClient, Version, apt.New())
 
 	corsMiddlware, err := cors.NewMiddleware(
 		cors.Config{

internal/api/api.go

@@ -5,23 +5,25 @@ import (
 	"net/http"
 
 	"github.com/arduino/arduino-app-cli/internal/api/handlers"
+	"github.com/arduino/arduino-app-cli/internal/apt"
 
 	dockerClient "github.com/docker/docker/client"
 )
 
 //go:embed docs
 var docsFS embed.FS
 
-func NewHTTPRouter(dockerClient *dockerClient.Client, version string, eventsBroker *handlers.UpdateEventsBroker) http.Handler {
+func NewHTTPRouter(dockerClient *dockerClient.Client, version string, aptClient *apt.Service) http.Handler {
 	mux := http.NewServeMux()
 
 	mux.Handle("GET /v1/version", handlers.HandlerVersion(version))
 	mux.Handle("GET /v1/config", handlers.HandleConfig())
 	mux.Handle("GET /v1/bricks", handlers.HandleBrickList())
 	mux.Handle("GET /v1/bricks/{brickID}", handlers.HandleBrickDetails())
-	mux.Handle("GET /v1/system/update/check", handlers.HandleCheckUpgradable())
-	mux.Handle("GET /v1/system/update/events", handlers.HandleUpdateEvents(eventsBroker))
-	mux.Handle("PUT /v1/system/update/apply", handlers.HandleUpdateApply(eventsBroker))
+
+	mux.Handle("GET /v1/system/update/check", handlers.HandleCheckUpgradable(aptClient))
+	mux.Handle("GET /v1/system/update/events", handlers.HandleUpdateEvents(aptClient))
+	mux.Handle("PUT /v1/system/update/apply", handlers.HandleUpdateApply(aptClient))
 
 	mux.Handle("GET /v1/models", handlers.HandleModelsList())
 	mux.Handle("GET /v1/models/{modelID}", handlers.HandlerModelByID())

internal/api/handlers/events.go

@@ -1,27 +1,27 @@
 package handlers
 
 import (
+	"errors"
 	"net/http"
 	"strings"
-	"sync/atomic"
 
 	"log/slog"
 
 	"go.bug.st/f"
 
-	"github.com/arduino/arduino-app-cli/internal/orchestrator"
+	"github.com/arduino/arduino-app-cli/internal/apt"
 	"github.com/arduino/arduino-app-cli/pkg/render"
 )
 
-var matchArduinoPackage = func(p orchestrator.UpgradablePackage) bool {
+var matchArduinoPackage = func(p apt.UpgradablePackage) bool {
 	return strings.HasPrefix(p.Name, "arduino-")
 }
 
-var matchAllPackages = func(p orchestrator.UpgradablePackage) bool {
+var matchAllPackages = func(p apt.UpgradablePackage) bool {
 	return true
 }
 
-func HandleCheckUpgradable() http.HandlerFunc {
+func HandleCheckUpgradable(aptClient *apt.Service) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		queryParams := r.URL.Query()
 
@@ -35,8 +35,12 @@ func HandleCheckUpgradable() http.HandlerFunc {
 			filterFunc = matchArduinoPackage
 		}
 
-		pkgs, err := orchestrator.GetUpgradablePackages(r.Context(), filterFunc)
+		pkgs, err := aptClient.ListUpgradablePackages(r.Context(), filterFunc)
 		if err != nil {
+			if errors.Is(err, apt.ErrOperationAlreadyInProgress) {
+				render.EncodeResponse(w, http.StatusConflict, err.Error())
+				return
+			}
 			render.EncodeResponse(w, http.StatusBadRequest, "Error checking for upgradable packages: "+err.Error())
 			return
 		}
@@ -53,25 +57,11 @@ func HandleCheckUpgradable() http.HandlerFunc {
 }
 
 type UpdateCheckResult struct {
-	Packages []orchestrator.UpgradablePackage `json:"packages"`
+	Packages []apt.UpgradablePackage `json:"packages"`
 }
 
-var inProgress atomic.Bool
-
-func HandleUpdateApply(eventsBroker *UpdateEventsBroker) http.HandlerFunc {
+func HandleUpdateApply(aptClient *apt.Service) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
-		// Check if an upgrade is already in progress
-		if inProgress.Load() {
-			render.EncodeResponse(w, http.StatusConflict, "Upgrade already in progress")
-			return
-		}
-
-		// Set upgrade in progress
-		if !inProgress.CompareAndSwap(false, true) {
-			render.EncodeResponse(w, http.StatusConflict, "Upgrade already in progress")
-			return
-		}
-
 		queryParams := r.URL.Query()
 		onlyArduinoPackages := false
 		if val := queryParams.Get("only-arduino"); val != "" {
@@ -83,8 +73,12 @@ func HandleUpdateApply(eventsBroker *UpdateEventsBroker) http.HandlerFunc {
 			filterFunc = matchArduinoPackage
 		}
 
-		pkgs, err := orchestrator.GetUpgradablePackages(r.Context(), filterFunc)
+		pkgs, err := aptClient.ListUpgradablePackages(r.Context(), filterFunc)
 		if err != nil {
+			if errors.Is(err, apt.ErrOperationAlreadyInProgress) {
+				render.EncodeResponse(w, http.StatusConflict, err.Error())
+				return
+			}
 			slog.Error("Unable to get upgradable packages", slog.String("error", err.Error()))
 			render.EncodeResponse(w, http.StatusInternalServerError, "Error checking for upgradable packages")
 			return
@@ -95,41 +89,25 @@ func HandleUpdateApply(eventsBroker *UpdateEventsBroker) http.HandlerFunc {
 			return
 		}
 
-		go func() {
-			defer inProgress.Store(false)
-
-			names := f.Map(pkgs, func(p orchestrator.UpgradablePackage) string {
-				return p.Name
-			})
-
-			eventsBroker.PublishLog("Upgrading: " + strings.Join(names, ", "))
-
-			iter, err := orchestrator.RunUpgradeCommand(r.Context(), names)
-			if err != nil {
-				slog.Error("Error running upgrade command", slog.String("error", err.Error()))
-				eventsBroker.PublishError(render.SSEErrorData{Message: "failed to upgrade the packages"})
-				return
-			}
-
-			for item := range iter {
-				eventsBroker.PublishLog(item)
-			}
-
-			eventsBroker.Restarting()
+		names := f.Map(pkgs, func(p apt.UpgradablePackage) string {
+			return p.Name
+		})
 
-			err = orchestrator.RestartServices(r.Context())
-			if err != nil {
-				slog.Error("Error restarting services", slog.String("error", err.Error()))
-				eventsBroker.PublishError(render.SSEErrorData{Message: "failed to restart services"})
+		err = aptClient.UpgradePackages(names)
+		if err != nil {
+			if errors.Is(err, apt.ErrOperationAlreadyInProgress) {
+				render.EncodeResponse(w, http.StatusConflict, err.Error())
 				return
 			}
-		}()
+			render.EncodeResponse(w, http.StatusInternalServerError, "Error upgrading packages")
+			return
+		}
 
 		render.EncodeResponse(w, http.StatusAccepted, "Upgrade started")
 	}
 }
 
-func HandleUpdateEvents(eventsBroker *UpdateEventsBroker) http.HandlerFunc {
+func HandleUpdateEvents(aptClient *apt.Service) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		sseStream, err := render.NewSSEStream(r.Context(), w)
 		if err != nil {
@@ -139,13 +117,28 @@ func HandleUpdateEvents(eventsBroker *UpdateEventsBroker) http.HandlerFunc {
 		}
 		defer sseStream.Close()
 
-		ch := eventsBroker.Subscribe()
-		defer eventsBroker.Unsubscribe(ch)
+		ch := aptClient.Subscribe()
+		defer aptClient.Unsubscribe(ch)
 
 		for {
 			select {
-			case event := <-ch:
-				sseStream.Send(event)
+			case event, ok := <-ch:
+				if !ok {
+					slog.Info("APT event channel closed, stopping SSE stream")
+					return
+				}
+				if event.Type == apt.ErrorEvent {
+					sseStream.SendError(render.SSEErrorData{
+						Code:    render.InternalServiceErr,
+						Message: event.Data,
+					})
+				} else {
+					sseStream.Send(render.SSEEvent{
+						Type: event.Type.String(),
+						Data: event.Data,
+					})
+				}
+
 			case <-r.Context().Done():
 				return
 			}

internal/api/handlers/update.go

@@ -0,0 +1,33 @@
+package apt
+
+// EventType defines the type of upgrade event.
+type EventType int
+
+const (
+	UpgradeLineEvent EventType = iota
+	StartEvent
+	RestartEvent
+	ErrorEvent
+)
+
+// Event represents a single event in the upgrade process.
+type Event struct {
+	Type EventType
+	Data string
+	Err  error // Optional error field for error events
+}
+
+func (t EventType) String() string {
+	switch t {
+	case UpgradeLineEvent:
+		return "log"
+	case RestartEvent:
+		return "restarting"
+	case StartEvent:
+		return "starting"
+	case ErrorEvent:
+		return "error"
+	default:
+		panic("unreachable")
+	}
+}