Add S3 fileshare implementation, improve cp behavior (#1896)

Adds the S3 `fileshare` implementation

This also updates `wsh file cp` so it behaves more like `cp` for things
like copying directories and directory entries. It's not meant to align
with `cp` on everything, though. Our `wsh cp` will be recursive and will
create intermediate directories by default.

This also adds new aliases for `wsh view`: `wsh preview` and `wsh open`

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: sawka <mike@commandline.dev>
Co-authored-by: Sylvia Crowe <software@oneirocosm.com>

Author: 4 people

cmd/wsh/cmd/wshcmd-connserver.go

@@ -21,6 +21,7 @@ import (
 	"github.com/wavetermdev/waveterm/pkg/remote/fileshare/wshfs"
 	"github.com/wavetermdev/waveterm/pkg/util/packetparser"
 	"github.com/wavetermdev/waveterm/pkg/util/sigutil"
+	"github.com/wavetermdev/waveterm/pkg/util/utilfn"
 	"github.com/wavetermdev/waveterm/pkg/wavebase"
 	"github.com/wavetermdev/waveterm/pkg/wshrpc"
 	"github.com/wavetermdev/waveterm/pkg/wshrpc/wshclient"
@@ -162,9 +163,7 @@ func serverRunRouter(jwtToken string) error {
 		// just ignore and drain the rawCh (stdin)
 		// when stdin is closed, shutdown
 		defer wshutil.DoShutdown("", 0, true)
-		for range rawCh {
-			// ignore
-		}
+		utilfn.DrainChannelSafe(rawCh, "serverRunRouter:stdin")
 	}()
 	go func() {
 		for msg := range termProxy.FromRemoteCh {

cmd/wsh/cmd/wshcmd-file-util.go

@@ -4,13 +4,15 @@
 package cmd
 
 import (
+	"context"
 	"encoding/base64"
 	"fmt"
 	"io"
 	"io/fs"
 	"strings"
 
 	"github.com/wavetermdev/waveterm/pkg/remote/connparse"
+	"github.com/wavetermdev/waveterm/pkg/remote/fileshare/fsutil"
 	"github.com/wavetermdev/waveterm/pkg/util/fileutil"
 	"github.com/wavetermdev/waveterm/pkg/util/wavefileutil"
 	"github.com/wavetermdev/waveterm/pkg/wshrpc"
@@ -27,15 +29,15 @@ func convertNotFoundErr(err error) error {
 	return err
 }
 
-func ensureFile(origName string, fileData wshrpc.FileData) (*wshrpc.FileInfo, error) {
-	info, err := wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: DefaultFileTimeout})
+func ensureFile(fileData wshrpc.FileData) (*wshrpc.FileInfo, error) {
+	info, err := wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
 	err = convertNotFoundErr(err)
 	if err == fs.ErrNotExist {
-		err = wshclient.FileCreateCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: DefaultFileTimeout})
+		err = wshclient.FileCreateCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
 		if err != nil {
 			return nil, fmt.Errorf("creating file: %w", err)
 		}
-		info, err = wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: DefaultFileTimeout})
+		info, err = wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
 		if err != nil {
 			return nil, fmt.Errorf("getting file info: %w", err)
 		}
@@ -51,12 +53,12 @@ func streamWriteToFile(fileData wshrpc.FileData, reader io.Reader) error {
 	// First truncate the file with an empty write
 	emptyWrite := fileData
 	emptyWrite.Data64 = ""
-	err := wshclient.FileWriteCommand(RpcClient, emptyWrite, &wshrpc.RpcOpts{Timeout: DefaultFileTimeout})
+	err := wshclient.FileWriteCommand(RpcClient, emptyWrite, &wshrpc.RpcOpts{Timeout: fileTimeout})
 	if err != nil {
 		return fmt.Errorf("initializing file with empty write: %w", err)
 	}
 
-	const chunkSize = 32 * 1024 // 32KB chunks
+	const chunkSize = wshrpc.FileChunkSize // 32KB chunks
 	buf := make([]byte, chunkSize)
 	totalWritten := int64(0)
 
@@ -89,40 +91,9 @@ func streamWriteToFile(fileData wshrpc.FileData, reader io.Reader) error {
 	return nil
 }
 
-func streamReadFromFile(fileData wshrpc.FileData, size int64, writer io.Writer) error {
-	const chunkSize = 32 * 1024 // 32KB chunks
-	for offset := int64(0); offset < size; offset += chunkSize {
-		// Calculate the length of this chunk
-		length := chunkSize
-		if offset+int64(length) > size {
-			length = int(size - offset)
-		}
-
-		// Set up the ReadAt request
-		fileData.At = &wshrpc.FileDataAt{
-			Offset: offset,
-			Size:   length,
-		}
-
-		// Read the chunk
-		data, err := wshclient.FileReadCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: int64(fileTimeout)})
-		if err != nil {
-			return fmt.Errorf("reading chunk at offset %d: %w", offset, err)
-		}
-
-		// Decode and write the chunk
-		chunk, err := base64.StdEncoding.DecodeString(data.Data64)
-		if err != nil {
-			return fmt.Errorf("decoding chunk at offset %d: %w", offset, err)
-		}
-
-		_, err = writer.Write(chunk)
-		if err != nil {
-			return fmt.Errorf("writing chunk at offset %d: %w", offset, err)
-		}
-	}
-
-	return nil
+func streamReadFromFile(ctx context.Context, fileData wshrpc.FileData, writer io.Writer) error {
+	ch := wshclient.FileReadStreamCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
+	return fsutil.ReadFileStreamToWriter(ctx, ch, writer)
 }
 
 type fileListResult struct {

cmd/wsh/cmd/wshcmd-file.go

@@ -9,7 +9,6 @@ import (
 	"encoding/base64"
 	"fmt"
 	"io"
-	"io/fs"
 	"log"
 	"os"
 	"path"
@@ -31,8 +30,7 @@ const (
 	WaveFileScheme = "wavefile"
 	WaveFilePrefix = "wavefile://"
 
-	DefaultFileTimeout = 5000
-	TimeoutYear        = int64(365) * 24 * 60 * 60 * 1000
+	TimeoutYear = int64(365) * 24 * 60 * 60 * 1000
 
 	UriHelpText = `
 
@@ -83,12 +81,12 @@ Wave Terminal is capable of managing files from remote SSH hosts, S3-compatible
 systems, and the internal Wave filesystem. Files are addressed via URIs, which
 vary depending on the storage system.` + UriHelpText}
 
-var fileTimeout int
+var fileTimeout int64
 
 func init() {
 	rootCmd.AddCommand(fileCmd)
 
-	fileCmd.PersistentFlags().IntVarP(&fileTimeout, "timeout", "t", 15000, "timeout in milliseconds for long operations")
+	fileCmd.PersistentFlags().Int64VarP(&fileTimeout, "timeout", "t", 15000, "timeout in milliseconds for long operations")
 
 	fileListCmd.Flags().BoolP("recursive", "r", false, "list subdirectories recursively")
 	fileListCmd.Flags().BoolP("long", "l", false, "use long listing format")
@@ -103,7 +101,6 @@ func init() {
 	fileCmd.AddCommand(fileInfoCmd)
 	fileCmd.AddCommand(fileAppendCmd)
 	fileCpCmd.Flags().BoolP("merge", "m", false, "merge directories")
-	fileCpCmd.Flags().BoolP("recursive", "r", false, "copy directories recursively")
 	fileCpCmd.Flags().BoolP("force", "f", false, "force overwrite of existing files")
 	fileCmd.AddCommand(fileCpCmd)
 	fileMvCmd.Flags().BoolP("recursive", "r", false, "move directories recursively")
@@ -174,7 +171,7 @@ var fileAppendCmd = &cobra.Command{
 var fileCpCmd = &cobra.Command{
 	Use:     "cp [source-uri] [destination-uri]" + UriHelpText,
 	Aliases: []string{"copy"},
-	Short:   "copy files between storage systems",
+	Short:   "copy files between storage systems, recursively if needed",
 	Long:    "Copy files between different storage systems." + UriHelpText,
 	Example: "  wsh file cp wavefile://block/config.txt ./local-config.txt\n  wsh file cp ./local-config.txt wavefile://block/config.txt\n  wsh file cp wsh://user@ec2/home/user/config.txt wavefile://client/config.txt",
 	Args:    cobra.ExactArgs(2),
@@ -202,17 +199,7 @@ func fileCatRun(cmd *cobra.Command, args []string) error {
 		Info: &wshrpc.FileInfo{
 			Path: path}}
 
-	// Get file info first to check existence and get size
-	info, err := wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: 2000})
-	err = convertNotFoundErr(err)
-	if err == fs.ErrNotExist {
-		return fmt.Errorf("%s: no such file", path)
-	}
-	if err != nil {
-		return fmt.Errorf("getting file info: %w", err)
-	}
-
-	err = streamReadFromFile(fileData, info.Size, os.Stdout)
+	err = streamReadFromFile(cmd.Context(), fileData, os.Stdout)
 	if err != nil {
 		return fmt.Errorf("reading file: %w", err)
 	}
@@ -229,7 +216,7 @@ func fileInfoRun(cmd *cobra.Command, args []string) error {
 		Info: &wshrpc.FileInfo{
 			Path: path}}
 
-	info, err := wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: DefaultFileTimeout})
+	info, err := wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
 	err = convertNotFoundErr(err)
 	if err != nil {
 		return fmt.Errorf("getting file info: %w", err)
@@ -265,20 +252,8 @@ func fileRmRun(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return err
 	}
-	fileData := wshrpc.FileData{
-		Info: &wshrpc.FileInfo{
-			Path: path}}
-
-	_, err = wshclient.FileInfoCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: DefaultFileTimeout})
-	err = convertNotFoundErr(err)
-	if err == fs.ErrNotExist {
-		return fmt.Errorf("%s: no such file", path)
-	}
-	if err != nil {
-		return fmt.Errorf("getting file info: %w", err)
-	}
 
-	err = wshclient.FileDeleteCommand(RpcClient, wshrpc.CommandDeleteFileData{Path: path, Recursive: recursive}, &wshrpc.RpcOpts{Timeout: DefaultFileTimeout})
+	err = wshclient.FileDeleteCommand(RpcClient, wshrpc.CommandDeleteFileData{Path: path, Recursive: recursive}, &wshrpc.RpcOpts{Timeout: fileTimeout})
 	if err != nil {
 		return fmt.Errorf("removing file: %w", err)
 	}
@@ -295,14 +270,31 @@ func fileWriteRun(cmd *cobra.Command, args []string) error {
 		Info: &wshrpc.FileInfo{
 			Path: path}}
 
-	_, err = ensureFile(path, fileData)
+	capability, err := wshclient.FileShareCapabilityCommand(RpcClient, fileData.Info.Path, &wshrpc.RpcOpts{Timeout: fileTimeout})
 	if err != nil {
-		return err
+		return fmt.Errorf("getting fileshare capability: %w", err)
 	}
-
-	err = streamWriteToFile(fileData, WrappedStdin)
-	if err != nil {
-		return fmt.Errorf("writing file: %w", err)
+	if capability.CanAppend {
+		err = streamWriteToFile(fileData, WrappedStdin)
+		if err != nil {
+			return fmt.Errorf("writing file: %w", err)
+		}
+	} else {
+		buf := make([]byte, MaxFileSize)
+		n, err := WrappedStdin.Read(buf)
+		if err != nil && err != io.EOF {
+			return fmt.Errorf("reading input: %w", err)
+		}
+		if int64(n) == MaxFileSize {
+			if _, err := WrappedStdin.Read(make([]byte, 1)); err != io.EOF {
+				return fmt.Errorf("input exceeds maximum file size of %d bytes", MaxFileSize)
+			}
+		}
+		fileData.Data64 = base64.StdEncoding.EncodeToString(buf[:n])
+		err = wshclient.FileWriteCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
+		if err != nil {
+			return fmt.Errorf("writing file: %w", err)
+		}
 	}
 
 	return nil
@@ -317,7 +309,7 @@ func fileAppendRun(cmd *cobra.Command, args []string) error {
 		Info: &wshrpc.FileInfo{
 			Path: path}}
 
-	info, err := ensureFile(path, fileData)
+	info, err := ensureFile(fileData)
 	if err != nil {
 		return err
 	}
@@ -346,7 +338,7 @@ func fileAppendRun(cmd *cobra.Command, args []string) error {
 
 		if buf.Len() >= 8192 { // 8KB batch size
 			fileData.Data64 = base64.StdEncoding.EncodeToString(buf.Bytes())
-			err = wshclient.FileAppendCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: int64(fileTimeout)})
+			err = wshclient.FileAppendCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
 			if err != nil {
 				return fmt.Errorf("appending to file: %w", err)
 			}
@@ -357,7 +349,7 @@ func fileAppendRun(cmd *cobra.Command, args []string) error {
 
 	if buf.Len() > 0 {
 		fileData.Data64 = base64.StdEncoding.EncodeToString(buf.Bytes())
-		err = wshclient.FileAppendCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: int64(fileTimeout)})
+		err = wshclient.FileAppendCommand(RpcClient, fileData, &wshrpc.RpcOpts{Timeout: fileTimeout})
 		if err != nil {
 			return fmt.Errorf("appending to file: %w", err)
 		}
@@ -398,10 +390,6 @@ func getTargetPath(src, dst string) (string, error) {
 
 func fileCpRun(cmd *cobra.Command, args []string) error {
 	src, dst := args[0], args[1]
-	recursive, err := cmd.Flags().GetBool("recursive")
-	if err != nil {
-		return err
-	}
 	merge, err := cmd.Flags().GetBool("merge")
 	if err != nil {
 		return err
@@ -419,9 +407,9 @@ func fileCpRun(cmd *cobra.Command, args []string) error {
 	if err != nil {
 		return fmt.Errorf("unable to parse dest path: %w", err)
 	}
-	log.Printf("Copying %s to %s; recursive: %v, merge: %v, force: %v", srcPath, destPath, recursive, merge, force)
+	log.Printf("Copying %s to %s; merge: %v, force: %v", srcPath, destPath, merge, force)
 	rpcOpts := &wshrpc.RpcOpts{Timeout: TimeoutYear}
-	err = wshclient.FileCopyCommand(RpcClient, wshrpc.CommandFileCopyData{SrcUri: srcPath, DestUri: destPath, Opts: &wshrpc.FileCopyOpts{Recursive: recursive, Merge: merge, Overwrite: force, Timeout: TimeoutYear}}, rpcOpts)
+	err = wshclient.FileCopyCommand(RpcClient, wshrpc.CommandFileCopyData{SrcUri: srcPath, DestUri: destPath, Opts: &wshrpc.FileCopyOpts{Merge: merge, Overwrite: force, Timeout: TimeoutYear}}, rpcOpts)
 	if err != nil {
 		return fmt.Errorf("copying file: %w", err)
 	}
@@ -449,7 +437,7 @@ func fileMvRun(cmd *cobra.Command, args []string) error {
 	}
 	log.Printf("Moving %s to %s; recursive: %v, force: %v", srcPath, destPath, recursive, force)
 	rpcOpts := &wshrpc.RpcOpts{Timeout: TimeoutYear}
-	err = wshclient.FileMoveCommand(RpcClient, wshrpc.CommandFileCopyData{SrcUri: srcPath, DestUri: destPath, Opts: &wshrpc.FileCopyOpts{Recursive: recursive, Overwrite: force, Timeout: TimeoutYear}}, rpcOpts)
+	err = wshclient.FileMoveCommand(RpcClient, wshrpc.CommandFileCopyData{SrcUri: srcPath, DestUri: destPath, Opts: &wshrpc.FileCopyOpts{Overwrite: force, Timeout: TimeoutYear, Recursive: recursive}}, rpcOpts)
 	if err != nil {
 		return fmt.Errorf("moving file: %w", err)
 	}
@@ -562,10 +550,7 @@ func fileListRun(cmd *cobra.Command, args []string) error {
 
 	filesChan := wshclient.FileListStreamCommand(RpcClient, wshrpc.FileListData{Path: path, Opts: &wshrpc.FileListOpts{All: recursive}}, &wshrpc.RpcOpts{Timeout: 2000})
 	// Drain the channel when done
-	defer func() {
-		for range filesChan {
-		}
-	}()
+	defer utilfn.DrainChannelSafe(filesChan, "fileListRun")
 	if longForm {
 		return filePrintLong(filesChan)
 	}

cmd/wsh/cmd/wshcmd-view.go

@@ -19,6 +19,7 @@ var viewMagnified bool
 
 var viewCmd = &cobra.Command{
 	Use:     "view {file|directory|URL}",
+	Aliases: []string{"preview", "open"},
 	Short:   "preview/edit a file or directory",
 	RunE:    viewRun,
 	PreRunE: preRunSetupRpcClient,

frontend/app/block/blockframe.tsx

@@ -604,7 +604,8 @@ const BlockFrame_Default_Component = (props: BlockFrameProps) => {
                     "--magnified-block-blur": `${magnifiedBlockBlur}px`,
                 } as React.CSSProperties
             }
-            {...({ inert: preview ? "1" : undefined } as any)} // sets insert="1" ... but tricks TS into accepting it
+            // @ts-ignore: inert does exist in the DOM, just not in react
+            inert={preview ? "1" : undefined} //
         >
             <BlockMask nodeModel={nodeModel} />
             {preview || viewModel == null ? null : (

frontend/app/modals/conntypeahead.tsx

@@ -377,13 +377,10 @@ const ChangeConnectionBlockModal = React.memo(
                     // typeahead was opened. good candidate for verbose log level.
                     //console.log("unable to load wsl list from backend. using blank list: ", e)
                 });
-            /////////
-            // TODO-S3
-            // this needs an rpc call to generate a list of s3 profiles
-            const newS3List = [];
-            setS3List(newS3List);
-            /////////
-        }, [changeConnModalOpen, setConnList]);
+            RpcApi.ConnListAWSCommand(TabRpcClient, { timeout: 2000 })
+                .then((s3List) => setS3List(s3List ?? []))
+                .catch((e) => console.log("unable to load s3 list from backend:", e));
+        }, [changeConnModalOpen]);
 
         const changeConnection = React.useCallback(
             async (connName: string) => {
@@ -393,10 +390,13 @@ const ChangeConnectionBlockModal = React.memo(
                 if (connName == blockData?.meta?.connection) {
                     return;
                 }
+                const isAws = connName?.startsWith("aws:");
                 const oldCwd = blockData?.meta?.file ?? "";
                 let newCwd: string;
                 if (oldCwd == "") {
                     newCwd = "";
+                } else if (isAws) {
+                    newCwd = "/";
                 } else {
                     newCwd = "~";
                 }

frontend/app/store/global.ts

@@ -672,6 +672,17 @@ function getConnStatusAtom(conn: string): PrimitiveAtom<ConnStatus> {
                 wshenabled: false,
             };
             rtn = atom(connStatus);
+        } else if (conn.startsWith("aws:")) {
+            const connStatus: ConnStatus = {
+                connection: conn,
+                connected: true,
+                error: null,
+                status: "connected",
+                hasconnected: true,
+                activeconnnum: 0,
+                wshenabled: false,
+            };
+            rtn = atom(connStatus);
         } else {
             const connStatus: ConnStatus = {
                 connection: conn,