diff --git a/main.go b/main.go index 0699ddec..0a295aca 100644 --- a/main.go +++ b/main.go @@ -9,6 +9,7 @@ import ( "context" "embed" "fmt" + "io/fs" "log" "net/http" "os" @@ -18,6 +19,7 @@ import ( "syscall" "time" + "github.com/google/uuid" "github.com/wavetermdev/thenextwave/pkg/blockstore" "github.com/wavetermdev/thenextwave/pkg/eventbus" "github.com/wavetermdev/thenextwave/pkg/service/blockservice" @@ -98,13 +100,56 @@ type waveAssetHandler struct { AssetHandler http.Handler } +func serveBlockFile(w http.ResponseWriter, r *http.Request) { + blockId := r.URL.Query().Get("blockid") + name := r.URL.Query().Get("name") + if _, err := uuid.Parse(blockId); err != nil { + http.Error(w, fmt.Sprintf("invalid blockid: %v", err), http.StatusBadRequest) + return + } + if name == "" { + http.Error(w, "name is required", http.StatusBadRequest) + return + + } + file, err := blockstore.GBS.Stat(r.Context(), blockId, name) + if err == fs.ErrNotExist { + http.NotFound(w, r) + return + } + if err != nil { + http.Error(w, fmt.Sprintf("error getting file info: %v", err), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", fmt.Sprintf("%d", file.Size)) + for offset := file.DataStartIdx(); offset < file.Size; offset += blockstore.DefaultPartDataSize { + _, data, err := blockstore.GBS.ReadAt(r.Context(), blockId, name, offset, blockstore.DefaultPartDataSize) + if err != nil { + if offset == 0 { + http.Error(w, fmt.Sprintf("error reading file: %v", err), http.StatusInternalServerError) + } else { + // nothing to do, the headers have already been sent + log.Printf("error reading file %s/%s @ %d: %v\n", blockId, name, offset, err) + } + return + } + w.Write(data) + } +} + func serveWaveUrls(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-cache") if r.URL.Path == "/wave/stream-file" { fileName := r.URL.Query().Get("path") fileName = wavebase.ExpandHomeDir(fileName) http.ServeFile(w, r, fileName) return } + if r.URL.Path == "/wave/blockfile" { + serveBlockFile(w, r) + return + } http.NotFound(w, r) } diff --git a/pkg/blockcontroller/blockcontroller.go b/pkg/blockcontroller/blockcontroller.go index d78f4edb..69f951dd 100644 --- a/pkg/blockcontroller/blockcontroller.go +++ b/pkg/blockcontroller/blockcontroller.go @@ -15,6 +15,7 @@ import ( "github.com/creack/pty" "github.com/wailsapp/wails/v3/pkg/application" + "github.com/wavetermdev/thenextwave/pkg/blockstore" "github.com/wavetermdev/thenextwave/pkg/eventbus" "github.com/wavetermdev/thenextwave/pkg/shellexec" "github.com/wavetermdev/thenextwave/pkg/wstore" @@ -86,7 +87,35 @@ func (bc *BlockController) Close() { } } +const DefaultTermMaxFileSize = 256 * 1024 + +func (bc *BlockController) handleShellProcData(data []byte, seqNum int) error { + ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout) + defer cancelFn() + err := blockstore.GBS.AppendData(ctx, bc.BlockId, "main", data) + if err != nil { + return fmt.Errorf("error appending to blockfile: %w", err) + } + eventbus.SendEvent(application.WailsEvent{ + Name: "block:ptydata", + Data: map[string]any{ + "blockid": bc.BlockId, + "blockfile": "main", + "ptydata": base64.StdEncoding.EncodeToString(data), + "seqnum": seqNum, + }, + }) + return nil +} + func (bc *BlockController) DoRunShellCommand(rc *RunShellOpts) error { + // create a circular blockfile for the output + ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second) + defer cancelFn() + err := blockstore.GBS.MakeFile(ctx, bc.BlockId, "main", nil, blockstore.FileOptsType{MaxSize: DefaultTermMaxFileSize, Circular: true}) + if err != nil && err != blockstore.ErrAlreadyExists { + return fmt.Errorf("error creating blockfile: %w", err) + } if bc.getShellProc() != nil { return nil } @@ -114,15 +143,13 @@ func (bc *BlockController) DoRunShellCommand(rc *RunShellOpts) error { for { nr, err := bc.ShellProc.Pty.Read(buf) seqNum++ - eventbus.SendEvent(application.WailsEvent{ - Name: "block:ptydata", - Data: map[string]any{ - "blockid": bc.BlockId, - "blockfile": "main", - "ptydata": base64.StdEncoding.EncodeToString(buf[:nr]), - "seqnum": seqNum, - }, - }) + if nr > 0 { + handleDataErr := bc.handleShellProcData(buf[:nr], seqNum) + if handleDataErr != nil { + log.Printf("error handling shell data: %v\n", handleDataErr) + break + } + } if err == io.EOF { break } diff --git a/pkg/blockstore/blockstore.go b/pkg/blockstore/blockstore.go index fcf19f9f..516cbaff 100644 --- a/pkg/blockstore/blockstore.go +++ b/pkg/blockstore/blockstore.go @@ -35,9 +35,9 @@ var GBS *BlockStore = &BlockStore{ } type FileOptsType struct { - MaxSize int64 - Circular bool - IJson bool + MaxSize int64 `json:"maxsize,omitempty"` + Circular bool `json:"circular,omitempty"` + IJson bool `json:"ijson,omitempty"` } type FileMeta = map[string]any @@ -55,6 +55,24 @@ type BlockFile struct { Meta FileMeta `json:"meta"` // only top-level keys can be updated (lower levels are immutable) } +// for regular files this is just Size +// for circular files this is min(Size, MaxSize) +func (f BlockFile) DataLength() int64 { + if f.Opts.Circular { + return minInt64(f.Size, f.Opts.MaxSize) + } + return f.Size +} + +// for regular files this is just 0 +// for circular files this is the index of the first byte of data we have +func (f BlockFile) DataStartIdx() int64 { + if f.Opts.Circular && f.Size > f.Opts.MaxSize { + return f.Size - f.Opts.MaxSize + } + return 0 +} + // this works because lower levels are immutable func copyMeta(meta FileMeta) FileMeta { newMeta := make(FileMeta) diff --git a/pkg/blockstore/blockstore_dbops.go b/pkg/blockstore/blockstore_dbops.go index 5b9a4196..112c25aa 100644 --- a/pkg/blockstore/blockstore_dbops.go +++ b/pkg/blockstore/blockstore_dbops.go @@ -11,10 +11,16 @@ import ( "github.com/wavetermdev/waveterm/wavesrv/pkg/dbutil" ) +var ErrAlreadyExists = fmt.Errorf("file already exists") + func dbInsertFile(ctx context.Context, file *BlockFile) error { // will fail if file already exists return WithTx(ctx, func(tx *TxWrap) error { - query := "INSERT INTO db_block_file (blockid, name, size, createdts, modts, opts, meta) VALUES (?, ?, ?, ?, ?, ?, ?)" + query := "SELECT blockid FROM db_block_file WHERE blockid = ? AND name = ?" + if tx.Exists(query, file.BlockId, file.Name) { + return ErrAlreadyExists + } + query = "INSERT INTO db_block_file (blockid, name, size, createdts, modts, opts, meta) VALUES (?, ?, ?, ?, ?, ?, ?)" tx.Exec(query, file.BlockId, file.Name, file.Size, file.CreatedTs, file.ModTs, dbutil.QuickJson(file.Opts), dbutil.QuickJson(file.Meta)) return nil })