Repository: digitalocean/doctl
Pull request #1612 (2024-11-13 12:05:04)

APPS-9859 Add doctl apps console command to start a console session to a component
This PR adds a `doctl apps console` command to start a console session to a component.

To try this out:
```
go build -o /tmp/doctl cmd/doctl/main.go
/tmp/doctl apps console <app-id> <component-name>
```

Note: The tests for this PR will pass only after https://github.com/digitalocean/godo/pull/754 is released and go.mod updated here.


Patch:
diff --git a/commands/apps.go b/commands/apps.go
index def9e921..8a78ea94 100644
--- a/commands/apps.go
+++ b/commands/apps.go
@@ -16,12 +16,14 @@ package commands
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
 	"net/url"
 	"os"
+	"os/signal"
 	"strings"
 	"time"
 
@@ -29,9 +31,11 @@ import (
 	"github.com/digitalocean/doctl/commands/displayers"
 	"github.com/digitalocean/doctl/do"
 	"github.com/digitalocean/doctl/internal/apps"
+	"github.com/digitalocean/doctl/pkg/terminal"
 	"github.com/digitalocean/godo"
 	multierror "github.com/hashicorp/go-multierror"
 	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
 	"sigs.k8s.io/yaml"
 )
 
@@ -191,6 +195,19 @@ For more information about logs, see [How to View Logs](https://www.digitalocean
 
 	logs.Example = `The following example retrieves the build logs for the app with the ID ` + "`" + `f81d4fae-7dec-11d0-a765-00a0c91e6bf6` + "`" + ` and the component ` + "`" + `web` + "`" + `: doctl apps logs f81d4fae-7dec-11d0-a765-00a0c91e6bf6 web --type build`
 
+	console := CmdBuilder(
+		cmd,
+		RunAppsConsole,
+		"console <app id> <component name>",
+		"Starts a console session",
+		`Instantiates a console session for a component of an app.`,
+		Writer,
+		aliasOpt("l"),
+	)
+	AddStringFlag(console, doctl.ArgAppDeployment, "", "", "Starts a console session for a specific deployment ID. Defaults to current deployment.")
+
+	console.Example = `The following example initiates a console session for the app with the ID ` + "`" + `f81d4fae-7dec-11d0-a765-00a0c91e6bf6` + "`" + ` and the component ` + "`" + `web` + "`" + `: doctl apps console f81d4fae-7dec-11d0-a765-00a0c91e6bf6 web`
+
 	listRegions := CmdBuilder(
 		cmd,
 		RunAppsListRegions,
@@ -662,8 +679,10 @@ func RunAppsGetLogs(c *CmdConfig) error {
 			url.Scheme = "wss"
 		}
 
-		listener := c.Doit.Listen(url, token, schemaFunc, c.Out)
-		err = listener.Start()
+		listener := c.Doit.Listen(url, token, schemaFunc, c.Out, nil)
+		ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
+		defer cancel()
+		err = listener.Listen(ctx)
 		if err != nil {
 			return err
 		}
@@ -695,6 +714,117 @@ func RunAppsGetLogs(c *CmdConfig) error {
 	return nil
 }
 
+// RunAppsConsole initiates a console session for an app.
+func RunAppsConsole(c *CmdConfig) error {
+	if len(c.Args) < 2 {
+		return doctl.NewMissingArgsErr(c.NS)
+	}
+	appID := c.Args[0]
+	component := c.Args[1]
+
+	deploymentID, err := c.Doit.GetString(c.NS, doctl.ArgAppDeployment)
+	if err != nil {
+		return err
+	}
+
+	execResp, err := c.Apps().GetExec(appID, deploymentID, component)
+	if err != nil {
+		return err
+	}
+	url, err := url.Parse(execResp.URL)
+	if err != nil {
+		return err
+	}
+	token := url.Query().Get("token")
+
+	schemaFunc := func(message []byte) (io.Reader, error) {
+		data := struct {
+			Data string `json:"data"`
+		}{}
+		err = json.Unmarshal(message, &data)
+		if err != nil {
+			return nil, err
+		}
+		r := strings.NewReader(data.Data)
+		return r, nil
+	}
+
+	inputCh := make(chan []byte)
+
+	listener := c.Doit.Listen(url, token, schemaFunc, c.Out, inputCh)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	grp, ctx := errgroup.WithContext(ctx)
+
+	term := c.Doit.Terminal()
+	stdinCh := make(chan string)
+	restoreTerminal, err := term.ReadRawStdin(ctx, stdinCh)
+	if err != nil {
+		return err
+	}
+	defer restoreTerminal()
+
+	resizeEvents := make(chan terminal.TerminalSize)
+	grp.Go(func() error {
+		return term.MonitorResizeEvents(ctx, resizeEvents)
+	})
+
+	grp.Go(func() error {
+		keepaliveTicker := time.NewTicker(30 * time.Second)
+		defer keepaliveTicker.Stop()
+		type stdinOp struct {
+			Op   string `json:"op"`
+			Data string `json:"data"`
+		}
+		type resizeOp struct {
+			Op     string `json:"op"`
+			Width  int    `json:"width"`
+			Height int    `json:"height"`
+		}
+		for {
+			select {
+			case <-ctx.Done():
+				return nil
+			case in := <-stdinCh:
+				b, err := json.Marshal(stdinOp{Op: "stdin", Data: in})
+				if err != nil {
+					return fmt.Errorf("error encoding stdin: %v", err)
+				}
+				inputCh <- b
+			case <-keepaliveTicker.C:
+				b, err := json.Marshal(stdinOp{Op: "stdin", Data: ""})
+				if err != nil {
+					return fmt.Errorf("error encoding keepalive event: %v", err)
+				}
+				inputCh <- b
+			case ev := <-resizeEvents:
+				b, err := json.Marshal(resizeOp{Op: "resize", Width: ev.Width, Height: ev.Height})
+				if err != nil {
+					return fmt.Errorf("error encoding resize event: %v", err)
+				}
+				inputCh <- b
+			}
+		}
+	})
+
+	grp.Go(func() error {
+		err = listener.Listen(ctx)
+		if err != nil {
+			return err
+		}
+		cancel() // cancel the context to stop the other goroutines
+		return nil
+	})
+
+	if err := grp.Wait(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 // RunAppsPropose proposes an app spec
 func RunAppsPropose(c *CmdConfig) error {
 	appID, err := c.Doit.GetString(c.NS, doctl.ArgApp)
diff --git a/do/apps.go b/do/apps.go
index d236d6cb..184d2145 100644
--- a/do/apps.go
+++ b/do/apps.go
@@ -33,6 +33,7 @@ type AppsService interface {
 	ListDeployments(appID string) ([]*godo.Deployment, error)
 
 	GetLogs(appID, deploymentID, component string, logType godo.AppLogType, follow bool, tail int) (*godo.AppLogs, error)
+	GetExec(appID, deploymentID, component string) (*godo.AppExec, error)
 
 	ListRegions() ([]*godo.AppRegion, error)
 
@@ -186,6 +187,14 @@ func (s *appsService) GetLogs(appID, deploymentID, component string, logType god
 	return logs, nil
 }
 
+func (s *appsService) GetExec(appID, deploymentID, component string) (*godo.AppExec, error) {
+	exec, _, err := s.client.Apps.GetExec(s.ctx, appID, deploymentID, component)
+	if err != nil {
+		return nil, err
+	}
+	return exec, nil
+}
+
 func (s *appsService) ListRegions() ([]*godo.AppRegion, error) {
 	regions, _, err := s.client.Apps.ListRegions(s.ctx)
 	if err != nil {
diff --git a/do/mocks/AppsService.go b/do/mocks/AppsService.go
index f0774d7d..e71d8ef1 100644
--- a/do/mocks/AppsService.go
+++ b/do/mocks/AppsService.go
@@ -114,6 +114,21 @@ func (mr *MockAppsServiceMockRecorder) GetDeployment(appID, deploymentID any) *g
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetDeployment", reflect.TypeOf((*MockAppsService)(nil).GetDeployment), appID, deploymentID)
 }
 
+// GetExec mocks base method.
+func (m *MockAppsService) GetExec(appID, deploymentID, component string) (*godo.AppExec, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetExec", appID, deploymentID, component)
+	ret0, _ := ret[0].(*godo.AppExec)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetExec indicates an expected call of GetExec.
+func (mr *MockAppsServiceMockRecorder) GetExec(appID, deploymentID, component any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetExec", reflect.TypeOf((*MockAppsService)(nil).GetExec), appID, deploymentID, component)
+}
+
 // GetInstanceSize mocks base method.
 func (m *MockAppsService) GetInstanceSize(slug string) (*godo.AppInstanceSize, error) {
 	m.ctrl.T.Helper()
diff --git a/do/mocks/Listen.go b/do/mocks/Listen.go
index 94882a4e..59e22e02 100644
--- a/do/mocks/Listen.go
+++ b/do/mocks/Listen.go
@@ -10,6 +10,7 @@
 package mocks
 
 import (
+	context "context"
 	reflect "reflect"
 
 	gomock "go.uber.org/mock/gomock"
@@ -39,28 +40,16 @@ func (m *MockListenerService) EXPECT() *MockListenerServiceMockRecorder {
 	return m.recorder
 }
 
-// Start mocks base method.
-func (m *MockListenerService) Start() error {
+// Listen mocks base method.
+func (m *MockListenerService) Listen(ctx context.Context) error {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "Start")
+	ret := m.ctrl.Call(m, "Listen", ctx)
 	ret0, _ := ret[0].(error)
 	return ret0
 }
 
-// Start indicates an expected call of Start.
-func (mr *MockListenerServiceMockRecorder) Start() *gomock.Call {
+// Listen indicates an expected call of Listen.
+func (mr *MockListenerServiceMockRecorder) Listen(ctx any) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Start", reflect.TypeOf((*MockListenerService)(nil).Start))
-}
-
-// Stop mocks base method.
-func (m *MockListenerService) Stop() {
-	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "Stop")
-}
-
-// Stop indicates an expected call of Stop.
-func (mr *MockListenerServiceMockRecorder) Stop() *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stop", reflect.TypeOf((*MockListenerService)(nil).Stop))
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Listen", reflect.TypeOf((*MockListenerService)(nil).Listen), ctx)
 }
diff --git a/do/mocks/Terminal.go b/do/mocks/Terminal.go
new file mode 100644
index 00000000..9129e339
--- /dev/null
+++ b/do/mocks/Terminal.go
@@ -0,0 +1,71 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: ../pkg/terminal/terminal.go
+//
+// Generated by this command:
+//
+//	mockgen -source ../pkg/terminal/terminal.go -package=mocks Terminal
+//
+
+// Package mocks is a generated GoMock package.
+package mocks
+
+import (
+	context "context"
+	reflect "reflect"
+
+	terminal "github.com/digitalocean/doctl/pkg/terminal"
+	gomock "go.uber.org/mock/gomock"
+)
+
+// MockTerminal is a mock of Terminal interface.
+type MockTerminal struct {
+	ctrl     *gomock.Controller
+	recorder *MockTerminalMockRecorder
+	isgomock struct{}
+}
+
+// MockTerminalMockRecorder is the mock recorder for MockTerminal.
+type MockTerminalMockRecorder struct {
+	mock *MockTerminal
+}
+
+// NewMockTerminal creates a new mock instance.
+func NewMockTerminal(ctrl *gomock.Controller) *MockTerminal {
+	mock := &MockTerminal{ctrl: ctrl}
+	mock.recorder = &MockTerminalMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *MockTerminal) EXPECT() *MockTerminalMockRecorder {
+	return m.recorder
+}
+
+// MonitorResizeEvents mocks base method.
+func (m *MockTerminal) MonitorResizeEvents(ctx context.Context, resizeEvents chan<- terminal.TerminalSize) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "MonitorResizeEvents", ctx, resizeEvents)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// MonitorResizeEvents indicates an expected call of MonitorResizeEvents.
+func (mr *MockTerminalMockRecorder) MonitorResizeEvents(ctx, resizeEvents any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MonitorResizeEvents", reflect.TypeOf((*MockTerminal)(nil).MonitorResizeEvents), ctx, resizeEvents)
+}
+
+// ReadRawStdin mocks base method.
+func (m *MockTerminal) ReadRawStdin(ctx context.Context, stdinCh chan<- string) (func(), error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ReadRawStdin", ctx, stdinCh)
+	ret0, _ := ret[0].(func())
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ReadRawStdin indicates an expected call of ReadRawStdin.
+func (mr *MockTerminalMockRecorder) ReadRawStdin(ctx, stdinCh any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReadRawStdin", reflect.TypeOf((*MockTerminal)(nil).ReadRawStdin), ctx, stdinCh)
+}
diff --git a/doit.go b/doit.go
index 818ff72c..ef6ff0ef 100644
--- a/doit.go
+++ b/doit.go
@@ -34,6 +34,7 @@ import (
 	"github.com/digitalocean/doctl/pkg/listen"
 	"github.com/digitalocean/doctl/pkg/runner"
 	"github.com/digitalocean/doctl/pkg/ssh"
+	"github.com/digitalocean/doctl/pkg/terminal"
 	"github.com/digitalocean/godo"
 	"github.com/docker/docker/client"
 	"github.com/spf13/viper"
@@ -210,7 +211,8 @@ type Config interface {
 	GetGodoClient(trace, allowRetries bool, accessToken string) (*godo.Client, error)
 	GetDockerEngineClient() (builder.DockerEngineClient, error)
 	SSH(user, host, keyPath string, port int, opts ssh.Options) runner.Runner
-	Listen(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer) listen.ListenerService
+	Listen(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, inCh <-chan []byte) listen.ListenerService
+	Terminal() terminal.Terminal
 	Set(ns, key string, val any)
 	IsSet(key string) bool
 	GetString(ns, key string) (string, error)
@@ -326,8 +328,13 @@ func (c *LiveConfig) SSH(user, host, keyPath string, port int, opts ssh.Options)
 }
 
 // Listen creates a websocket connection
-func (c *LiveConfig) Listen(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer) listen.ListenerService {
-	return listen.NewListener(url, token, schemaFunc, out)
+func (c *LiveConfig) Listen(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, inCh <-chan []byte) listen.ListenerService {
+	return listen.NewListener(url, token, schemaFunc, out, inCh)
+}
+
+// Terminal returns a terminal.
+func (c *LiveConfig) Terminal() terminal.Terminal {
+	return terminal.New()
 }
 
 // Set sets a config key.
@@ -483,7 +490,8 @@ func isRequired(key string) bool {
 // TestConfig is an implementation of Config for testing.
 type TestConfig struct {
 	SSHFn              func(user, host, keyPath string, port int, opts ssh.Options) runner.Runner
-	ListenFn           func(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer) listen.ListenerService
+	ListenFn           func(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, inCh <-chan []byte) listen.ListenerService
+	TerminalFn         func() terminal.Terminal
 	v                  *viper.Viper
 	IsSetMap           map[string]bool
 	DockerEngineClient builder.DockerEngineClient
@@ -497,9 +505,12 @@ func NewTestConfig() *TestConfig {
 		SSHFn: func(u, h, kp string, p int, opts ssh.Options) runner.Runner {
 			return &MockRunner{}
 		},
-		ListenFn: func(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer) listen.ListenerService {
+		ListenFn: func(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, inCh <-chan []byte) listen.ListenerService {
 			return &MockListener{}
 		},
+		TerminalFn: func() terminal.Terminal {
+			return &MockTerminal{}
+		},
 		v:        viper.New(),
 		IsSetMap: make(map[string]bool),
 	}
@@ -523,8 +534,13 @@ func (c *TestConfig) SSH(user, host, keyPath string, port int, opts ssh.Options)
 }
 
 // Listen returns a mock websocket listener
-func (c *TestConfig) Listen(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer) listen.ListenerService {
-	return c.ListenFn(url, token, schemaFunc, out)
+func (c *TestConfig) Listen(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, inCh <-chan []byte) listen.ListenerService {
+	return c.ListenFn(url, token, schemaFunc, out, inCh)
+}
+
+// Terminal returns a mock terminal.
+func (c *TestConfig) Terminal() terminal.Terminal {
+	return c.TerminalFn()
 }
 
 // Set sets a config key.
diff --git a/go.mod b/go.mod
index 91b3ac28..78549b1f 100644
--- a/go.mod
+++ b/go.mod
@@ -5,7 +5,7 @@ go 1.22
 require (
 	github.com/blang/semver v3.5.1+incompatible
 	github.com/creack/pty v1.1.21
-	github.com/digitalocean/godo v1.128.1-0.20241025145008-2654a9d1e887
+	github.com/digitalocean/godo v1.129.1-0.20241113153826-8f5ac1bc9671
 	github.com/docker/cli v24.0.5+incompatible
 	github.com/docker/docker v25.0.6+incompatible
 	github.com/docker/docker-credential-helpers v0.7.0 // indirect
@@ -31,7 +31,7 @@ require (
 	golang.org/x/crypto v0.22.0
 	golang.org/x/net v0.24.0 // indirect
 	golang.org/x/oauth2 v0.23.0
-	golang.org/x/sys v0.25.0 // indirect
+	golang.org/x/sys v0.25.0
 	gopkg.in/yaml.v2 v2.4.0
 	k8s.io/api v0.26.2
 	k8s.io/apimachinery v0.26.2
@@ -53,6 +53,7 @@ require (
 	github.com/muesli/termenv v0.12.0
 	github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8
 	go.uber.org/mock v0.2.0
+	golang.org/x/sync v0.3.0
 	golang.org/x/term v0.19.0
 	gopkg.in/yaml.v3 v3.0.1
 )
diff --git a/go.sum b/go.sum
index 04b75b62..f688b8dc 100644
--- a/go.sum
+++ b/go.sum
@@ -91,8 +91,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/digitalocean/godo v1.128.1-0.20241025145008-2654a9d1e887 h1:kdXNbMfHEDbQilcqllKkNrJ85ftyJSvSDpsQvzrhHbg=
-github.com/digitalocean/godo v1.128.1-0.20241025145008-2654a9d1e887/go.mod h1:PU8JB6I1XYkQIdHFop8lLAY9ojp6M0XcU0TWaQSxbrc=
+github.com/digitalocean/godo v1.129.1-0.20241113153826-8f5ac1bc9671 h1:ZWx8CdoEZ0ZM9jy+hUOVL6OZJtm9Nb6rBCVhF2sokog=
+github.com/digitalocean/godo v1.129.1-0.20241113153826-8f5ac1bc9671/go.mod h1:PU8JB6I1XYkQIdHFop8lLAY9ojp6M0XcU0TWaQSxbrc=
 github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
 github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
 github.com/docker/cli v24.0.5+incompatible h1:WeBimjvS0eKdH4Ygx+ihVq1Q++xg36M/rMi4aXAvodc=
diff --git a/pkg/listen/listen.go b/pkg/listen/listen.go
index c5ca1aab..7746e23f 100644
--- a/pkg/listen/listen.go
+++ b/pkg/listen/listen.go
@@ -2,12 +2,13 @@ package listen
 
 import (
 	"bytes"
+	"context"
+	"fmt"
 	"io"
 	"net/url"
-	"os"
-	"os/signal"
 
 	"github.com/gorilla/websocket"
+	"golang.org/x/sync/errgroup"
 )
 
 // SchemaFunc takes a slice of bytes and returns an io.Reader (See Listener.SchemaFunc)
@@ -27,34 +28,30 @@ type Listener struct {
 	// Out is an io.Writer to output to.
 	// doctl hint: this should usually be commands.CmdConfig.Out
 	Out io.Writer
-
-	done chan bool
-	stop chan bool
+	// InputCh is a channel to send input to the websocket
+	InCh <-chan []byte
 }
 
 // ListenerService listens to a websocket connection and outputs to the provided io.Writer
 type ListenerService interface {
-	Start() error
-	Stop()
+	Listen(ctx context.Context) error
 }
 
 var _ ListenerService = &Listener{}
 
 // NewListener returns a configured Listener
-func NewListener(url *url.URL, token string, schemaFunc SchemaFunc, out io.Writer) ListenerService {
+func NewListener(url *url.URL, token string, schemaFunc SchemaFunc, out io.Writer, inCh <-chan []byte) ListenerService {
 	return &Listener{
 		URL:        url,
 		Token:      token,
 		SchemaFunc: schemaFunc,
 		Out:        out,
-
-		done: make(chan bool),
-		stop: make(chan bool),
+		InCh:       inCh,
 	}
 }
 
-// Start makes the websocket connection and writes messages to the io.Writer
-func (l *Listener) Start() error {
+// Listen makes the websocket connection and writes messages to the io.Writer
+func (l *Listener) Listen(ctx context.Context) error {
 	if l.Token != "" {
 		params := l.URL.Query()
 		params.Set("token", l.Token)
@@ -63,19 +60,21 @@ func (l *Listener) Start() error {
 
 	c, _, err := websocket.DefaultDialer.Dial(l.URL.String(), nil)
 	if err != nil {
-		return err
+		return fmt.Errorf("error creating websocket connection: %w", err)
 	}
 	defer c.Close()
 
-	interrupt := make(chan os.Signal, 1)
-	signal.Notify(interrupt, os.Interrupt)
-	done := l.done
-	go func() error {
+	done := make(chan struct{})
+	grp, ctx := errgroup.WithContext(ctx)
+	grp.Go(func() error {
 		defer close(done)
 		for {
 			_, message, err := c.ReadMessage()
 			if err != nil {
-				return err
+				if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
+					return nil
+				}
+				return fmt.Errorf("error reading from websocket: %w", err)
 			}
 
 			var r io.Reader
@@ -90,34 +89,27 @@ func (l *Listener) Start() error {
 
 			io.Copy(l.Out, r)
 		}
-	}()
+	})
 
-	for {
-		select {
-		case <-done:
-			return nil
-		case <-interrupt:
-			return writeCloseMessage(c)
-		case <-l.stop:
-			return writeCloseMessage(c)
+	grp.Go(func() error {
+		for {
+			select {
+			case data := <-l.InCh:
+				if err := c.WriteMessage(websocket.TextMessage, data); err != nil {
+					return fmt.Errorf("error writing to websocket: %w", err)
+				}
+			case <-ctx.Done():
+				if err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")); err != nil {
+					return fmt.Errorf("error writing close message: %w", err)
+				}
+				return nil
+			case <-done:
+				return nil
+			}
 		}
-	}
-}
-
-// Stop signals the Listener to close the websocket connection
-func (l *Listener) Stop() {
-	select {
-	case <-l.done:
-	default:
-		l.stop <- true
-	}
-}
-
-func writeCloseMessage(c *websocket.Conn) error {
-	err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
-	if err != nil {
+	})
+	if err := grp.Wait(); err != nil {
 		return err
 	}
-
 	return nil
 }
diff --git a/pkg/terminal/rawstdin.go b/pkg/terminal/rawstdin.go
new file mode 100644
index 00000000..1ad53c0a
--- /dev/null
+++ b/pkg/terminal/rawstdin.go
@@ -0,0 +1,38 @@
+package terminal
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"golang.org/x/term"
+)
+
+// ReadRawStdin sets the terminal to raw mode and reads from stdin one byte at a time, sending each byte to the provided channel.
+func (t *terminal) ReadRawStdin(ctx context.Context, stdinCh chan<- string) (restoreTerminalFn func(), err error) {
+	// Set terminal to raw mode
+	oldState, err := term.MakeRaw(int(os.Stdin.Fd()))
+	if err != nil {
+		return nil, fmt.Errorf("error setting terminal to raw mode: %v", err)
+	}
+	restoreTerminalFn = func() {
+		term.Restore(int(os.Stdin.Fd()), oldState) // Restore terminal on exit
+	}
+
+	go func() {
+		for {
+			var b [1]byte
+			_, err := os.Stdin.Read(b[:]) // Read one byte at a time
+			if err != nil {
+				continue
+			}
+
+			select {
+			case stdinCh <- string(b[:]):
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return
+}
diff --git a/pkg/terminal/resize.go b/pkg/terminal/resize.go
new file mode 100644
index 00000000..b78a3db2
--- /dev/null
+++ b/pkg/terminal/resize.go
@@ -0,0 +1,37 @@
+//go:build !windows
+// +build !windows
+
+package terminal
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+
+	"golang.org/x/sys/unix"
+	"golang.org/x/term"
+)
+
+// MonitorResizeEvents monitors the terminal for resize events and sends them to the provided channel.
+func (t *terminal) MonitorResizeEvents(ctx context.Context, resizeEvents chan<- TerminalSize) error {
+	winch := make(chan os.Signal, 1)
+	signal.Notify(winch, unix.SIGWINCH)
+	defer signal.Stop(winch)
+
+	for {
+		width, height, err := term.GetSize(int(os.Stdin.Fd()))
+		if err != nil {
+			return fmt.Errorf("error getting terminal size: %w", err)
+		}
+		terminalSize := TerminalSize{Width: width, Height: height}
+
+		resizeEvents <- terminalSize
+
+		select {
+		case <-winch:
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
diff --git a/pkg/terminal/resize_windows.go b/pkg/terminal/resize_windows.go
new file mode 100644
index 00000000..babf4a0e
--- /dev/null
+++ b/pkg/terminal/resize_windows.go
@@ -0,0 +1,35 @@
+package terminal
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"golang.org/x/term"
+)
+
+// MonitorResizeEvents monitors the terminal for resize events and sends them to the provided channel.
+func (t *terminal) MonitorResizeEvents(ctx context.Context, resizeEvents chan<- TerminalSize) error {
+	var prevTerminalSize TerminalSize
+
+	ticker := time.NewTicker(250 * time.Millisecond)
+	for {
+		width, height, err := term.GetSize(int(os.Stdin.Fd()))
+		if err != nil {
+			return fmt.Errorf("error getting terminal size: %w", err)
+		}
+		terminalSize := TerminalSize{Width: width, Height: height}
+		if terminalSize != prevTerminalSize {
+			prevTerminalSize = terminalSize
+			resizeEvents <- terminalSize
+		}
+
+		// sleep to avoid hot looping
+		select {
+		case <-ticker.C:
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
diff --git a/pkg/terminal/terminal.go b/pkg/terminal/terminal.go
new file mode 100644
index 00000000..502cc366
--- /dev/null
+++ b/pkg/terminal/terminal.go
@@ -0,0 +1,20 @@
+package terminal
+
+import "context"
+
+// Terminal provides an interface for interacting with the terminal
+type Terminal interface {
+	ReadRawStdin(ctx context.Context, stdinCh chan<- string) (restoreTerminalFn func(), err error)
+	MonitorResizeEvents(ctx context.Context, resizeEvents chan<- TerminalSize) error
+}
+
+// terminal is an implementation of Terminal
+type terminal struct{}
+
+// Ensure terminal implements Terminal
+var _ Terminal = &terminal{}
+
+// New returns a new Terminal
+func New() Terminal {
+	return &terminal{}
+}
diff --git a/pkg/terminal/terminalsize.go b/pkg/terminal/terminalsize.go
new file mode 100644
index 00000000..2aeca245
--- /dev/null
+++ b/pkg/terminal/terminalsize.go
@@ -0,0 +1,7 @@
+package terminal
+
+// TerminalSize represents the size of a terminal
+type TerminalSize struct {
+	Width  int
+	Height int
+}
diff --git a/scripts/regenmocks.sh b/scripts/regenmocks.sh
index 93599c94..817711a2 100755
--- a/scripts/regenmocks.sh
+++ b/scripts/regenmocks.sh
@@ -41,6 +41,7 @@ mockgen -source vpcs.go -package=mocks VPCsService > mocks/VPCsService.go
 mockgen -source 1_clicks.go -package=mocks OneClickService > mocks/OneClickService.go
 mockgen -source ../pkg/runner/runner.go -package=mocks Runner > mocks/Runner.go
 mockgen -source ../pkg/listen/listen.go -package=mocks Listen > mocks/Listen.go
+mockgen -source ../pkg/terminal/terminal.go -package=mocks Terminal > mocks/Terminal.go
 mockgen -source monitoring.go -package=mocks MonitoringService > mocks/MonitoringService.go
 mockgen -source reserved_ip_actions.go -package=mocks ReservedIPActionsService > mocks/ReservedIPActionsService.go
 mockgen -source reserved_ips.go -package=mocks ReservedIPsService > mocks/ReservedIPsService.go
diff --git a/util.go b/util.go
index 4adddc1d..88942456 100644
--- a/util.go
+++ b/util.go
@@ -14,8 +14,11 @@ limitations under the License.
 package doctl
 
 import (
+	"context"
+
 	"github.com/digitalocean/doctl/pkg/listen"
 	"github.com/digitalocean/doctl/pkg/runner"
+	"github.com/digitalocean/doctl/pkg/terminal"
 )
 
 // MockRunner is an implementation of Runner for mocking.
@@ -37,12 +40,24 @@ type MockListener struct {
 
 var _ listen.ListenerService = &MockListener{}
 
-// Start mocks ListenerService.Start
-func (tr *MockListener) Start() error {
+// Listen mocks ListenerService.Listen
+func (tr *MockListener) Listen(ctx context.Context) error {
 	return tr.Err
 }
 
-// Stop mocks ListenerService.Stop
-func (tr *MockListener) Stop() {
-	return
+// MockTerminal is an implementation of Terminal for mocking.
+type MockTerminal struct {
+	Err error
+}
+
+var _ terminal.Terminal = &MockTerminal{}
+
+// ReadRawStdin mocks Terminal.ReadRawStdin
+func (tr *MockTerminal) ReadRawStdin(ctx context.Context, stdinCh chan<- string) (restoreFn func(), err error) {
+	return func() {}, tr.Err
+}
+
+// MonitorResizeEvents mocks Terminal.MonitorResizeEvents
+func (tr *MockTerminal) MonitorResizeEvents(ctx context.Context, resizeEvents chan<- terminal.TerminalSize) error {
+	return tr.Err
 }
diff --git a/vendor/github.com/digitalocean/godo/CHANGELOG.md b/vendor/github.com/digitalocean/godo/CHANGELOG.md
index 667255b2..735ab4ff 100644
--- a/vendor/github.com/digitalocean/godo/CHANGELOG.md
+++ b/vendor/github.com/digitalocean/godo/CHANGELOG.md
@@ -1,5 +1,12 @@
 # Change Log
 
+## [v1.129.0] - 2024-11-06
+
+- #752 - @andrewsomething - Support maps in Stringify
+- #749 - @loosla - [droplets]: add droplet backup policies
+- #730 - @rak16 - DOCR-1201: Add new RegistriesService to support methods for multiple-registry open beta
+- #748 - @andrewsomething - Support Droplet GPU information
+
 ## [v1.128.0] - 2024-10-24
 
 - #746 - @blesswinsamuel - Add archive field to AppSpec to archive/restore apps
diff --git a/vendor/github.com/digitalocean/godo/apps.go b/vendor/github.com/digitalocean/godo/apps.go
index ac792658..24fe738e 100644
--- a/vendor/github.com/digitalocean/godo/apps.go
+++ b/vendor/github.com/digitalocean/godo/apps.go
@@ -40,6 +40,7 @@ type AppsService interface {
 	CreateDeployment(ctx context.Context, appID string, create ...*DeploymentCreateRequest) (*Deployment, *Response, error)
 
 	GetLogs(ctx context.Context, appID, deploymentID, component string, logType AppLogType, follow bool, tailLines int) (*AppLogs, *Response, error)
+	GetExec(ctx context.Context, appID, deploymentID, component string) (*AppExec, *Response, error)
 
 	ListRegions(ctx context.Context) ([]*AppRegion, *Response, error)
 
@@ -77,6 +78,11 @@ type AppLogs struct {
 	HistoricURLs []string `json:"historic_urls"`
 }
 
+// AppExec represents the websocket URL used for sending/receiving console input and output.
+type AppExec struct {
+	URL string `json:"url"`
+}
+
 // AppUpdateRequest represents a request to update an app.
 type AppUpdateRequest struct {
 	Spec *AppSpec `json:"spec"`
@@ -368,6 +374,27 @@ func (s *AppsServiceOp) GetLogs(ctx context.Context, appID, deploymentID, compon
 	return logs, resp, nil
 }
 
+// GetExec retrieves the websocket URL used for sending/receiving console input and output.
+func (s *AppsServiceOp) GetExec(ctx context.Context, appID, deploymentID, component string) (*AppExec, *Response, error) {
+	var url string
+	if deploymentID == "" {
+		url = fmt.Sprintf("%s/%s/components/%s/exec", appsBasePath, appID, component)
+	} else {
+		url = fmt.Sprintf("%s/%s/deployments/%s/components/%s/exec", appsBasePath, appID, deploymentID, component)
+	}
+
+	req, err := s.client.NewRequest(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	logs := new(AppExec)
+	resp, err := s.client.Do(ctx, req, logs)
+	if err != nil {
+		return nil, resp, err
+	}
+	return logs, resp, nil
+}
+
 // ListRegions lists all regions supported by App Platform.
 func (s *AppsServiceOp) ListRegions(ctx context.Context) ([]*AppRegion, *Response, error) {
 	path := fmt.Sprintf("%s/regions", appsBasePath)
diff --git a/vendor/github.com/digitalocean/godo/databases.go b/vendor/github.com/digitalocean/godo/databases.go
index 276fb4a6..e2a7d567 100644
--- a/vendor/github.com/digitalocean/godo/databases.go
+++ b/vendor/github.com/digitalocean/godo/databases.go
@@ -589,6 +589,9 @@ type PostgreSQLConfig struct {
 	BackupMinute                    *int                         `json:"backup_minute,omitempty"`
 	WorkMem                         *int                         `json:"work_mem,omitempty"`
 	TimeScaleDB                     *PostgreSQLTimeScaleDBConfig `json:"timescaledb,omitempty"`
+	SynchronousReplication          *string                      `json:"synchronous_replication,omitempty"`
+	StatMonitorEnable               *bool                        `json:"stat_monitor_enable,omitempty"`
+	MaxFailoverReplicationTimeLag   *int64                       `json:"max_failover_replication_time_lag,omitempty"`
 }
 
 // PostgreSQLBouncerConfig configuration
@@ -653,6 +656,13 @@ type MySQLConfig struct {
 	BackupHour                   *int     `json:"backup_hour,omitempty"`
 	BackupMinute                 *int     `json:"backup_minute,omitempty"`
 	BinlogRetentionPeriod        *int     `json:"binlog_retention_period,omitempty"`
+	InnodbChangeBufferMaxSize    *int     `json:"innodb_change_buffer_max_size,omitempty"`
+	InnodbFlushNeighbors         *int     `json:"innodb_flush_neighbors,omitempty"`
+	InnodbReadIoThreads          *int     `json:"innodb_read_io_threads,omitempty"`
+	InnodbThreadConcurrency      *int     `json:"innodb_thread_concurrency,omitempty"`
+	InnodbWriteIoThreads         *int     `json:"innodb_write_io_threads,omitempty"`
+	NetBufferLength              *int     `json:"net_buffer_length,omitempty"`
+	LogOutput                    *string  `json:"log_output,omitempty"`
 }
 
 // MongoDBConfig holds advanced configurations for MongoDB database clusters.
diff --git a/vendor/github.com/digitalocean/godo/droplet_actions.go b/vendor/github.com/digitalocean/godo/droplet_actions.go
index 2e09d0c5..ed0f583c 100644
--- a/vendor/github.com/digitalocean/godo/droplet_actions.go
+++ b/vendor/github.com/digitalocean/godo/droplet_actions.go
@@ -30,6 +30,8 @@ type DropletActionsService interface {
 	SnapshotByTag(context.Context, string, string) ([]Action, *Response, error)
 	EnableBackups(context.Context, int) (*Action, *Response, error)
 	EnableBackupsByTag(context.Context, string) ([]Action, *Response, error)
+	EnableBackupsWithPolicy(context.Context, int, *DropletBackupPolicyRequest) (*Action, *Response, error)
+	ChangeBackupPolicy(context.Context, int, *DropletBackupPolicyRequest) (*Action, *Response, error)
 	DisableBackups(context.Context, int) (*Action, *Response, error)
 	DisableBackupsByTag(context.Context, string) ([]Action, *Response, error)
 	PasswordReset(context.Context, int) (*Action, *Response, error)
@@ -169,6 +171,42 @@ func (s *DropletActionsServiceOp) EnableBackupsByTag(ctx context.Context, tag st
 	return s.doActionByTag(ctx, tag, request)
 }
 
+// EnableBackupsWithPolicy enables droplet's backup with a backup policy applied.
+func (s *DropletActionsServiceOp) EnableBackupsWithPolicy(ctx context.Context, id int, policy *DropletBackupPolicyRequest) (*Action, *Response, error) {
+	if policy == nil {
+		return nil, nil, NewArgError("policy", "policy can't be nil")
+	}
+
+	policyMap := map[string]interface{}{
+		"plan":    policy.Plan,
+		"weekday": policy.Weekday,
+	}
+	if policy.Hour != nil {
+		policyMap["hour"] = policy.Hour
+	}
+
+	request := &ActionRequest{"type": "enable_backups", "backup_policy": policyMap}
+	return s.doAction(ctx, id, request)
+}
+
+// ChangeBackupPolicy updates a backup policy when backups are enabled.
+func (s *DropletActionsServiceOp) ChangeBackupPolicy(ctx context.Context, id int, policy *DropletBackupPolicyRequest) (*Action, *Response, error) {
+	if policy == nil {
+		return nil, nil, NewArgError("policy", "policy can't be nil")
+	}
+
+	policyMap := map[string]interface{}{
+		"plan":    policy.Plan,
+		"weekday": policy.Weekday,
+	}
+	if policy.Hour != nil {
+		policyMap["hour"] = policy.Hour
+	}
+
+	request := &ActionRequest{"type": "change_backup_policy", "backup_policy": policyMap}
+	return s.doAction(ctx, id, request)
+}
+
 // DisableBackups disables backups for a Droplet.
 func (s *DropletActionsServiceOp) DisableBackups(ctx context.Context, id int) (*Action, *Response, error) {
 	request := &ActionRequest{"type": "disable_backups"}
diff --git a/vendor/github.com/digitalocean/godo/droplets.go b/vendor/github.com/digitalocean/godo/droplets.go
index 1ed09ec8..2ddd7d6b 100644
--- a/vendor/github.com/digitalocean/godo/droplets.go
+++ b/vendor/github.com/digitalocean/godo/droplets.go
@@ -30,6 +30,9 @@ type DropletsService interface {
 	Backups(context.Context, int, *ListOptions) ([]Image, *Response, error)
 	Actions(context.Context, int, *ListOptions) ([]Action, *Response, error)
 	Neighbors(context.Context, int) ([]Droplet, *Response, error)
+	GetBackupPolicy(context.Context, int) (*DropletBackupPolicy, *Response, error)
+	ListBackupPolicies(context.Context, *ListOptions) (map[int]*DropletBackupPolicy, *Response, error)
+	ListSupportedBackupPolicies(context.Context) ([]*SupportedBackupPolicy, *Response, error)
 }
 
 // DropletsServiceOp handles communication with the Droplet related methods of the
@@ -218,37 +221,46 @@ func (d DropletCreateSSHKey) MarshalJSON() ([]byte, error) {
 
 // DropletCreateRequest represents a request to create a Droplet.
 type DropletCreateRequest struct {
-	Name              string                `json:"name"`
-	Region            string                `json:"region"`
-	Size              string                `json:"size"`
-	Image             DropletCreateImage    `json:"image"`
-	SSHKeys           []DropletCreateSSHKey `json:"ssh_keys"`
-	Backups           bool                  `json:"backups"`
-	IPv6              bool                  `json:"ipv6"`
-	PrivateNetworking bool                  `json:"private_networking"`
-	Monitoring        bool                  `json:"monitoring"`
-	UserData          string                `json:"user_data,omitempty"`
-	Volumes           []DropletCreateVolume `json:"volumes,omitempty"`
-	Tags              []string              `json:"tags"`
-	VPCUUID           string                `json:"vpc_uuid,omitempty"`
-	WithDropletAgent  *bool                 `json:"with_droplet_agent,omitempty"`
+	Name              string                      `json:"name"`
+	Region            string                      `json:"region"`
+	Size              string                      `json:"size"`
+	Image             DropletCreateImage          `json:"image"`
+	SSHKeys           []DropletCreateSSHKey       `json:"ssh_keys"`
+	Backups           bool                        `json:"backups"`
+	IPv6              bool                        `json:"ipv6"`
+	PrivateNetworking bool                        `json:"private_networking"`
+	Monitoring        bool                        `json:"monitoring"`
+	UserData          string                      `json:"user_data,omitempty"`
+	Volumes           []DropletCreateVolume       `json:"volumes,omitempty"`
+	Tags              []string                    `json:"tags"`
+	VPCUUID           string                      `json:"vpc_uuid,omitempty"`
+	WithDropletAgent  *bool                       `json:"with_droplet_agent,omitempty"`
+	BackupPolicy      *DropletBackupPolicyRequest `json:"backup_policy,omitempty"`
 }
 
 // DropletMultiCreateRequest is a request to create multiple Droplets.
 type DropletMultiCreateRequest struct {
-	Names             []string              `json:"names"`
-	Region            string                `json:"region"`
-	Size              string                `json:"size"`
-	Image             DropletCreateImage    `json:"image"`
-	SSHKeys           []DropletCreateSSHKey `json:"ssh_keys"`
-	Backups           bool                  `json:"backups"`
-	IPv6              bool                  `json:"ipv6"`
-	PrivateNetworking bool                  `json:"private_networking"`
-	Monitoring        bool                  `json:"monitoring"`
-	UserData          string                `json:"user_data,omitempty"`
-	Tags              []string              `json:"tags"`
-	VPCUUID           string                `json:"vpc_uuid,omitempty"`
-	WithDropletAgent  *bool                 `json:"with_droplet_agent,omitempty"`
+	Names             []string                    `json:"names"`
+	Region            string                      `json:"region"`
+	Size              string                      `json:"size"`
+	Image             DropletCreateImage          `json:"image"`
+	SSHKeys           []DropletCreateSSHKey       `json:"ssh_keys"`
+	Backups           bool                        `json:"backups"`
+	IPv6              bool                        `json:"ipv6"`
+	PrivateNetworking bool                        `json:"private_networking"`
+	Monitoring        bool                        `json:"monitoring"`
+	UserData          string                      `json:"user_data,omitempty"`
+	Tags              []string                    `json:"tags"`
+	VPCUUID           string                      `json:"vpc_uuid,omitempty"`
+	WithDropletAgent  *bool                       `json:"with_droplet_agent,omitempty"`
+	BackupPolicy      *DropletBackupPolicyRequest `json:"backup_policy,omitempty"`
+}
+
+// DropletBackupPolicyRequest defines the backup policy when creating a Droplet.
+type DropletBackupPolicyRequest struct {
+	Plan    string `json:"plan,omitempty"`
+	Weekday string `json:"weekday,omitempty"`
+	Hour    *int   `json:"hour,omitempty"`
 }
 
 func (d DropletCreateRequest) String() string {
@@ -618,3 +630,109 @@ func (s *DropletsServiceOp) dropletActionStatus(ctx context.Context, uri string)
 
 	return action.Status, nil
 }
+
+// DropletBackupPolicy defines the information about a droplet's backup policy.
+type DropletBackupPolicy struct {
+	DropletID        int                        `json:"droplet_id,omitempty"`
+	BackupEnabled    bool                       `json:"backup_enabled,omitempty"`
+	BackupPolicy     *DropletBackupPolicyConfig `json:"backup_policy,omitempty"`
+	NextBackupWindow *BackupWindow              `json:"next_backup_window,omitempty"`
+}
+
+// DropletBackupPolicyConfig defines the backup policy for a Droplet.
+type DropletBackupPolicyConfig struct {
+	Plan                string `json:"plan,omitempty"`
+	Weekday             string `json:"weekday,omitempty"`
+	Hour                int    `json:"hour,omitempty"`
+	WindowLengthHours   int    `json:"window_length_hours,omitempty"`
+	RetentionPeriodDays int    `json:"retention_period_days,omitempty"`
+}
+
+// dropletBackupPolicyRoot represents a DropletBackupPolicy root
+type dropletBackupPolicyRoot struct {
+	DropletBackupPolicy *DropletBackupPolicy `json:"policy,omitempty"`
+}
+
+type dropletBackupPoliciesRoot struct {
+	DropletBackupPolicies map[int]*DropletBackupPolicy `json:"policies,omitempty"`
+	Links                 *Links                       `json:"links,omitempty"`
+	Meta                  *Meta                        `json:"meta"`
+}
+
+// Get individual droplet backup policy.
+func (s *DropletsServiceOp) GetBackupPolicy(ctx context.Context, dropletID int) (*DropletBackupPolicy, *Response, error) {
+	if dropletID < 1 {
+		return nil, nil, NewArgError("dropletID", "cannot be less than 1")
+	}
+
+	path := fmt.Sprintf("%s/%d/backups/policy", dropletBasePath, dropletID)
+
+	req, err := s.client.NewRequest(ctx, http.MethodGet, path, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	root := new(dropletBackupPolicyRoot)
+	resp, err := s.client.Do(ctx, req, root)
+	if err != nil {
+		return nil, resp, err
+	}
+
+	return root.DropletBackupPolicy, resp, err
+}
+
+// List all droplet backup policies.
+func (s *DropletsServiceOp) ListBackupPolicies(ctx context.Context, opt *ListOptions) (map[int]*DropletBackupPolicy, *Response, error) {
+	path := fmt.Sprintf("%s/backups/policies", dropletBasePath)
+	path, err := addOptions(path, opt)
+	if err != nil {
+		return nil, nil, err
+	}
+	req, err := s.client.NewRequest(ctx, http.MethodGet, path, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	root := new(dropletBackupPoliciesRoot)
+	resp, err := s.client.Do(ctx, req, root)
+	if err != nil {
+		return nil, resp, err
+	}
+	if l := root.Links; l != nil {
+		resp.Links = l
+	}
+	if m := root.Meta; m != nil {
+		resp.Meta = m
+	}
+
+	return root.DropletBackupPolicies, resp, nil
+}
+
+type SupportedBackupPolicy struct {
+	Name                 string   `json:"name,omitempty"`
+	PossibleWindowStarts []int    `json:"possible_window_starts,omitempty"`
+	WindowLengthHours    int      `json:"window_length_hours,omitempty"`
+	RetentionPeriodDays  int      `json:"retention_period_days,omitempty"`
+	PossibleDays         []string `json:"possible_days,omitempty"`
+}
+
+type dropletSupportedBackupPoliciesRoot struct {
+	SupportedBackupPolicies []*SupportedBackupPolicy `json:"supported_policies,omitempty"`
+}
+
+// List supported droplet backup policies.
+func (s *DropletsServiceOp) ListSupportedBackupPolicies(ctx context.Context) ([]*SupportedBackupPolicy, *Response, error) {
+	path := fmt.Sprintf("%s/backups/supported_policies", dropletBasePath)
+	req, err := s.client.NewRequest(ctx, http.MethodGet, path, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	root := new(dropletSupportedBackupPoliciesRoot)
+	resp, err := s.client.Do(ctx, req, root)
+	if err != nil {
+		return nil, resp, err
+	}
+
+	return root.SupportedBackupPolicies, resp, nil
+}
diff --git a/vendor/github.com/digitalocean/godo/godo.go b/vendor/github.com/digitalocean/godo/godo.go
index edf0f6d4..a399bc3e 100644
--- a/vendor/github.com/digitalocean/godo/godo.go
+++ b/vendor/github.com/digitalocean/godo/godo.go
@@ -21,7 +21,7 @@ import (
 )
 
 const (
-	libraryVersion = "1.128.0"
+	libraryVersion = "1.129.0"
 	defaultBaseURL = "https://api.digitalocean.com/"
 	userAgent      = "godo/" + libraryVersion
 	mediaType      = "application/json"
@@ -81,6 +81,7 @@ type Client struct {
 	Projects          ProjectsService
 	Regions           RegionsService
 	Registry          RegistryService
+	Registries        RegistriesService
 	ReservedIPs       ReservedIPsService
 	ReservedIPActions ReservedIPActionsService
 	Sizes             SizesService
@@ -292,6 +293,7 @@ func NewClient(httpClient *http.Client) *Client {
 	c.Projects = &ProjectsServiceOp{client: c}
 	c.Regions = &RegionsServiceOp{client: c}
 	c.Registry = &RegistryServiceOp{client: c}
+	c.Registries = &RegistriesServiceOp{client: c}
 	c.ReservedIPs = &ReservedIPsServiceOp{client: c}
 	c.ReservedIPActions = &ReservedIPActionsServiceOp{client: c}
 	c.Sizes = &SizesServiceOp{client: c}
diff --git a/vendor/github.com/digitalocean/godo/registry.go b/vendor/github.com/digitalocean/godo/registry.go
index b0c24328..e6482268 100644
--- a/vendor/github.com/digitalocean/godo/registry.go
+++ b/vendor/github.com/digitalocean/godo/registry.go
@@ -14,6 +14,9 @@ const (
 	registryPath = "/v2/registry"
 	// RegistryServer is the hostname of the DigitalOcean registry service
 	RegistryServer = "registry.digitalocean.com"
+
+	// Multi-registry Open Beta API constants
+	registriesPath = "/v2/registries"
 )
 
 // RegistryService is an interface for interfacing with the Registry endpoints
@@ -240,6 +243,19 @@ type RegistryValidateNameRequest struct {
 	Name string `json:"name"`
 }
 
+// Multi-registry Open Beta API structs
+
+type registriesRoot struct {
+	Registries             []*Registry `json:"registries,omitempty"`
+	TotalStorageUsageBytes uint64      `json:"total_storage_usage_bytes,omitempty"`
+}
+
+// RegistriesCreateRequest represents a request to create a secondary registry.
+type RegistriesCreateRequest struct {
+	Name   string `json:"name,omitempty"`
+	Region string `json:"region,omitempty"`
+}
+
 // Get retrieves the details of a Registry.
 func (svc *RegistryServiceOp) Get(ctx context.Context) (*Registry, *Response, error) {
 	req, err := svc.client.NewRequest(ctx, http.MethodGet, registryPath, nil)
@@ -610,3 +626,107 @@ func (svc *RegistryServiceOp) ValidateName(ctx context.Context, request *Registr
 	}
 	return resp, nil
 }
+
+// RegistriesService is an interface for interfacing with the new multiple-registry beta endpoints
+// of the DigitalOcean API.
+//
+// We are creating a separate Service in alignment with the new /v2/registries endpoints.
+type RegistriesService interface {
+	Get(context.Context, string) (*Registry, *Response, error)
+	List(context.Context) ([]*Registry, *Response, error)
+	Create(context.Context, *RegistriesCreateRequest) (*Registry, *Response, error)
+	Delete(context.Context, string) (*Response, error)
+	DockerCredentials(context.Context, string, *RegistryDockerCredentialsRequest) (*DockerCredentials, *Response, error)
+}
+
+var _ RegistriesService = &RegistriesServiceOp{}
+
+// RegistriesServiceOp handles communication with the multiple-registry beta methods.
+type RegistriesServiceOp struct {
+	client *Client
+}
+
+// Get returns the details of a named Registry.
+func (svc *RegistriesServiceOp) Get(ctx context.Context, registry string) (*Registry, *Response, error) {
+	path := fmt.Sprintf("%s/%s", registriesPath, registry)
+	req, err := svc.client.NewRequest(ctx, http.MethodGet, path, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	root := new(registryRoot)
+	resp, err := svc.client.Do(ctx, req, root)
+	if err != nil {
+		return nil, resp, err
+	}
+	return root.Registry, resp, nil
+}
+
+// List returns a list of the named Registries.
+func (svc *RegistriesServiceOp) List(ctx context.Context) ([]*Registry, *Response, error) {
+	req, err := svc.client.NewRequest(ctx, http.MethodGet, registriesPath, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+	root := new(registriesRoot)
+	resp, err := svc.client.Do(ctx, req, root)
+	if err != nil {
+		return nil, resp, err
+	}
+	return root.Registries, resp, nil
+}
+
+// Create creates a named Registry.
+func (svc *RegistriesServiceOp) Create(ctx context.Context, create *RegistriesCreateRequest) (*Registry, *Response, error) {
+	req, err := svc.client.NewRequest(ctx, http.MethodPost, registriesPath, create)
+	if err != nil {
+		return nil, nil, err
+	}
+	root := new(registryRoot)
+	resp, err := svc.client.Do(ctx, req, root)
+	if err != nil {
+		return nil, resp, err
+	}
+	return root.Registry, resp, nil
+}
+
+// Delete deletes a named Registry. There is no way to recover a Registry once it has
+// been destroyed.
+func (svc *RegistriesServiceOp) Delete(ctx context.Context, registry string) (*Response, error) {
+	path := fmt.Sprintf("%s/%s", registriesPath, registry)
+	req, err := svc.client.NewRequest(ctx, http.MethodDelete, path, nil)
+	if err != nil {
+		return nil, err
+	}
+	resp, err := svc.client.Do(ctx, req, nil)
+	if err != nil {
+		return resp, err
+	}
+	return resp, nil
+}
+
+// DockerCredentials retrieves a Docker config file containing named Registry's credentials.
+func (svc *RegistriesServiceOp) DockerCredentials(ctx context.Context, registry string, request *RegistryDockerCredentialsRequest) (*DockerCredentials, *Response, error) {
+	path := fmt.Sprintf("%s/%s/%s", registriesPath, registry, "docker-credentials")
+	req, err := svc.client.NewRequest(ctx, http.MethodGet, path, nil)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	q := req.URL.Query()
+	q.Add("read_write", strconv.FormatBool(request.ReadWrite))
+	if request.ExpirySeconds != nil {
+		q.Add("expiry_seconds", strconv.Itoa(*request.ExpirySeconds))
+	}
+	req.URL.RawQuery = q.Encode()
+
+	var buf bytes.Buffer
+	resp, err := svc.client.Do(ctx, req, &buf)
+	if err != nil {
+		return nil, resp, err
+	}
+
+	dc := &DockerCredentials{
+		DockerConfigJSON: buf.Bytes(),
+	}
+	return dc, resp, nil
+}
diff --git a/vendor/github.com/digitalocean/godo/strings.go b/vendor/github.com/digitalocean/godo/strings.go
index f92893ed..5a258131 100644
--- a/vendor/github.com/digitalocean/godo/strings.go
+++ b/vendor/github.com/digitalocean/godo/strings.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"reflect"
+	"sort"
 	"strings"
 )
 
@@ -46,6 +47,8 @@ func stringifyValue(w io.Writer, val reflect.Value) {
 		return
 	case reflect.Struct:
 		stringifyStruct(w, v)
+	case reflect.Map:
+		stringifyMap(w, v)
 	default:
 		if v.CanInterface() {
 			fmt.Fprint(w, v.Interface())
@@ -66,6 +69,27 @@ func stringifySlice(w io.Writer, v reflect.Value) {
 	_, _ = w.Write([]byte{']'})
 }
 
+func stringifyMap(w io.Writer, v reflect.Value) {
+	_, _ = w.Write([]byte("map["))
+
+	// Sort the keys so that the output is stable
+	keys := v.MapKeys()
+	sort.Slice(keys, func(i, j int) bool {
+		return fmt.Sprintf("%v", keys[i]) < fmt.Sprintf("%v", keys[j])
+	})
+
+	for i, key := range keys {
+		stringifyValue(w, key)
+		_, _ = w.Write([]byte{':'})
+		stringifyValue(w, v.MapIndex(key))
+		if i < len(keys)-1 {
+			_, _ = w.Write([]byte(", "))
+		}
+	}
+
+	_, _ = w.Write([]byte("]"))
+}
+
 func stringifyStruct(w io.Writer, v reflect.Value) {
 	if v.Type().Name() != "" {
 		_, _ = w.Write([]byte(v.Type().String()))
diff --git a/vendor/golang.org/x/sync/LICENSE b/vendor/golang.org/x/sync/LICENSE
new file mode 100644
index 00000000..6a66aea5
--- /dev/null
+++ b/vendor/golang.org/x/sync/LICENSE
@@ -0,0 +1,27 @@
+Copyright (c) 2009 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
diff --git a/vendor/golang.org/x/sync/PATENTS b/vendor/golang.org/x/sync/PATENTS
new file mode 100644
index 00000000..73309904
--- /dev/null
+++ b/vendor/golang.org/x/sync/PATENTS
@@ -0,0 +1,22 @@
+Additional IP Rights Grant (Patents)
+
+"This implementation" means the copyrightable works distributed by
+Google as part of the Go project.
+
+Google hereby grants to You a perpetual, worldwide, non-exclusive,
+no-charge, royalty-free, irrevocable (except as stated in this section)
+patent license to make, have made, use, offer to sell, sell, import,
+transfer and otherwise run, modify and propagate the contents of this
+implementation of Go, where such license applies only to those patent
+claims, both currently owned or controlled by Google and acquired in
+the future, licensable by Google that are necessarily infringed by this
+implementation of Go.  This grant does not include claims that would be
+infringed only as a consequence of further modification of this
+implementation.  If you or your agent or exclusive licensee institute or
+order or agree to the institution of patent litigation against any
+entity (including a cross-claim or counterclaim in a lawsuit) alleging
+that this implementation of Go or any code incorporated within this
+implementation of Go constitutes direct or contributory patent
+infringement, or inducement of patent infringement, then any patent
+rights granted to you under this License for this implementation of Go
+shall terminate as of the date such litigation is filed.
diff --git a/vendor/golang.org/x/sync/errgroup/errgroup.go b/vendor/golang.org/x/sync/errgroup/errgroup.go
new file mode 100644
index 00000000..b18efb74
--- /dev/null
+++ b/vendor/golang.org/x/sync/errgroup/errgroup.go
@@ -0,0 +1,132 @@
+// Copyright 2016 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package errgroup provides synchronization, error propagation, and Context
+// cancelation for groups of goroutines working on subtasks of a common task.
+package errgroup
+
+import (
+	"context"
+	"fmt"
+	"sync"
+)
+
+type token struct{}
+
+// A Group is a collection of goroutines working on subtasks that are part of
+// the same overall task.
+//
+// A zero Group is valid, has no limit on the number of active goroutines,
+// and does not cancel on error.
+type Group struct {
+	cancel func(error)
+
+	wg sync.WaitGroup
+
+	sem chan token
+
+	errOnce sync.Once
+	err     error
+}
+
+func (g *Group) done() {
+	if g.sem != nil {
+		<-g.sem
+	}
+	g.wg.Done()
+}
+
+// WithContext returns a new Group and an associated Context derived from ctx.
+//
+// The derived Context is canceled the first time a function passed to Go
+// returns a non-nil error or the first time Wait returns, whichever occurs
+// first.
+func WithContext(ctx context.Context) (*Group, context.Context) {
+	ctx, cancel := withCancelCause(ctx)
+	return &Group{cancel: cancel}, ctx
+}
+
+// Wait blocks until all function calls from the Go method have returned, then
+// returns the first non-nil error (if any) from them.
+func (g *Group) Wait() error {
+	g.wg.Wait()
+	if g.cancel != nil {
+		g.cancel(g.err)
+	}
+	return g.err
+}
+
+// Go calls the given function in a new goroutine.
+// It blocks until the new goroutine can be added without the number of
+// active goroutines in the group exceeding the configured limit.
+//
+// The first call to return a non-nil error cancels the group's context, if the
+// group was created by calling WithContext. The error will be returned by Wait.
+func (g *Group) Go(f func() error) {
+	if g.sem != nil {
+		g.sem <- token{}
+	}
+
+	g.wg.Add(1)
+	go func() {
+		defer g.done()
+
+		if err := f(); err != nil {
+			g.errOnce.Do(func() {
+				g.err = err
+				if g.cancel != nil {
+					g.cancel(g.err)
+				}
+			})
+		}
+	}()
+}
+
+// TryGo calls the given function in a new goroutine only if the number of
+// active goroutines in the group is currently below the configured limit.
+//
+// The return value reports whether the goroutine was started.
+func (g *Group) TryGo(f func() error) bool {
+	if g.sem != nil {
+		select {
+		case g.sem <- token{}:
+			// Note: this allows barging iff channels in general allow barging.
+		default:
+			return false
+		}
+	}
+
+	g.wg.Add(1)
+	go func() {
+		defer g.done()
+
+		if err := f(); err != nil {
+			g.errOnce.Do(func() {
+				g.err = err
+				if g.cancel != nil {
+					g.cancel(g.err)
+				}
+			})
+		}
+	}()
+	return true
+}
+
+// SetLimit limits the number of active goroutines in this group to at most n.
+// A negative value indicates no limit.
+//
+// Any subsequent call to the Go method will block until it can add an active
+// goroutine without exceeding the configured limit.
+//
+// The limit must not be modified while any goroutines in the group are active.
+func (g *Group) SetLimit(n int) {
+	if n < 0 {
+		g.sem = nil
+		return
+	}
+	if len(g.sem) != 0 {
+		panic(fmt.Errorf("errgroup: modify limit while %v goroutines in the group are still active", len(g.sem)))
+	}
+	g.sem = make(chan token, n)
+}
diff --git a/vendor/golang.org/x/sync/errgroup/go120.go b/vendor/golang.org/x/sync/errgroup/go120.go
new file mode 100644
index 00000000..7d419d37
--- /dev/null
+++ b/vendor/golang.org/x/sync/errgroup/go120.go
@@ -0,0 +1,14 @@
+// Copyright 2023 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+//go:build go1.20
+// +build go1.20
+
+package errgroup
+
+import "context"
+
+func withCancelCause(parent context.Context) (context.Context, func(error)) {
+	return context.WithCancelCause(parent)
+}
diff --git a/vendor/golang.org/x/sync/errgroup/pre_go120.go b/vendor/golang.org/x/sync/errgroup/pre_go120.go
new file mode 100644
index 00000000..1795c18a
--- /dev/null
+++ b/vendor/golang.org/x/sync/errgroup/pre_go120.go
@@ -0,0 +1,15 @@
+// Copyright 2023 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+//go:build !go1.20
+// +build !go1.20
+
+package errgroup
+
+import "context"
+
+func withCancelCause(parent context.Context) (context.Context, func(error)) {
+	ctx, cancel := context.WithCancel(parent)
+	return ctx, func(error) { cancel() }
+}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 1ee0dde8..20b23e3b 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -61,7 +61,7 @@ github.com/creack/pty
 # github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 ## explicit
 github.com/davecgh/go-spew/spew
-# github.com/digitalocean/godo v1.128.1-0.20241025145008-2654a9d1e887
+# github.com/digitalocean/godo v1.129.1-0.20241113153826-8f5ac1bc9671
 ## explicit; go 1.22
 github.com/digitalocean/godo
 github.com/digitalocean/godo/metrics
@@ -443,6 +443,9 @@ golang.org/x/net/proxy
 ## explicit; go 1.18
 golang.org/x/oauth2
 golang.org/x/oauth2/internal
+# golang.org/x/sync v0.3.0
+## explicit; go 1.17
+golang.org/x/sync/errgroup
 # golang.org/x/sys v0.25.0
 ## explicit; go 1.18
 golang.org/x/sys/cpu


Test patch:
diff --git a/commands/apps_test.go b/commands/apps_test.go
index 56c4f8bf..cea98533 100644
--- a/commands/apps_test.go
+++ b/commands/apps_test.go
@@ -11,16 +11,19 @@ import (
 
 	"github.com/digitalocean/doctl"
 	"github.com/digitalocean/doctl/pkg/listen"
+	"github.com/digitalocean/doctl/pkg/terminal"
 	"github.com/digitalocean/godo"
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 )
 
 func TestAppsCommand(t *testing.T) {
 	cmd := Apps()
 	require.NotNil(t, cmd)
 	assertCommandNames(t, cmd,
+		"console",
 		"create",
 		"get",
 		"list",
@@ -431,10 +434,10 @@ func TestRunAppsGetLogs(t *testing.T) {
 	for typeStr, logType := range types {
 		withTestClient(t, func(config *CmdConfig, tm *tcMocks) {
 			tm.apps.EXPECT().GetLogs(appID, deploymentID, component, logType, true, 1).Times(1).Return(&godo.AppLogs{LiveURL: "https://proxy-apps-prod-ams3-001.ondigitalocean.app/?token=aa-bb-11-cc-33"}, nil)
-			tm.listen.EXPECT().Start().Times(1).Return(nil)
+			tm.listen.EXPECT().Listen(gomock.Any()).Times(1).Return(nil)
 
 			tc := config.Doit.(*doctl.TestConfig)
-			tc.ListenFn = func(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer) listen.ListenerService {
+			tc.ListenFn = func(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, in <-chan []byte) listen.ListenerService {
 				assert.Equal(t, "aa-bb-11-cc-33", token)
 				assert.Equal(t, "wss://proxy-apps-prod-ams3-001.ondigitalocean.app/?token=aa-bb-11-cc-33", url.String())
 				return tm.listen
@@ -452,6 +455,35 @@ func TestRunAppsGetLogs(t *testing.T) {
 	}
 }
 
+func TestRunAppsConsole(t *testing.T) {
+	appID := uuid.New().String()
+	deploymentID := uuid.New().String()
+	component := "service"
+
+	withTestClient(t, func(config *CmdConfig, tm *tcMocks) {
+		tm.apps.EXPECT().GetExec(appID, deploymentID, component).Times(1).Return(&godo.AppExec{URL: "wss://proxy-apps-prod-ams3-001.ondigitalocean.app/?token=aa-bb-11-cc-33"}, nil)
+		tm.listen.EXPECT().Listen(gomock.Any()).Times(1).Return(nil)
+		tm.terminal.EXPECT().ReadRawStdin(gomock.Any(), gomock.Any()).Times(1).Return(func() {}, nil)
+		tm.terminal.EXPECT().MonitorResizeEvents(gomock.Any(), gomock.Any()).Times(1).Return(nil)
+
+		tc := config.Doit.(*doctl.TestConfig)
+		tc.ListenFn = func(url *url.URL, token string, schemaFunc listen.SchemaFunc, out io.Writer, in <-chan []byte) listen.ListenerService {
+			assert.Equal(t, "aa-bb-11-cc-33", token)
+			assert.Equal(t, "wss://proxy-apps-prod-ams3-001.ondigitalocean.app/?token=aa-bb-11-cc-33", url.String())
+			return tm.listen
+		}
+		tc.TerminalFn = func() terminal.Terminal {
+			return tm.terminal
+		}
+
+		config.Args = append(config.Args, appID, component)
+		config.Doit.Set(config.NS, doctl.ArgAppDeployment, deploymentID)
+
+		err := RunAppsConsole(config)
+		require.NoError(t, err)
+	})
+}
+
 const (
 	validJSONSpec = `{
 	"name": "test",
diff --git a/commands/commands_test.go b/commands/commands_test.go
index 1837453f..db3c2003 100644
--- a/commands/commands_test.go
+++ b/commands/commands_test.go
@@ -179,6 +179,7 @@ type tcMocks struct {
 	vpcs                  *domocks.MockVPCsService
 	oneClick              *domocks.MockOneClickService
 	listen                *domocks.MockListenerService
+	terminal              *domocks.MockTerminal
 	monitoring            *domocks.MockMonitoringService
 	serverless            *domocks.MockServerlessService
 	appBuilderFactory     *builder.MockComponentBuilderFactory
@@ -226,6 +227,7 @@ func withTestClient(t *testing.T, tFn testFn) {
 		vpcs:                  domocks.NewMockVPCsService(ctrl),
 		oneClick:              domocks.NewMockOneClickService(ctrl),
 		listen:                domocks.NewMockListenerService(ctrl),
+		terminal:              domocks.NewMockTerminal(ctrl),
 		monitoring:            domocks.NewMockMonitoringService(ctrl),
 		serverless:            domocks.NewMockServerlessService(ctrl),
 		appBuilderFactory:     builder.NewMockComponentBuilderFactory(ctrl),
diff --git a/integration/apps_test.go b/integration/apps_test.go
index 531df000..da3320ca 100644
--- a/integration/apps_test.go
+++ b/integration/apps_test.go
@@ -991,14 +991,14 @@ var _ = suite("apps/get-logs", func(t *testing.T, when spec.G, it spec.S) {
 				json.NewEncoder(buf).Encode(data)
 
 				err = c.WriteMessage(websocket.TextMessage, buf.Bytes())
-				if err != nil {
-					require.NoError(t, err)
-				}
+				require.NoError(t, err)
 
 				if i == finish {
 					break
 				}
 			}
+			err = c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
+			require.NoError(t, err)
 		}))
 		logsURL = wsServer.URL
 	})
@@ -1018,7 +1018,7 @@ var _ = suite("apps/get-logs", func(t *testing.T, when spec.G, it spec.S) {
 		)
 
 		output, err := cmd.CombinedOutput()
-		expect.NoError(err)
+		expect.NoError(err, "output: %v", string(output))
 
 		logLine := fmt.Sprintf("foo-service %v fake logs", now)
 		expectedOutput := fmt.Sprintf("%s\n%s\n%s\n%s\n%s", logLine, logLine, logLine, logLine, logLine)
@@ -1040,7 +1040,7 @@ var _ = suite("apps/get-logs", func(t *testing.T, when spec.G, it spec.S) {
 		)
 
 		output, err := cmd.CombinedOutput()
-		expect.NoError(err)
+		expect.NoError(err, "output: %v", string(output))
 
 		expectedOutput := "fake logs\nfake logs\nfake logs\nfake logs\nfake logs"
 		expect.Equal(expectedOutput, strings.TrimSpace(string(output)))
diff --git a/pkg/listen/listen_test.go b/pkg/listen/listen_test.go
index 9727be24..5f37db87 100644
--- a/pkg/listen/listen_test.go
+++ b/pkg/listen/listen_test.go
@@ -2,6 +2,7 @@ package listen
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -20,13 +21,23 @@ var (
 	upgrader = websocket.Upgrader{}
 )
 
-func wsHandler(t *testing.T) http.HandlerFunc {
+func wsHandler(t *testing.T, recvBuffer *bytes.Buffer) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		c, err := upgrader.Upgrade(w, r, nil)
 		require.NoError(t, err)
 		defer c.Close()
 		i := 0
 		finish := 5
+		go func() {
+			// Read messages from websocket and write to buffer
+			for {
+				_, message, err := c.ReadMessage()
+				if err != nil {
+					return
+				}
+				recvBuffer.Write(message)
+			}
+		}()
 		for {
 			// Give the Close test a chance to close before any sent
 			time.Sleep(time.Millisecond * 10)
@@ -38,7 +49,8 @@ func wsHandler(t *testing.T) http.HandlerFunc {
 				Message: fmt.Sprintf("%d\n", i),
 			}
 			buf := new(bytes.Buffer)
-			json.NewEncoder(buf).Encode(data)
+			err := json.NewEncoder(buf).Encode(data)
+			require.NoError(t, err)
 
 			err = c.WriteMessage(websocket.TextMessage, buf.Bytes())
 			require.NoError(t, err)
@@ -47,11 +59,13 @@ func wsHandler(t *testing.T) http.HandlerFunc {
 				break
 			}
 		}
+		err = c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
+		require.NoError(t, err)
 	}
 }
 
 func TestListener(t *testing.T) {
-	server := httptest.NewServer(wsHandler(t))
+	server := httptest.NewServer(wsHandler(t, nil))
 	defer server.Close()
 
 	u := "ws" + strings.TrimPrefix(server.URL, "http")
@@ -60,8 +74,8 @@ func TestListener(t *testing.T) {
 
 	buffer := &bytes.Buffer{}
 
-	listener := NewListener(url, "", nil, buffer)
-	err = listener.Start()
+	listener := NewListener(url, "", nil, buffer, nil)
+	err = listener.Listen(context.Background())
 	require.NoError(t, err)
 
 	want := `{"message":"1\n"}
@@ -74,7 +88,7 @@ func TestListener(t *testing.T) {
 }
 
 func TestListenerWithSchemaFunc(t *testing.T) {
-	server := httptest.NewServer(wsHandler(t))
+	server := httptest.NewServer(wsHandler(t, nil))
 	defer server.Close()
 
 	u := "ws" + strings.TrimPrefix(server.URL, "http")
@@ -96,8 +110,8 @@ func TestListenerWithSchemaFunc(t *testing.T) {
 		return r, nil
 	}
 
-	listener := NewListener(url, "", schemaFunc, buffer)
-	err = listener.Start()
+	listener := NewListener(url, "", schemaFunc, buffer, nil)
+	err = listener.Listen(context.Background())
 	require.NoError(t, err)
 
 	want := `1
@@ -109,20 +123,30 @@ func TestListenerWithSchemaFunc(t *testing.T) {
 	require.Equal(t, want, buffer.String())
 }
 
-func TestListenerStop(t *testing.T) {
-	server := httptest.NewServer(wsHandler(t))
+func TestListenerWithInput(t *testing.T) {
+	wsInBuf := &bytes.Buffer{}
+	server := httptest.NewServer(wsHandler(t, wsInBuf))
 	defer server.Close()
 
 	u := "ws" + strings.TrimPrefix(server.URL, "http")
 	url, err := url.Parse(u)
 	require.NoError(t, err)
 
-	buffer := &bytes.Buffer{}
-
-	listener := NewListener(url, "", nil, buffer)
-	go listener.Start()
-	// Stop before any messages have been sent
-	listener.Stop()
+	inputCh := make(chan []byte, 5)
+	for i := 0; i < 5; i++ {
+		inputCh <- []byte{byte('a' + i)}
+	}
+	wsOutBuf := &bytes.Buffer{}
+	listener := NewListener(url, "", nil, wsOutBuf, inputCh)
+	err = listener.Listen(context.Background())
+	require.NoError(t, err)
 
-	require.Equal(t, "", buffer.String())
+	want := `{"message":"1\n"}
+{"message":"2\n"}
+{"message":"3\n"}
+{"message":"4\n"}
+{"message":"5\n"}
+`
+	require.Equal(t, want, wsOutBuf.String())
+	require.Equal(t, "abcde", wsInBuf.String())
 }