feat: add support for docker over ssh tunnel

api/api/versions.json

@@ -3,7 +3,7 @@
     {
       "version": "v1",
       "status": "active",
-      "release_date": "2025-12-10T22:01:46.6664+05:30",
+      "release_date": "2025-12-11T15:39:34.260407+05:30",
       "end_of_life": "0001-01-01T00:00:00Z",
       "changes": [
         "Initial API version"

api/internal/features/container/controller/get_container.go

@@ -13,7 +13,7 @@ import (
 func (c *ContainerController) GetContainer(f fuego.ContextNoBody) (*shared_types.Response, error) {
 	containerID := f.PathParam("container_id")
 
-	containerInfo, err := c.dockerService.GetContainerById(containerID)
+	containerInfo, err := c.getDockerService(f.Request().Context()).GetContainerById(containerID)
 	if err != nil {
 		c.logger.Log(logger.Error, err.Error(), "")
 		return nil, fuego.HTTPError{

api/internal/features/container/controller/get_container_logs.go

@@ -23,7 +23,7 @@ func (c *ContainerController) GetContainerLogs(f fuego.ContextWithBody[types.Con
 		}
 	}
 
-	logsReader, err := c.dockerService.GetContainerLogs(req.ID, container.LogsOptions{
+	logsReader, err := c.getDockerService(f.Request().Context()).GetContainerLogs(req.ID, container.LogsOptions{
 		Follow:     req.Follow,
 		Tail:       strconv.Itoa(req.Tail),
 		Since:      req.Since,

api/internal/features/container/controller/init.go

@@ -10,14 +10,14 @@ import (
 	"github.com/raghavyuva/nixopus-api/internal/features/notification"
 	shared_storage "github.com/raghavyuva/nixopus-api/internal/storage"
 	shared_types "github.com/raghavyuva/nixopus-api/internal/types"
+	"github.com/raghavyuva/nixopus-api/internal/utils"
 )
 
 type ContainerController struct {
-	store         *shared_storage.Store
-	dockerService *docker.DockerService
-	ctx           context.Context
-	logger        logger.Logger
-	notification  *notification.NotificationManager
+	store        *shared_storage.Store
+	ctx          context.Context
+	logger       logger.Logger
+	notification *notification.NotificationManager
 }
 
 func NewContainerController(
@@ -27,16 +27,22 @@ func NewContainerController(
 	notificationManager *notification.NotificationManager,
 ) *ContainerController {
 	return &ContainerController{
-		store:         store,
-		dockerService: docker.NewDockerService(),
-		ctx:           ctx,
-		logger:        l,
-		notification:  notificationManager,
+		store:        store,
+		ctx:          ctx,
+		logger:       l,
+		notification: notificationManager,
 	}
 }
 
-func (c *ContainerController) isProtectedContainer(containerID string, action string) (*shared_types.Response, bool) {
-	details, err := c.dockerService.GetContainerById(containerID)
+// getDockerService creates a DockerService using the active server from the database
+// It extracts the organization ID from the request context (set by supertokens middleware)
+func (c *ContainerController) getDockerService(ctx context.Context) *docker.DockerService {
+	organizationID := utils.GetOrganizationIDFromContext(ctx)
+	return docker.NewDockerServiceWithServer(c.store.DB, ctx, organizationID)
+}
+
+func (c *ContainerController) isProtectedContainer(ctx context.Context, containerID string, action string) (*shared_types.Response, bool) {
+	details, err := c.getDockerService(ctx).GetContainerById(containerID)
 	if err != nil {
 		return nil, false
 	}

api/internal/features/container/controller/list_containers.go

@@ -1,6 +1,7 @@
 package controller
 
 import (
+	"context"
 	"net/http"
 	"sort"
 	"strconv"
@@ -19,7 +20,7 @@ func (c *ContainerController) ListContainers(fuegoCtx fuego.ContextNoBody) (*sha
 	params := parseContainerListParams(fuegoCtx.Request())
 
 	// Get pre-filtered summaries from Docker
-	containers, err := c.dockerService.ListContainers(container.ListOptions{
+	containers, err := c.getDockerService(fuegoCtx.Request().Context()).ListContainers(container.ListOptions{
 		All:     true,
 		Filters: buildDockerFilters(params),
 	})
@@ -34,7 +35,7 @@ func (c *ContainerController) ListContainers(fuegoCtx fuego.ContextNoBody) (*sha
 	rows := summarizeContainers(containers)
 	pageRows, totalCount := applySearchSortPaginate(rows, params)
 
-	result := c.appendContainerInfo(pageRows, containers)
+	result := c.appendContainerInfo(fuegoCtx.Request().Context(), pageRows, containers)
 
 	return &shared_types.Response{
 		Status:  "success",
@@ -186,10 +187,11 @@ func applySearchSortPaginate(rows []containertypes.ContainerListRow, p container
 	return rows[start:end], totalCount
 }
 
-func (c *ContainerController) appendContainerInfo(pageRows []containertypes.ContainerListRow, summaries []container.Summary) []containertypes.Container {
+func (c *ContainerController) appendContainerInfo(ctx context.Context, pageRows []containertypes.ContainerListRow, summaries []container.Summary) []containertypes.Container {
+	dockerService := c.getDockerService(ctx)
 	result := make([]containertypes.Container, 0, len(pageRows))
 	for _, r := range pageRows {
-		info, err := c.dockerService.GetContainerById(r.ID)
+		info, err := dockerService.GetContainerById(r.ID)
 		if err != nil {
 			c.logger.Log(logger.Error, "Error inspecting container", r.ID)
 			continue

api/internal/features/container/controller/list_images.go

@@ -26,9 +26,10 @@ func (c *ContainerController) ListImages(f fuego.ContextWithBody[ListImagesReque
 		}
 	}
 
+	dockerService := c.getDockerService(f.Request().Context())
 	filterArgs := filters.NewArgs()
 	if req.ContainerID != "" {
-		_, err := c.dockerService.GetContainerById(req.ContainerID)
+		_, err := dockerService.GetContainerById(req.ContainerID)
 		if err != nil {
 			return nil, fuego.HTTPError{
 				Err:    err,
@@ -45,7 +46,7 @@ func (c *ContainerController) ListImages(f fuego.ContextWithBody[ListImagesReque
 		filterArgs.Add("reference", pattern)
 	}
 
-	images := c.dockerService.ListAllImages(image.ListOptions{
+	images := dockerService.ListAllImages(image.ListOptions{
 		All:     req.All,
 		Filters: filterArgs,
 	})

api/internal/features/container/controller/prune_build_cache.go

@@ -22,7 +22,7 @@ func (c *ContainerController) PruneBuildCache(f fuego.ContextWithBody[PruneBuild
 			Status: http.StatusBadRequest,
 		}
 	}
-	err = c.dockerService.PruneBuildCache(types.BuildCachePruneOptions{
+	err = c.getDockerService(f.Request().Context()).PruneBuildCache(types.BuildCachePruneOptions{
 		All: req.All,
 	})
 	if err != nil {

api/internal/features/container/controller/prune_images.go

@@ -34,7 +34,7 @@ func (c *ContainerController) PruneImages(f fuego.ContextWithBody[PruneImagesReq
 		filterArgs.Add("dangling", "true")
 	}
 
-	pruneReport, err := c.dockerService.PruneImages(filterArgs)
+	pruneReport, err := c.getDockerService(f.Request().Context()).PruneImages(filterArgs)
 	if err != nil {
 		c.logger.Log(logger.Error, err.Error(), "")
 		return nil, fuego.HTTPError{

api/internal/features/container/controller/remove_container.go

@@ -11,12 +11,13 @@ import (
 
 func (c *ContainerController) RemoveContainer(f fuego.ContextNoBody) (*shared_types.Response, error) {
 	containerID := f.PathParam("container_id")
+	ctx := f.Request().Context()
 
-	if resp, skipped := c.isProtectedContainer(containerID, "remove"); skipped {
+	if resp, skipped := c.isProtectedContainer(ctx, containerID, "remove"); skipped {
 		return resp, nil
 	}
 
-	err := c.dockerService.RemoveContainer(containerID, container.RemoveOptions{Force: true})
+	err := c.getDockerService(ctx).RemoveContainer(containerID, container.RemoveOptions{Force: true})
 	if err != nil {
 		c.logger.Log(logger.Error, err.Error(), "")
 		return nil, fuego.HTTPError{

api/internal/features/container/controller/restart_container.go

@@ -12,7 +12,7 @@ import (
 func (c *ContainerController) RestartContainer(f fuego.ContextNoBody) (*shared_types.Response, error) {
 	containerID := f.PathParam("container_id")
 
-	err := c.dockerService.RestartContainer(containerID, container.StopOptions{})
+	err := c.getDockerService(f.Request().Context()).RestartContainer(containerID, container.StopOptions{})
 	if err != nil {
 		c.logger.Log(logger.Error, err.Error(), "")
 		return nil, fuego.HTTPError{

api/internal/features/container/controller/start_container.go

@@ -11,12 +11,13 @@ import (
 
 func (c *ContainerController) StartContainer(f fuego.ContextNoBody) (*shared_types.Response, error) {
 	containerID := f.PathParam("container_id")
+	ctx := f.Request().Context()
 
-	if resp, skipped := c.isProtectedContainer(containerID, "start"); skipped {
+	if resp, skipped := c.isProtectedContainer(ctx, containerID, "start"); skipped {
 		return resp, nil
 	}
 
-	err := c.dockerService.StartContainer(containerID, container.StartOptions{})
+	err := c.getDockerService(ctx).StartContainer(containerID, container.StartOptions{})
 	if err != nil {
 		c.logger.Log(logger.Error, err.Error(), "")
 		return nil, fuego.HTTPError{

api/internal/features/container/controller/stop_container.go

@@ -11,12 +11,13 @@ import (
 
 func (c *ContainerController) StopContainer(f fuego.ContextNoBody) (*shared_types.Response, error) {
 	containerID := f.PathParam("container_id")
+	ctx := f.Request().Context()
 
-	if resp, skipped := c.isProtectedContainer(containerID, "stop"); skipped {
+	if resp, skipped := c.isProtectedContainer(ctx, containerID, "stop"); skipped {
 		return resp, nil
 	}
 
-	err := c.dockerService.StopContainer(containerID, container.StopOptions{})
+	err := c.getDockerService(ctx).StopContainer(containerID, container.StopOptions{})
 	if err != nil {
 		c.logger.Log(logger.Error, err.Error(), "")
 		return nil, fuego.HTTPError{

api/internal/features/deploy/controller/init.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net/http"
 
+	"github.com/google/uuid"
 	"github.com/raghavyuva/nixopus-api/internal/features/deploy/docker"
 	"github.com/raghavyuva/nixopus-api/internal/features/deploy/service"
 	"github.com/raghavyuva/nixopus-api/internal/features/deploy/storage"
@@ -38,7 +39,7 @@ func NewDeployController(
 	notificationManager *notification.NotificationManager,
 ) *DeployController {
 	storage := storage.DeployStorage{DB: store.DB, Ctx: ctx}
-	docker_repo := docker.NewDockerService()
+	docker_repo := docker.NewDockerServiceWithServer(store.DB, ctx, uuid.Nil)
 	github_service := github_service.NewGithubConnectorService(store, ctx, l, &github_storage.GithubConnectorStorage{DB: store.DB, Ctx: ctx})
 	taskService := tasks.NewTaskService(&storage, l, docker_repo, github_service, store)
 	taskService.SetupCreateDeploymentQueue()

api/internal/features/deploy/docker/init.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/api/types/container"
@@ -14,15 +15,26 @@ import (
 	"github.com/docker/docker/api/types/swarm"
 	"github.com/docker/docker/api/types/volume"
 	"github.com/docker/docker/client"
+	"github.com/google/uuid"
 	v1 "github.com/opencontainers/image-spec/specs-go/v1"
 	"github.com/raghavyuva/nixopus-api/internal/features/logger"
 	"github.com/raghavyuva/nixopus-api/internal/features/ssh"
+	"github.com/raghavyuva/nixopus-api/internal/utils"
+	"github.com/uptrace/bun"
 )
 
 type DockerService struct {
-	Cli    *client.Client
-	Ctx    context.Context
-	logger logger.Logger
+	Cli       *client.Client
+	Ctx       context.Context
+	logger    logger.Logger
+	sshTunnel *SSHTunnel
+}
+
+type SSHTunnel struct {
+	localSocket string
+	sshClient   *ssh.SSH
+	listener    net.Listener
+	cleanup     func() error
 }
 
 type DockerRepository interface {
@@ -78,28 +90,79 @@ type DockerClient struct {
 }
 
 // NewDockerService creates a new instance of DockerService using the default docker client.
+// For server-specific SSH configuration, use NewDockerServiceWithServer instead.
 func NewDockerService() *DockerService {
-	client := NewDockerClient()
-	service := &DockerService{
-		Cli:    client,
-		Ctx:    context.Background(),
-		logger: logger.NewLogger(),
+	return NewDockerServiceWithServer(nil, nil, uuid.Nil)
+}
+
+// NewDockerServiceWithServer creates a new instance of DockerService using server SSH configuration when available.
+// If db, ctx, and organizationID are provided, it will use the active server from the database.
+// Otherwise, it falls back to the default SSH configuration.
+func NewDockerServiceWithServer(db *bun.DB, ctx context.Context, organizationID uuid.UUID) *DockerService {
+	lgr := logger.NewLogger()
+	cli, tunnel := newDockerClientWithOptionalSSHTunnel(lgr, db, ctx, organizationID)
+	svc := &DockerService{Cli: cli, Ctx: context.Background(), logger: lgr, sshTunnel: tunnel}
+
+	if !isClusterInitialized(svc.Cli) {
+		if err := svc.InitCluster(); err != nil {
+			svc.logger.Log(logger.Warning, "Failed to initialize cluster", err.Error())
+		} else {
+			svc.logger.Log(logger.Info, "Cluster initialized successfully", "")
+		}
+	} else {
+		svc.logger.Log(logger.Info, "Cluster already initialized", "")
 	}
 
-	// Initialize cluster if not already initialized, this should be run on master node only
-	// TODO: Add a check to see if the node is the master node
-	// WARNING: This should be thought again during multi-server architecture feature
-	if !isClusterInitialized(client) {
-		if err := service.InitCluster(); err != nil {
-			service.logger.Log(logger.Warning, "Failed to initialize cluster", err.Error())
+	return svc
+}
+
+func newDockerClientWithOptionalSSHTunnel(lgr logger.Logger, db *bun.DB, ctx context.Context, organizationID uuid.UUID) (*client.Client, *SSHTunnel) {
+	var sshClient *ssh.SSH
+
+	// Try to use server SSH configuration if db, ctx, and organizationID are available
+	if db != nil && ctx != nil {
+		// If organizationID is not provided, try to get it from context
+		if organizationID == uuid.Nil {
+			organizationID = utils.GetOrganizationIDFromContext(ctx)
+		}
+
+		// If we have organizationID, use server SSH configuration
+		if organizationID != uuid.Nil {
+			lgr.Log(logger.Info, "Using SSH with server configuration", fmt.Sprintf("organizationID: %s", organizationID.String()))
+			sshClient = ssh.NewSSHWithServer(db, ctx, organizationID)
 		} else {
-			service.logger.Log(logger.Info, "Cluster initialized successfully", "")
+			lgr.Log(logger.Info, "No organization ID available, using default SSH config", "")
+			sshClient = ssh.NewSSH()
 		}
 	} else {
-		service.logger.Log(logger.Info, "Cluster already initialized", "")
+		// Fall back to default SSH configuration
+		lgr.Log(logger.Info, "Using default SSH configuration", "")
+		sshClient = ssh.NewSSH()
 	}
 
-	return service
+	// Try to create an SSH tunnel to the remote Docker daemon socket, if it fails, use the local docker socket
+	tunnel, err := CreateSSHTunnel(sshClient, lgr)
+	if err != nil || tunnel == nil {
+		if err != nil {
+			lgr.Log(logger.Info, "SSH tunnel not established, using local docker socket", err.Error())
+		}
+		// If the SSH tunnel creation fails, use the local docker socket
+		lgr.Log(logger.Info, "Using local docker socket", "")
+		return NewDockerClient(), nil
+	}
+
+	host := fmt.Sprintf("unix://%s", tunnel.localSocket)
+	lgr.Log(logger.Info, "SSH tunnel established; using tunneled docker socket", host)
+	cli, cliErr := client.NewClientWithOpts(
+		client.WithHost(host),
+		client.WithAPIVersionNegotiation(),
+	)
+	if cliErr != nil {
+		lgr.Log(logger.Warning, "Failed to create docker client over SSH tunnel, using local", cliErr.Error())
+		return NewDockerClient(), nil
+	}
+	lgr.Log(logger.Info, "Docker client created over SSH tunnel", "")
+	return cli, tunnel
 }
 
 func isClusterInitialized(cli *client.Client) bool {