From 2d63b1da9603aa01627e1fbf0e86d0fffb72a78e Mon Sep 17 00:00:00 2001 From: sawka Date: Sun, 30 Jul 2023 17:16:43 -0700 Subject: [PATCH] checkpoint on line migration --- db/migrations/000020_linecmd.down.sql | 2 + db/migrations/000020_linecmd.up.sql | 74 ++++++++++ pkg/cmdrunner/cmdrunner.go | 62 ++++----- pkg/pcloud/pcloud.go | 16 +-- pkg/pcloud/pclouddata.go | 56 ++++---- pkg/remote/remote.go | 41 +++--- pkg/rtnstate/rtnstate.go | 4 +- pkg/scbase/scbase.go | 10 +- pkg/sstore/dbops.go | 187 ++++++++++---------------- pkg/sstore/fileops.go | 40 +++--- pkg/sstore/migrate.go | 13 +- pkg/sstore/quick.go | 1 + pkg/sstore/sstore.go | 99 +++++++------- pkg/sstore/sstore_migrate.go | 77 ++++++++++- pkg/sstore/updatebus.go | 17 +-- 15 files changed, 390 insertions(+), 309 deletions(-) create mode 100644 db/migrations/000020_linecmd.down.sql create mode 100644 db/migrations/000020_linecmd.up.sql diff --git a/db/migrations/000020_linecmd.down.sql b/db/migrations/000020_linecmd.down.sql new file mode 100644 index 00000000..6332dc5b --- /dev/null +++ b/db/migrations/000020_linecmd.down.sql @@ -0,0 +1,2 @@ +-- invalid, will throw an error, cannot migrate down +SELECT x; diff --git a/db/migrations/000020_linecmd.up.sql b/db/migrations/000020_linecmd.up.sql new file mode 100644 index 00000000..7e2332e9 --- /dev/null +++ b/db/migrations/000020_linecmd.up.sql @@ -0,0 +1,74 @@ +-- remove cmdid from line, history, and cmd (use lineid everywhere) + +CREATE TABLE cmd_new ( + screenid varchar(36) NOT NULL, + lineid varchar(36) NOT NULL, + remoteownerid varchar(36) NOT NULL, + remoteid varchar(36) NOT NULL, + remotename varchar(50) NOT NULL, + cmdstr text NOT NULL, + rawcmdstr text NOT NULL, + festate json NOT NULL, + statebasehash varchar(36) NOT NULL, + statediffhasharr json NOT NULL, + termopts json NOT NULL, + origtermopts json NOT NULL, + status varchar(10) NOT NULL, + cmdpid int NOT NULL, + remotepid int NOT NULL, + donets bigint NOT NULL, + exitcode int NOT NULL, + durationms int NOT NULL, + rtnstate boolean NOT NULL, + rtnbasehash varchar(36) NOT NULL, + rtndiffhasharr json NOT NULL, + runout json NOT NULL, + PRIMARY KEY (screenid, lineid) +); + +DROP TABLE IF EXISTS cmd_migrate; + +CREATE TABLE cmd_migrate ( + screenid varchar(36) NOT NULL, + lineid varchar(36) NOT NULL, + cmdid varchar(36) NOT NULL, + PRIMARY KEY (screenid, lineid) +); + +INSERT INTO cmd_migrate +SELECT screenid, lineid, cmdid +FROM line; + +INSERT INTO cmd_new +SELECT + c.screenid, + l.lineid, + c.remoteownerid, + c.remoteid, + c.remotename, + c.cmdstr, + c.rawcmdstr, + c.festate, + c.statebasehash, + c.statediffhasharr, + c.termopts, + c.origtermopts, + c.status, + coalesce(json_extract(startpk, '$.pid'), 0), + coalesce(json_extract(startpk, '$.mshellpid'), 0), + coalesce(json_extract(doneinfo, '$.ts'), 0), + coalesce(json_extract(doneinfo, '$.exitcode'), 0), + coalesce(json_extract(doneinfo, '$.durationms'), 0), + c.rtnstate, + c.rtnbasehash, + c.rtndiffhasharr, + c.runout +FROM cmd c +JOIN line l ON (l.cmdid = c.cmdid); + +DROP TABLE cmd; + +ALTER TABLE cmd_new RENAME TO cmd; + +ALTER TABLE history DROP COLUMN cmdid; +ALTER TABLE line DROP COLUMN cmdid; diff --git a/pkg/cmdrunner/cmdrunner.go b/pkg/cmdrunner/cmdrunner.go index d0f40778..6d89f487 100644 --- a/pkg/cmdrunner/cmdrunner.go +++ b/pkg/cmdrunner/cmdrunner.go @@ -17,7 +17,6 @@ import ( "time" "unicode" - "github.com/google/uuid" "github.com/commandlinedev/apishell/pkg/base" "github.com/commandlinedev/apishell/pkg/packet" "github.com/commandlinedev/apishell/pkg/shexec" @@ -29,6 +28,7 @@ import ( "github.com/commandlinedev/prompt-server/pkg/scpacket" "github.com/commandlinedev/prompt-server/pkg/sstore" "github.com/commandlinedev/prompt-server/pkg/utilfn" + "github.com/google/uuid" ) const ( @@ -110,7 +110,6 @@ type SetVarScope struct { type historyContextType struct { LineId string LineNum int64 - CmdId string RemotePtr *sstore.RemotePtrType } @@ -527,7 +526,6 @@ func addToHistory(ctx context.Context, pk *scpacket.FeCommandPacketType, history LineId: historyContext.LineId, LineNum: historyContext.LineNum, HadError: hadError, - CmdId: historyContext.CmdId, CmdStr: cmdStr, IsMetaCmd: isMetaCmd, Incognito: isIncognito, @@ -1325,7 +1323,7 @@ func writeErrorToPty(cmd *sstore.CmdType, errStr string, outputPos int64) { } errCtx, cancelFn := context.WithTimeout(context.Background(), 5*time.Second) defer cancelFn() - update, err := sstore.AppendToCmdPtyBlob(errCtx, cmd.ScreenId, cmd.CmdId, errBytes, outputPos) + update, err := sstore.AppendToCmdPtyBlob(errCtx, cmd.ScreenId, cmd.LineId, errBytes, outputPos) if err != nil { log.Printf("error writing ptyupdate for openai response: %v\n", err) return @@ -1339,7 +1337,7 @@ func writePacketToPty(ctx context.Context, cmd *sstore.CmdType, pk packet.Packet if err != nil { return err } - update, err := sstore.AppendToCmdPtyBlob(ctx, cmd.ScreenId, cmd.CmdId, outBytes, *outputPos) + update, err := sstore.AppendToCmdPtyBlob(ctx, cmd.ScreenId, cmd.LineId, outBytes, *outputPos) if err != nil { return err } @@ -1364,18 +1362,17 @@ func doOpenAICompletion(cmd *sstore.CmdType, opts *sstore.OpenAIOptsType, prompt } duration := time.Since(startTime) cmdStatus := sstore.CmdStatusDone - var exitCode int64 + var exitCode int if hadError { cmdStatus = sstore.CmdStatusError exitCode = 1 } - doneInfo := &sstore.CmdDoneInfo{ - Ts: time.Now().UnixMilli(), - ExitCode: exitCode, - DurationMs: duration.Milliseconds(), - } - ck := base.MakeCommandKey(cmd.ScreenId, cmd.CmdId) - update, err := sstore.UpdateCmdDoneInfo(context.Background(), ck, doneInfo, cmdStatus) + ck := base.MakeCommandKey(cmd.ScreenId, cmd.LineId) + donePk := packet.MakeCmdDonePacket(ck) + donePk.Ts = time.Now().UnixMilli() + donePk.ExitCode = exitCode + donePk.DurationMs = duration.Milliseconds() + update, err := sstore.UpdateCmdDoneInfo(context.Background(), ck, donePk, cmdStatus) if err != nil { // nothing to do log.Printf("error updating cmddoneinfo (in openai): %v\n", err) @@ -1414,18 +1411,17 @@ func doOpenAIStreamCompletion(cmd *sstore.CmdType, opts *sstore.OpenAIOptsType, } duration := time.Since(startTime) cmdStatus := sstore.CmdStatusDone - var exitCode int64 + var exitCode int if hadError { cmdStatus = sstore.CmdStatusError exitCode = 1 } - doneInfo := &sstore.CmdDoneInfo{ - Ts: time.Now().UnixMilli(), - ExitCode: exitCode, - DurationMs: duration.Milliseconds(), - } - ck := base.MakeCommandKey(cmd.ScreenId, cmd.CmdId) - update, err := sstore.UpdateCmdDoneInfo(context.Background(), ck, doneInfo, cmdStatus) + ck := base.MakeCommandKey(cmd.ScreenId, cmd.LineId) + donePk := packet.MakeCmdDonePacket(ck) + donePk.Ts = time.Now().UnixMilli() + donePk.ExitCode = exitCode + donePk.DurationMs = duration.Milliseconds() + update, err := sstore.UpdateCmdDoneInfo(context.Background(), ck, donePk, cmdStatus) if err != nil { // nothing to do log.Printf("error updating cmddoneinfo (in openai): %v\n", err) @@ -1545,14 +1541,12 @@ func CrCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sstore.Up func makeDynCmd(ctx context.Context, metaCmd string, ids resolvedIds, cmdStr string, termOpts sstore.TermOpts) (*sstore.CmdType, error) { cmd := &sstore.CmdType{ ScreenId: ids.ScreenId, - CmdId: scbase.GenPromptUUID(), + LineId: scbase.GenPromptUUID(), CmdStr: cmdStr, RawCmdStr: cmdStr, Remote: ids.Remote.RemotePtr, TermOpts: termOpts, Status: sstore.CmdStatusRunning, - StartPk: nil, - DoneInfo: nil, RunOut: nil, } if ids.Remote.StatePtr != nil { @@ -1561,7 +1555,7 @@ func makeDynCmd(ctx context.Context, metaCmd string, ids resolvedIds, cmdStr str if ids.Remote.FeState != nil { cmd.FeState = ids.Remote.FeState } - err := sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.CmdId, cmd.TermOpts.MaxPtySize) + err := sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.LineId, cmd.TermOpts.MaxPtySize) if err != nil { // TODO tricky error since the command was a success, but we can't show the output return nil, fmt.Errorf("cannot create local ptyout file for %s command: %w", metaCmd, err) @@ -1572,14 +1566,12 @@ func makeDynCmd(ctx context.Context, metaCmd string, ids resolvedIds, cmdStr str func makeStaticCmd(ctx context.Context, metaCmd string, ids resolvedIds, cmdStr string, cmdOutput []byte) (*sstore.CmdType, error) { cmd := &sstore.CmdType{ ScreenId: ids.ScreenId, - CmdId: scbase.GenPromptUUID(), + LineId: scbase.GenPromptUUID(), CmdStr: cmdStr, RawCmdStr: cmdStr, Remote: ids.Remote.RemotePtr, TermOpts: sstore.TermOpts{Rows: shexec.DefaultTermRows, Cols: shexec.DefaultTermCols, FlexRows: true, MaxPtySize: remote.DefaultMaxPtySize}, Status: sstore.CmdStatusDone, - StartPk: nil, - DoneInfo: nil, RunOut: nil, } if ids.Remote.StatePtr != nil { @@ -1588,13 +1580,13 @@ func makeStaticCmd(ctx context.Context, metaCmd string, ids resolvedIds, cmdStr if ids.Remote.FeState != nil { cmd.FeState = ids.Remote.FeState } - err := sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.CmdId, cmd.TermOpts.MaxPtySize) + err := sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.LineId, cmd.TermOpts.MaxPtySize) if err != nil { // TODO tricky error since the command was a success, but we can't show the output return nil, fmt.Errorf("cannot create local ptyout file for %s command: %w", metaCmd, err) } // can ignore ptyupdate - _, err = sstore.AppendToCmdPtyBlob(ctx, ids.ScreenId, cmd.CmdId, cmdOutput, 0) + _, err = sstore.AppendToCmdPtyBlob(ctx, ids.ScreenId, cmd.LineId, cmdOutput, 0) if err != nil { // TODO tricky error since the command was a success, but we can't show the output return nil, fmt.Errorf("cannot append to local ptyout file for %s command: %v", metaCmd, err) @@ -1644,7 +1636,6 @@ func updateHistoryContext(ctx context.Context, line *sstore.LineType, cmd *sstor hctx.LineNum = line.LineNum } if cmd != nil { - hctx.CmdId = cmd.CmdId hctx.RemotePtr = &cmd.Remote } } @@ -2487,7 +2478,7 @@ func splitLinesForInfo(str string) []string { func resizeRunningCommand(ctx context.Context, cmd *sstore.CmdType, newCols int) error { siPk := packet.MakeSpecialInputPacket() - siPk.CK = base.MakeCommandKey(cmd.ScreenId, cmd.CmdId) + siPk.CK = base.MakeCommandKey(cmd.ScreenId, cmd.LineId) siPk.WinSize = &packet.WinSize{Rows: int(cmd.TermOpts.Rows), Cols: newCols} msh := remote.GetRemoteById(cmd.Remote.RemoteId) if msh == nil { @@ -2499,7 +2490,7 @@ func resizeRunningCommand(ctx context.Context, cmd *sstore.CmdType, newCols int) } newTermOpts := cmd.TermOpts newTermOpts.Cols = int64(newCols) - err = sstore.UpdateCmdTermOpts(ctx, cmd.ScreenId, cmd.CmdId, newTermOpts) + err = sstore.UpdateCmdTermOpts(ctx, cmd.ScreenId, cmd.LineId, newTermOpts) if err != nil { return err } @@ -2963,7 +2954,6 @@ func LineShowCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sst buf.WriteString(fmt.Sprintf(" %-15s %s\n", "renderer", "terminal")) } if cmd != nil { - buf.WriteString(fmt.Sprintf(" %-15s %s\n", "cmdid", cmd.CmdId)) buf.WriteString(fmt.Sprintf(" %-15s %s\n", "remote", cmd.Remote.MakeFullRemoteRef())) buf.WriteString(fmt.Sprintf(" %-15s %s\n", "status", cmd.Status)) if cmd.FeState["cwd"] != "" { @@ -2976,7 +2966,7 @@ func LineShowCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sst if cmd.RtnState { buf.WriteString(fmt.Sprintf(" %-15s %s\n", "rtnstate", "true")) } - stat, _ := sstore.StatCmdPtyFile(ctx, cmd.ScreenId, cmd.CmdId) + stat, _ := sstore.StatCmdPtyFile(ctx, cmd.ScreenId, cmd.LineId) if stat == nil { buf.WriteString(fmt.Sprintf(" %-15s %s\n", "file", "-")) } else { @@ -3073,7 +3063,7 @@ func SignalCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sstor return nil, fmt.Errorf("cannot send signal, remote is not connected") } siPk := packet.MakeSpecialInputPacket() - siPk.CK = base.MakeCommandKey(cmd.ScreenId, cmd.CmdId) + siPk.CK = base.MakeCommandKey(cmd.ScreenId, cmd.LineId) siPk.SigName = sigArg err = msh.SendSpecialInput(siPk) if err != nil { diff --git a/pkg/pcloud/pcloud.go b/pkg/pcloud/pcloud.go index 5d6c14bd..5e8f1e4b 100644 --- a/pkg/pcloud/pcloud.go +++ b/pkg/pcloud/pcloud.go @@ -299,34 +299,34 @@ func makeWebShareUpdate(ctx context.Context, update *sstore.ScreenUpdateType) (* } rtn.TermOpts = &cmd.TermOpts - case sstore.UpdateType_CmdDoneInfo: + case sstore.UpdateType_CmdExitCode, sstore.UpdateType_CmdDurationMs: _, cmd, err := sstore.GetLineCmdByLineId(ctx, update.ScreenId, update.LineId) if err != nil || cmd == nil { return nil, fmt.Errorf("error getting cmd: %v", defaultError(err, "not found")) } - rtn.DoneInfo = cmd.DoneInfo + if update.UpdateType == sstore.UpdateType_CmdExitCode { + rtn.IVal = int64(cmd.ExitCode) + } else if update.UpdateType == sstore.UpdateType_CmdDurationMs { + rtn.IVal = int64(cmd.DurationMs) + } case sstore.UpdateType_CmdRtnState: _, cmd, err := sstore.GetLineCmdByLineId(ctx, update.ScreenId, update.LineId) if err != nil || cmd == nil { return nil, fmt.Errorf("error getting cmd: %v", defaultError(err, "not found")) } - data, err := rtnstate.GetRtnStateDiff(ctx, update.ScreenId, cmd.CmdId) + data, err := rtnstate.GetRtnStateDiff(ctx, update.ScreenId, cmd.LineId) if err != nil { return nil, fmt.Errorf("cannot compute rtnstate: %v", err) } rtn.SVal = string(data) case sstore.UpdateType_PtyPos: - cmdId, err := sstore.GetCmdIdFromLineId(ctx, update.ScreenId, update.LineId) - if err != nil { - return nil, fmt.Errorf("error getting cmdid: %v", err) - } ptyPos, err := sstore.GetWebPtyPos(ctx, update.ScreenId, update.LineId) if err != nil { return nil, fmt.Errorf("error getting ptypos: %v", err) } - realOffset, data, err := sstore.ReadPtyOutFile(ctx, update.ScreenId, cmdId, ptyPos, MaxPtyUpdateSize+1) + realOffset, data, err := sstore.ReadPtyOutFile(ctx, update.ScreenId, update.LineId, ptyPos, MaxPtyUpdateSize+1) if err != nil { return nil, fmt.Errorf("error getting ptydata: %v", err) } diff --git a/pkg/pcloud/pclouddata.go b/pkg/pcloud/pclouddata.go index 5cca7626..87e43b4b 100644 --- a/pkg/pcloud/pclouddata.go +++ b/pkg/pcloud/pclouddata.go @@ -5,7 +5,6 @@ import ( "encoding/json" "fmt" - "github.com/commandlinedev/apishell/pkg/packet" "github.com/commandlinedev/prompt-server/pkg/remote" "github.com/commandlinedev/prompt-server/pkg/rtnstate" "github.com/commandlinedev/prompt-server/pkg/sstore" @@ -37,7 +36,6 @@ type WebShareUpdateType struct { SVal string `json:"sval,omitempty"` IVal int64 `json:"ival,omitempty"` BVal bool `json:"bval,omitempty"` - DoneInfo *sstore.CmdDoneInfo `json:"doneinfo,omitempty"` TermOpts *sstore.TermOpts `json:"termopts,omitempty"` } @@ -125,7 +123,6 @@ type WebShareLineType struct { ContentHeight int64 `json:"contentheight"` Renderer string `json:"renderer,omitempty"` Text string `json:"text,omitempty"` - CmdId string `json:"cmdid,omitempty"` } func webLineFromLine(line *sstore.LineType) (*WebShareLineType, error) { @@ -137,23 +134,25 @@ func webLineFromLine(line *sstore.LineType) (*WebShareLineType, error) { ContentHeight: line.ContentHeight, Renderer: line.Renderer, Text: line.Text, - CmdId: line.CmdId, } return rtn, nil } type WebShareCmdType struct { - LineId string `json:"lineid"` - CmdStr string `json:"cmdstr"` - RawCmdStr string `json:"rawcmdstr"` - Remote *WebShareRemote `json:"remote"` - FeState sstore.FeStateType `json:"festate"` - TermOpts sstore.TermOpts `json:"termopts"` - Status string `json:"status"` - StartPk *packet.CmdStartPacketType `json:"startpk,omitempty"` - DoneInfo *sstore.CmdDoneInfo `json:"doneinfo,omitempty"` - RtnState bool `json:"rtnstate,omitempty"` - RtnStateStr string `json:"rtnstatestr,omitempty"` + LineId string `json:"lineid"` + CmdStr string `json:"cmdstr"` + RawCmdStr string `json:"rawcmdstr"` + Remote *WebShareRemote `json:"remote"` + FeState sstore.FeStateType `json:"festate"` + TermOpts sstore.TermOpts `json:"termopts"` + Status string `json:"status"` + CmdPid int `json:"cmdpid"` + RemotePid int `json:"remotepid"` + DoneTs int64 `json:"donets,omitempty"` + ExitCode int `json:"exitcode,omitempty"` + DurationMs int `json:"durationms,omitempty"` + RtnState bool `json:"rtnstate,omitempty"` + RtnStateStr string `json:"rtnstatestr,omitempty"` } func webCmdFromCmd(lineId string, cmd *sstore.CmdType) (*WebShareCmdType, error) { @@ -166,21 +165,24 @@ func webCmdFromCmd(lineId string, cmd *sstore.CmdType) (*WebShareCmdType, error) } webRemote := webRemoteFromRemote(cmd.Remote, remote) rtn := &WebShareCmdType{ - LineId: lineId, - CmdStr: cmd.CmdStr, - RawCmdStr: cmd.RawCmdStr, - Remote: webRemote, - FeState: cmd.FeState, - TermOpts: cmd.TermOpts, - Status: cmd.Status, - StartPk: cmd.StartPk, - DoneInfo: cmd.DoneInfo, - RtnState: cmd.RtnState, + LineId: lineId, + CmdStr: cmd.CmdStr, + RawCmdStr: cmd.RawCmdStr, + Remote: webRemote, + FeState: cmd.FeState, + TermOpts: cmd.TermOpts, + Status: cmd.Status, + CmdPid: cmd.CmdPid, + RemotePid: cmd.RemotePid, + DoneTs: cmd.DoneTs, + ExitCode: cmd.ExitCode, + DurationMs: cmd.DurationMs, + RtnState: cmd.RtnState, } if cmd.RtnState { - barr, err := rtnstate.GetRtnStateDiff(context.Background(), cmd.ScreenId, cmd.CmdId) + barr, err := rtnstate.GetRtnStateDiff(context.Background(), cmd.ScreenId, cmd.LineId) if err != nil { - return nil, fmt.Errorf("error creating rtnstate diff for cmd:%s: %v", cmd.CmdId, err) + return nil, fmt.Errorf("error creating rtnstate diff for cmd:%s: %v", cmd.LineId, err) } rtn.RtnStateStr = string(barr) } diff --git a/pkg/remote/remote.go b/pkg/remote/remote.go index 823e93bd..326b8b91 100644 --- a/pkg/remote/remote.go +++ b/pkg/remote/remote.go @@ -1434,7 +1434,7 @@ func RunCommand(ctx context.Context, sessionId string, screenId string, remotePt } ok, existingPSC := msh.testAndSetPendingStateCmd(remotePtr.Name, newPSC) if !ok { - line, _, err := sstore.GetLineCmdByCmdId(ctx, screenId, existingPSC.GetCmdId()) + line, _, err := sstore.GetLineCmdByLineId(ctx, screenId, existingPSC.GetCmdId()) if err != nil { return nil, nil, fmt.Errorf("cannot run command while a stateful command is still running: %v", err) } @@ -1494,21 +1494,23 @@ func RunCommand(ctx context.Context, sessionId string, screenId string, remotePt status = sstore.CmdStatusDetached } cmd := &sstore.CmdType{ - ScreenId: runPacket.CK.GetGroupId(), - CmdId: runPacket.CK.GetCmdId(), - CmdStr: runPacket.Command, - RawCmdStr: runPacket.Command, - Remote: remotePtr, - FeState: sstore.FeStateFromShellState(currentState), - StatePtr: *statePtr, - TermOpts: makeTermOpts(runPacket), - Status: status, - StartPk: startPk, - DoneInfo: nil, - RunOut: nil, - RtnState: runPacket.ReturnState, + ScreenId: runPacket.CK.GetGroupId(), + LineId: runPacket.CK.GetCmdId(), + CmdStr: runPacket.Command, + RawCmdStr: runPacket.Command, + Remote: remotePtr, + FeState: sstore.FeStateFromShellState(currentState), + StatePtr: *statePtr, + TermOpts: makeTermOpts(runPacket), + Status: status, + CmdPid: startPk.Pid, + RemotePid: startPk.MShellPid, + ExitCode: 0, + DurationMs: 0, + RunOut: nil, + RtnState: runPacket.ReturnState, } - err = sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.CmdId, cmd.TermOpts.MaxPtySize) + err = sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.LineId, cmd.TermOpts.MaxPtySize) if err != nil { // TODO the cmd is running, so this is a tricky error to handle return nil, nil, fmt.Errorf("cannot create local ptyout file for running command: %v", err) @@ -1645,12 +1647,7 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) { if donePk.FinalStateDiff != nil { donePk.FinalStateDiff = stripScVarsFromStateDiff(donePk.FinalStateDiff) } - doneInfo := &sstore.CmdDoneInfo{ - Ts: donePk.Ts, - ExitCode: int64(donePk.ExitCode), - DurationMs: donePk.DurationMs, - } - update, err := sstore.UpdateCmdDoneInfo(context.Background(), donePk.CK, doneInfo, sstore.CmdStatusDone) + update, err := sstore.UpdateCmdDoneInfo(context.Background(), donePk.CK, donePk, sstore.CmdStatusDone) if err != nil { msh.WriteToPtyBuffer("*error updating cmddone: %v\n", err) return @@ -1713,7 +1710,7 @@ func (msh *MShellProc) handleCmdFinalPacket(finalPk *packet.CmdFinalPacketType) log.Printf("error calling GetCmdById in handleCmdFinalPacket: %v\n", err) return } - if rtnCmd == nil || rtnCmd.DoneInfo != nil { + if rtnCmd == nil || rtnCmd.DoneTs > 0 { return } log.Printf("finalpk %s (hangup): %s\n", finalPk.CK, finalPk.Error) diff --git a/pkg/rtnstate/rtnstate.go b/pkg/rtnstate/rtnstate.go index a220a7e5..2a27f0c0 100644 --- a/pkg/rtnstate/rtnstate.go +++ b/pkg/rtnstate/rtnstate.go @@ -168,8 +168,8 @@ func displayStateUpdateDiff(buf *bytes.Buffer, oldState packet.ShellState, newSt } } -func GetRtnStateDiff(ctx context.Context, screenId string, cmdId string) ([]byte, error) { - cmd, err := sstore.GetCmdByScreenId(ctx, screenId, cmdId) +func GetRtnStateDiff(ctx context.Context, screenId string, lineId string) ([]byte, error) { + cmd, err := sstore.GetCmdByScreenId(ctx, screenId, lineId) if err != nil { return nil, err } diff --git a/pkg/scbase/scbase.go b/pkg/scbase/scbase.go index 584abb1b..b234ca48 100644 --- a/pkg/scbase/scbase.go +++ b/pkg/scbase/scbase.go @@ -17,8 +17,8 @@ import ( "sync" "time" - "github.com/google/uuid" "github.com/commandlinedev/apishell/pkg/base" + "github.com/google/uuid" "golang.org/x/mod/semver" "golang.org/x/sys/unix" ) @@ -258,7 +258,7 @@ func PtyOutFile_Sessions(sessionId string, cmdId string) (string, error) { return fmt.Sprintf("%s/%s.ptyout.cf", sdir, cmdId), nil } -func PtyOutFile(screenId string, cmdId string) (string, error) { +func PtyOutFile(screenId string, lineId string) (string, error) { sdir, err := EnsureScreenDir(screenId) if err != nil { return "", err @@ -266,10 +266,10 @@ func PtyOutFile(screenId string, cmdId string) (string, error) { if screenId == "" { return "", fmt.Errorf("cannot get ptyout file for blank screenid") } - if cmdId == "" { - return "", fmt.Errorf("cannot get ptyout file for blank cmdid") + if lineId == "" { + return "", fmt.Errorf("cannot get ptyout file for blank lineid") } - return fmt.Sprintf("%s/%s.ptyout.cf", sdir, cmdId), nil + return fmt.Sprintf("%s/%s.ptyout.cf", sdir, lineId), nil } func GenPromptUUID() string { diff --git a/pkg/sstore/dbops.go b/pkg/sstore/dbops.go index 559fddb2..c9972ecb 100644 --- a/pkg/sstore/dbops.go +++ b/pkg/sstore/dbops.go @@ -10,17 +10,17 @@ import ( "sync" "time" - "github.com/google/uuid" - "github.com/jmoiron/sqlx" - "github.com/sawka/txwrap" "github.com/commandlinedev/apishell/pkg/base" "github.com/commandlinedev/apishell/pkg/packet" "github.com/commandlinedev/apishell/pkg/shexec" "github.com/commandlinedev/prompt-server/pkg/dbutil" "github.com/commandlinedev/prompt-server/pkg/scbase" + "github.com/google/uuid" + "github.com/jmoiron/sqlx" + "github.com/sawka/txwrap" ) -const HistoryCols = "h.historyid, h.ts, h.userid, h.sessionid, h.screenid, h.lineid, h.cmdid, h.haderror, h.cmdstr, h.remoteownerid, h.remoteid, h.remotename, h.ismetacmd, h.incognito, h.linenum" +const HistoryCols = "h.historyid, h.ts, h.userid, h.sessionid, h.screenid, h.lineid, h.haderror, h.cmdstr, h.remoteownerid, h.remoteid, h.remotename, h.ismetacmd, h.incognito, h.linenum" const DefaultMaxHistoryItems = 1000 var updateWriterCVar = sync.NewCond(&sync.Mutex{}) @@ -217,8 +217,8 @@ func InsertHistoryItem(ctx context.Context, hitem *HistoryItemType) error { } txErr := WithTx(ctx, func(tx *TxWrap) error { query := `INSERT INTO history - ( historyid, ts, userid, sessionid, screenid, lineid, cmdid, haderror, cmdstr, remoteownerid, remoteid, remotename, ismetacmd, incognito, linenum) VALUES - (:historyid,:ts,:userid,:sessionid,:screenid,:lineid,:cmdid,:haderror,:cmdstr,:remoteownerid,:remoteid,:remotename,:ismetacmd,:incognito,:linenum)` + ( historyid, ts, userid, sessionid, screenid, lineid, haderror, cmdstr, remoteownerid, remoteid, remotename, ismetacmd, incognito, linenum) VALUES + (:historyid,:ts,:userid,:sessionid,:screenid,:lineid,:haderror,:cmdstr,:remoteownerid,:remoteid,:remotename,:ismetacmd,:incognito,:linenum)` tx.NamedExec(query, hitem.ToMap()) return nil }) @@ -472,7 +472,7 @@ func GetScreenLinesById(ctx context.Context, screenId string) (*ScreenLinesType, } query = `SELECT * FROM line WHERE screenid = ? ORDER BY linenum` tx.Select(&screen.Lines, query, screen.ScreenId) - query = `SELECT * FROM cmd WHERE cmdid IN (SELECT cmdid FROM line WHERE screenid = ?)` + query = `SELECT * FROM cmd WHERE screenid = ?` screen.Cmds = dbutil.SelectMapsGen[*CmdType](tx, query, screen.ScreenId) return screen, nil }) @@ -759,14 +759,6 @@ func FindLineIdByArg(ctx context.Context, screenId string, lineArg string) (stri return lineId, nil } -func GetCmdIdFromLineId(ctx context.Context, screenId string, lineId string) (string, error) { - return WithTxRtn(ctx, func(tx *TxWrap) (string, error) { - query := `SELECT cmdid FROM line WHERE screenid = ? AND lineid = ?` - cmdId := tx.GetString(query, screenId, lineId) - return cmdId, nil - }) -} - func GetLineCmdByLineId(ctx context.Context, screenId string, lineId string) (*LineType, *CmdType, error) { return WithTxRtn3(ctx, func(tx *TxWrap) (*LineType, *CmdType, error) { var lineVal LineType @@ -776,24 +768,8 @@ func GetLineCmdByLineId(ctx context.Context, screenId string, lineId string) (*L return nil, nil, nil } var cmdRtn *CmdType - if lineVal.CmdId != "" { - query = `SELECT * FROM cmd WHERE screenid = ? AND cmdid = ?` - cmdRtn = dbutil.GetMapGen[*CmdType](tx, query, screenId, lineVal.CmdId) - } - return &lineVal, cmdRtn, nil - }) -} - -func GetLineCmdByCmdId(ctx context.Context, screenId string, cmdId string) (*LineType, *CmdType, error) { - return WithTxRtn3(ctx, func(tx *TxWrap) (*LineType, *CmdType, error) { - var lineVal LineType - query := `SELECT * FROM line WHERE screenid = ? AND cmdid = ?` - found := tx.Get(&lineVal, query, screenId, cmdId) - if !found { - return nil, nil, nil - } - query = `SELECT * FROM cmd WHERE screenid = ? AND cmdid = ?` - cmdRtn := dbutil.GetMapGen[*CmdType](tx, query, screenId, cmdId) + query = `SELECT * FROM cmd WHERE screenid = ? AND lineid = ?` + cmdRtn = dbutil.GetMapGen[*CmdType](tx, query, screenId, lineId) return &lineVal, cmdRtn, nil }) } @@ -819,8 +795,8 @@ func InsertLine(ctx context.Context, line *LineType, cmd *CmdType) error { query = `SELECT nextlinenum FROM screen WHERE screenid = ?` nextLineNum := tx.GetInt(query, line.ScreenId) line.LineNum = int64(nextLineNum) - query = `INSERT INTO line ( screenid, userid, lineid, ts, linenum, linenumtemp, linelocal, linetype, text, cmdid, renderer, ephemeral, contentheight, star, archived) - VALUES (:screenid,:userid,:lineid,:ts,:linenum,:linenumtemp,:linelocal,:linetype,:text,:cmdid,:renderer,:ephemeral,:contentheight,:star,:archived)` + query = `INSERT INTO line ( screenid, userid, lineid, ts, linenum, linenumtemp, linelocal, linetype, text, renderer, ephemeral, contentheight, star, archived) + VALUES (:screenid,:userid,:lineid,:ts,:linenum,:linenumtemp,:linelocal,:linetype,:text,:renderer,:ephemeral,:contentheight,:star,:archived)` tx.NamedExec(query, line) query = `UPDATE screen SET nextlinenum = ? WHERE screenid = ?` tx.Exec(query, nextLineNum+1, line.ScreenId) @@ -828,8 +804,8 @@ func InsertLine(ctx context.Context, line *LineType, cmd *CmdType) error { cmd.OrigTermOpts = cmd.TermOpts cmdMap := cmd.ToMap() query = ` -INSERT INTO cmd ( screenid, cmdid, remoteownerid, remoteid, remotename, cmdstr, rawcmdstr, festate, statebasehash, statediffhasharr, termopts, origtermopts, status, startpk, doneinfo, rtnstate, runout, rtnbasehash, rtndiffhasharr) - VALUES (:screenid,:cmdid,:remoteownerid,:remoteid,:remotename,:cmdstr,:rawcmdstr,:festate,:statebasehash,:statediffhasharr,:termopts,:origtermopts,:status,:startpk,:doneinfo,:rtnstate,:runout,:rtnbasehash,:rtndiffhasharr) +INSERT INTO cmd ( screenid, lineid, remoteownerid, remoteid, remotename, cmdstr, rawcmdstr, festate, statebasehash, statediffhasharr, termopts, origtermopts, status, startpk, doneinfo, rtnstate, runout, rtnbasehash, rtndiffhasharr) + VALUES (:screenid,:lineid,:remoteownerid,:remoteid,:remotename,:cmdstr,:rawcmdstr,:festate,:statebasehash,:statediffhasharr,:termopts,:origtermopts,:status,:startpk,:doneinfo,:rtnstate,:runout,:rtnbasehash,:rtndiffhasharr) ` tx.NamedExec(query, cmdMap) } @@ -840,11 +816,11 @@ INSERT INTO cmd ( screenid, cmdid, remoteownerid, remoteid, remotename, cmdstr, }) } -func GetCmdByScreenId(ctx context.Context, screenId string, cmdId string) (*CmdType, error) { +func GetCmdByScreenId(ctx context.Context, screenId string, lineId string) (*CmdType, error) { var cmd *CmdType err := WithTx(ctx, func(tx *TxWrap) error { - query := `SELECT * FROM cmd WHERE screenid = ? AND cmdid = ?` - cmd = dbutil.GetMapGen[*CmdType](tx, query, screenId, cmdId) + query := `SELECT * FROM cmd WHERE screenid = ? AND lineid = ?` + cmd = dbutil.GetMapGen[*CmdType](tx, query, screenId, lineId) return nil }) if err != nil { @@ -853,8 +829,8 @@ func GetCmdByScreenId(ctx context.Context, screenId string, cmdId string) (*CmdT return cmd, nil } -func UpdateCmdDoneInfo(ctx context.Context, ck base.CommandKey, doneInfo *CmdDoneInfo, status string) (*ModelUpdate, error) { - if doneInfo == nil { +func UpdateCmdDoneInfo(ctx context.Context, ck base.CommandKey, donePk *packet.CmdDonePacketType, status string) (*ModelUpdate, error) { + if donePk == nil { return nil, fmt.Errorf("invalid cmddone packet") } if ck.IsEmpty() { @@ -863,16 +839,17 @@ func UpdateCmdDoneInfo(ctx context.Context, ck base.CommandKey, doneInfo *CmdDon screenId := ck.GetGroupId() var rtnCmd *CmdType txErr := WithTx(ctx, func(tx *TxWrap) error { - query := `UPDATE cmd SET status = ?, doneinfo = ? WHERE screenid = ? AND cmdid = ?` - tx.Exec(query, status, quickJson(doneInfo), screenId, ck.GetCmdId()) + query := `UPDATE cmd SET status = ?, donets = ?, exitcode = ?, durationms = ? WHERE screenid = ? AND lineid = ?` + tx.Exec(query, status, donePk.Ts, donePk.ExitCode, donePk.DurationMs, screenId, lineIdFromCK(ck)) var err error - rtnCmd, err = GetCmdByScreenId(tx.Context(), screenId, ck.GetCmdId()) + rtnCmd, err = GetCmdByScreenId(tx.Context(), screenId, lineIdFromCK(ck)) if err != nil { return err } if isWebShare(tx, screenId) { - insertScreenCmdUpdate(tx, screenId, ck.GetCmdId(), UpdateType_CmdDoneInfo) - insertScreenCmdUpdate(tx, screenId, ck.GetCmdId(), UpdateType_CmdStatus) + insertScreenLineUpdate(tx, screenId, lineIdFromCK(ck), UpdateType_CmdExitCode) + insertScreenLineUpdate(tx, screenId, lineIdFromCK(ck), UpdateType_CmdDurationMs) + insertScreenLineUpdate(tx, screenId, lineIdFromCK(ck), UpdateType_CmdStatus) } return nil }) @@ -890,11 +867,12 @@ func UpdateCmdRtnState(ctx context.Context, ck base.CommandKey, statePtr ShellSt return fmt.Errorf("cannot update cmdrtnstate, empty ck") } screenId := ck.GetGroupId() + lineId := lineIdFromCK(ck) txErr := WithTx(ctx, func(tx *TxWrap) error { - query := `UPDATE cmd SET rtnbasehash = ?, rtndiffhasharr = ? WHERE screenid = ? AND cmdid = ?` - tx.Exec(query, statePtr.BaseHash, quickJsonArr(statePtr.DiffHashArr), screenId, ck.GetCmdId()) + query := `UPDATE cmd SET rtnbasehash = ?, rtndiffhasharr = ? WHERE screenid = ? AND lineid = ?` + tx.Exec(query, statePtr.BaseHash, quickJsonArr(statePtr.DiffHashArr), screenId, lineId) if isWebShare(tx, screenId) { - insertScreenCmdUpdate(tx, screenId, ck.GetCmdId(), UpdateType_CmdRtnState) + insertScreenLineUpdate(tx, screenId, lineId, UpdateType_CmdRtnState) } return nil }) @@ -910,8 +888,8 @@ func AppendCmdErrorPk(ctx context.Context, errPk *packet.CmdErrorPacketType) err } screenId := errPk.CK.GetGroupId() return WithTx(ctx, func(tx *TxWrap) error { - query := `UPDATE cmd SET runout = json_insert(runout, '$[#]', ?) WHERE screenid = ? AND cmdid = ?` - tx.Exec(query, quickJson(errPk), screenId, errPk.CK.GetCmdId()) + query := `UPDATE cmd SET runout = json_insert(runout, '$[#]', ?) WHERE screenid = ? AND lineid = ?` + tx.Exec(query, quickJson(errPk), screenId, lineIdFromCK(errPk.CK)) return nil }) } @@ -927,13 +905,13 @@ func ReInitFocus(ctx context.Context) error { func HangupAllRunningCmds(ctx context.Context) error { return WithTx(ctx, func(tx *TxWrap) error { var cmdPtrs []CmdPtr - query := `SELECT screenid, cmdid FROM cmd WHERE status = ?` + query := `SELECT screenid, lineid FROM cmd WHERE status = ?` tx.Select(&cmdPtrs, query, CmdStatusRunning) query = `UPDATE cmd SET status = ? WHERE status = ?` tx.Exec(query, CmdStatusHangup, CmdStatusRunning) for _, cmdPtr := range cmdPtrs { if isWebShare(tx, cmdPtr.ScreenId) { - insertScreenCmdUpdate(tx, cmdPtr.ScreenId, cmdPtr.CmdId, UpdateType_CmdStatus) + insertScreenLineUpdate(tx, cmdPtr.ScreenId, cmdPtr.LineId, UpdateType_CmdStatus) } } return nil @@ -944,16 +922,16 @@ func HangupAllRunningCmds(ctx context.Context) error { func HangupRunningCmdsByRemoteId(ctx context.Context, remoteId string) ([]*ScreenType, error) { return WithTxRtn(ctx, func(tx *TxWrap) ([]*ScreenType, error) { var cmdPtrs []CmdPtr - query := `SELECT screenid, cmdid FROM cmd WHERE status = ? AND remoteid = ?` + query := `SELECT screenid, lineid FROM cmd WHERE status = ? AND remoteid = ?` tx.Select(&cmdPtrs, query, CmdStatusRunning, remoteId) query = `UPDATE cmd SET status = ? WHERE status = ? AND remoteid = ?` tx.Exec(query, CmdStatusHangup, CmdStatusRunning, remoteId) var rtn []*ScreenType for _, cmdPtr := range cmdPtrs { if isWebShare(tx, cmdPtr.ScreenId) { - insertScreenCmdUpdate(tx, cmdPtr.ScreenId, cmdPtr.CmdId, UpdateType_CmdStatus) + insertScreenLineUpdate(tx, cmdPtr.ScreenId, cmdPtr.LineId, UpdateType_CmdStatus) } - screen, err := UpdateScreenFocusForDoneCmd(tx.Context(), cmdPtr.ScreenId, cmdPtr.CmdId) + screen, err := UpdateScreenFocusForDoneCmd(tx.Context(), cmdPtr.ScreenId, cmdPtr.LineId) if err != nil { return nil, err } @@ -968,12 +946,12 @@ func HangupRunningCmdsByRemoteId(ctx context.Context, remoteId string) ([]*Scree // TODO send update func HangupCmd(ctx context.Context, ck base.CommandKey) (*ScreenType, error) { return WithTxRtn(ctx, func(tx *TxWrap) (*ScreenType, error) { - query := `UPDATE cmd SET status = ? WHERE screenid = ? AND cmdid = ?` - tx.Exec(query, CmdStatusHangup, ck.GetGroupId(), ck.GetCmdId()) + query := `UPDATE cmd SET status = ? WHERE screenid = ? AND lineid = ?` + tx.Exec(query, CmdStatusHangup, ck.GetGroupId(), lineIdFromCK(ck)) if isWebShare(tx, ck.GetGroupId()) { - insertScreenCmdUpdate(tx, ck.GetGroupId(), ck.GetCmdId(), UpdateType_CmdStatus) + insertScreenLineUpdate(tx, ck.GetGroupId(), lineIdFromCK(ck), UpdateType_CmdStatus) } - screen, err := UpdateScreenFocusForDoneCmd(tx.Context(), ck.GetGroupId(), ck.GetCmdId()) + screen, err := UpdateScreenFocusForDoneCmd(tx.Context(), ck.GetGroupId(), lineIdFromCK(ck)) if err != nil { return nil, err } @@ -1029,17 +1007,17 @@ func SwitchScreenById(ctx context.Context, sessionId string, screenId string) (* func cleanScreenCmds(ctx context.Context, screenId string) error { var removedCmds []string txErr := WithTx(ctx, func(tx *TxWrap) error { - query := `SELECT cmdid FROM cmd WHERE screenid = ? AND cmdid NOT IN (SELECT cmdid FROM line WHERE screenid = ?)` + query := `SELECT lineid FROM cmd WHERE screenid = ? AND lineid NOT IN (SELECT lineid FROM line WHERE screenid = ?)` removedCmds = tx.SelectStrings(query, screenId, screenId) - query = `DELETE FROM cmd WHERE screenid = ? AND cmdid NOT IN (SELECT cmdid FROM line WHERE screenid = ?)` + query = `DELETE FROM cmd WHERE screenid = ? AND lineid NOT IN (SELECT lineid FROM line WHERE screenid = ?)` tx.Exec(query, screenId, screenId) return nil }) if txErr != nil { return txErr } - for _, cmdId := range removedCmds { - DeletePtyOutFile(ctx, screenId, cmdId) + for _, lineId := range removedCmds { + DeletePtyOutFile(ctx, screenId, lineId) } return nil } @@ -1461,7 +1439,7 @@ func PurgeScreenLines(ctx context.Context, screenId string) (*ModelUpdate, error func GetRunningScreenCmds(ctx context.Context, screenId string) ([]*CmdType, error) { var rtn []*CmdType txErr := WithTx(ctx, func(tx *TxWrap) error { - query := `SELECT * from cmd WHERE cmdid IN (SELECT cmdid FROM line WHERE screenid = ?) AND status = ?` + query := `SELECT * FROM cmd WHERE screenid = ? AND status = ?` rtn = dbutil.SelectMapsGen[*CmdType](tx, query, screenId, CmdStatusRunning) return nil }) @@ -1471,11 +1449,11 @@ func GetRunningScreenCmds(ctx context.Context, screenId string) ([]*CmdType, err return rtn, nil } -func UpdateCmdTermOpts(ctx context.Context, screenId string, cmdId string, termOpts TermOpts) error { +func UpdateCmdTermOpts(ctx context.Context, screenId string, lineId string, termOpts TermOpts) error { txErr := WithTx(ctx, func(tx *TxWrap) error { - query := `UPDATE cmd SET termopts = ? WHERE screenid = ? AND cmdid = ?` - tx.Exec(query, termOpts, screenId, cmdId) - insertScreenCmdUpdate(tx, screenId, cmdId, UpdateType_CmdTermOpts) + query := `UPDATE cmd SET termopts = ? WHERE screenid = ? AND lineid = ?` + tx.Exec(query, termOpts, screenId, lineId) + insertScreenLineUpdate(tx, screenId, lineId, UpdateType_CmdTermOpts) return nil }) return txErr @@ -1792,14 +1770,14 @@ func GetLineResolveItems(ctx context.Context, screenId string) ([]ResolveItem, e return rtn, nil } -func UpdateScreenFocusForDoneCmd(ctx context.Context, screenId string, cmdId string) (*ScreenType, error) { +func UpdateScreenFocusForDoneCmd(ctx context.Context, screenId string, lineId string) (*ScreenType, error) { return WithTxRtn(ctx, func(tx *TxWrap) (*ScreenType, error) { query := `SELECT screenid FROM screen s WHERE s.screenid = ? AND s.focustype = ? - AND s.selectedline IN (SELECT linenum FROM line l WHERE l.screenid = s.screenid AND l.cmdid = ?) + AND s.selectedline IN (SELECT linenum FROM line l WHERE l.screenid = s.screenid AND l.lineid = ?) ` - if !tx.Exists(query, screenId, ScreenFocusCmd, cmdId) { + if !tx.Exists(query, screenId, ScreenFocusCmd, lineId) { return nil, nil } editMap := make(map[string]interface{}) @@ -1983,11 +1961,15 @@ func SetLineArchivedById(ctx context.Context, screenId string, lineId string, ar return txErr } -func purgeCmdByScreenId(ctx context.Context, screenId string, cmdId string) error { +func purgeCmdByScreenId(ctx context.Context, screenId string, lineId string) error { txErr := WithTx(ctx, func(tx *TxWrap) error { - query := `DELETE FROM cmd WHERE screenid = ? AND cmdid = ?` - tx.Exec(query, screenId, cmdId) - return DeletePtyOutFile(tx.Context(), screenId, cmdId) + query := `DELETE FROM cmd WHERE screenid = ? AND lineid = ?` + tx.Exec(query, screenId, lineId) + if tx.Err != nil { + // short circuit here because we don't want to delete the ptyfile when the tx will be rolled back + return tx.Err + } + return DeletePtyOutFile(tx.Context(), screenId, lineId) }) return txErr } @@ -1996,17 +1978,13 @@ func PurgeLinesByIds(ctx context.Context, screenId string, lineIds []string) err txErr := WithTx(ctx, func(tx *TxWrap) error { isWS := isWebShare(tx, screenId) for _, lineId := range lineIds { - query := `SELECT cmdid FROM line WHERE screenid = ? AND lineid = ?` - cmdId := tx.GetString(query, screenId, lineId) - query = `DELETE FROM line WHERE screenid = ? AND lineid = ?` + query := `DELETE FROM line WHERE screenid = ? AND lineid = ?` tx.Exec(query, screenId, lineId) query = `DELETE FROM history WHERE screenid = ? AND lineid = ?` tx.Exec(query, screenId, lineId) - if cmdId != "" { - err := purgeCmdByScreenId(tx.Context(), screenId, cmdId) - if err != nil { - return err - } + err := purgeCmdByScreenId(tx.Context(), screenId, lineId) + if err != nil { + return err } if isWS { insertScreenLineUpdate(tx, screenId, lineId, UpdateType_LineDel) @@ -2412,26 +2390,17 @@ func getLineIdsFromHistoryItems(historyItems []*HistoryItemType) []string { return rtn } -func getCmdIdsFromHistoryItems(historyItems []*HistoryItemType) []string { - var rtn []string - for _, hitem := range historyItems { - if hitem.CmdId != "" { - rtn = append(rtn, hitem.CmdId) - } - } - return rtn -} - func GetLineCmdsFromHistoryItems(ctx context.Context, historyItems []*HistoryItemType) ([]*LineType, []*CmdType, error) { if len(historyItems) == 0 { return nil, nil, nil } return WithTxRtn3(ctx, func(tx *TxWrap) ([]*LineType, []*CmdType, error) { var lineArr []*LineType + lineIdsJsonArr := quickJsonArr(getLineIdsFromHistoryItems(historyItems)) query := `SELECT * FROM line WHERE lineid IN (SELECT value FROM json_each(?))` - tx.Select(&lineArr, query, quickJsonArr(getLineIdsFromHistoryItems(historyItems))) - query = `SELECT * FROM cmd WHERE cmdid IN (SELECT value FROM json_each(?))` - cmdArr := dbutil.SelectMapsGen[*CmdType](tx, query, quickJsonArr(getCmdIdsFromHistoryItems(historyItems))) + tx.Select(&lineArr, query, lineIdsJsonArr) + query = `SELECT * FROM cmd WHERE lineid IN (SELECT value FROM json_each(?))` + cmdArr := dbutil.SelectMapsGen[*CmdType](tx, query, lineIdsJsonArr) return lineArr, cmdArr, nil }) } @@ -2560,7 +2529,7 @@ func insertScreenNewUpdate(tx *TxWrap, screenId string) { SELECT screenid, lineid, ?, ? FROM line WHERE screenid = ? AND NOT archived ORDER BY linenum DESC` tx.Exec(query, UpdateType_LineNew, nowTs, screenId) query = `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets) - SELECT c.screenid, l.lineid, ?, ? FROM cmd c, line l WHERE c.screenid = ? AND l.cmdid = c.cmdid AND NOT l.archived ORDER BY l.linenum DESC` + SELECT c.screenid, c.lineid, ?, ? FROM cmd c, line l WHERE c.screenid = ? AND l.lineid = c.lineid AND NOT l.archived ORDER BY l.linenum DESC` tx.Exec(query, UpdateType_PtyPos, nowTs, screenId) NotifyUpdateWriter() } @@ -2600,22 +2569,6 @@ func insertScreenLineUpdate(tx *TxWrap, screenId string, lineId string, updateTy NotifyUpdateWriter() } -func insertScreenCmdUpdate(tx *TxWrap, screenId string, cmdId string, updateType string) { - if screenId == "" { - tx.SetErr(errors.New("invalid screen-update, screenid is empty")) - return - } - if cmdId == "" { - tx.SetErr(errors.New("invalid screen-update, cmdid is empty")) - return - } - query := `SELECT lineid FROM line WHERE screenid = ? AND cmdid = ?` - lineId := tx.GetString(query, screenId, cmdId) - if lineId != "" { - insertScreenLineUpdate(tx, screenId, lineId, updateType) - } -} - func GetScreenUpdates(ctx context.Context, maxNum int) ([]*ScreenUpdateType, error) { return WithTxRtn(ctx, func(tx *TxWrap) ([]*ScreenUpdateType, error) { var updates []*ScreenUpdateType @@ -2651,12 +2604,12 @@ func RemoveScreenUpdates(ctx context.Context, updateIds []int64) error { }) } -func MaybeInsertPtyPosUpdate(ctx context.Context, screenId string, cmdId string) error { +func MaybeInsertPtyPosUpdate(ctx context.Context, screenId string, lineId string) error { return WithTx(ctx, func(tx *TxWrap) error { if !isWebShare(tx, screenId) { return nil } - insertScreenCmdUpdate(tx, screenId, cmdId, UpdateType_PtyPos) + insertScreenLineUpdate(tx, screenId, lineId, UpdateType_PtyPos) return nil }) } diff --git a/pkg/sstore/fileops.go b/pkg/sstore/fileops.go index 61214836..953b5673 100644 --- a/pkg/sstore/fileops.go +++ b/pkg/sstore/fileops.go @@ -3,18 +3,20 @@ package sstore import ( "context" "encoding/base64" + "errors" "fmt" + "io/fs" "log" "os" "path" - "github.com/google/uuid" "github.com/commandlinedev/apishell/pkg/cirfile" "github.com/commandlinedev/prompt-server/pkg/scbase" + "github.com/google/uuid" ) -func CreateCmdPtyFile(ctx context.Context, screenId string, cmdId string, maxSize int64) error { - ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId) +func CreateCmdPtyFile(ctx context.Context, screenId string, lineId string, maxSize int64) error { + ptyOutFileName, err := scbase.PtyOutFile(screenId, lineId) if err != nil { return err } @@ -25,22 +27,22 @@ func CreateCmdPtyFile(ctx context.Context, screenId string, cmdId string, maxSiz return f.Close() } -func StatCmdPtyFile(ctx context.Context, screenId string, cmdId string) (*cirfile.Stat, error) { - ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId) +func StatCmdPtyFile(ctx context.Context, screenId string, lineId string) (*cirfile.Stat, error) { + ptyOutFileName, err := scbase.PtyOutFile(screenId, lineId) if err != nil { return nil, err } return cirfile.StatCirFile(ctx, ptyOutFileName) } -func AppendToCmdPtyBlob(ctx context.Context, screenId string, cmdId string, data []byte, pos int64) (*PtyDataUpdate, error) { +func AppendToCmdPtyBlob(ctx context.Context, screenId string, lineId string, data []byte, pos int64) (*PtyDataUpdate, error) { if screenId == "" { return nil, fmt.Errorf("cannot append to PtyBlob, screenid is not set") } if pos < 0 { return nil, fmt.Errorf("invalid seek pos '%d' in AppendToCmdPtyBlob", pos) } - ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId) + ptyOutFileName, err := scbase.PtyOutFile(screenId, lineId) if err != nil { return nil, err } @@ -56,22 +58,22 @@ func AppendToCmdPtyBlob(ctx context.Context, screenId string, cmdId string, data data64 := base64.StdEncoding.EncodeToString(data) update := &PtyDataUpdate{ ScreenId: screenId, - CmdId: cmdId, + LineId: lineId, PtyPos: pos, PtyData64: data64, PtyDataLen: int64(len(data)), } - err = MaybeInsertPtyPosUpdate(ctx, screenId, cmdId) + err = MaybeInsertPtyPosUpdate(ctx, screenId, lineId) if err != nil { // just log - log.Printf("error inserting ptypos update %s/%s: %v\n", screenId, cmdId, err) + log.Printf("error inserting ptypos update %s/%s: %v\n", screenId, lineId, err) } return update, nil } // returns (real-offset, data, err) -func ReadFullPtyOutFile(ctx context.Context, screenId string, cmdId string) (int64, []byte, error) { - ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId) +func ReadFullPtyOutFile(ctx context.Context, screenId string, lineId string) (int64, []byte, error) { + ptyOutFileName, err := scbase.PtyOutFile(screenId, lineId) if err != nil { return 0, nil, err } @@ -84,8 +86,8 @@ func ReadFullPtyOutFile(ctx context.Context, screenId string, cmdId string) (int } // returns (real-offset, data, err) -func ReadPtyOutFile(ctx context.Context, screenId string, cmdId string, offset int64, maxSize int64) (int64, []byte, error) { - ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId) +func ReadPtyOutFile(ctx context.Context, screenId string, lineId string, offset int64, maxSize int64) (int64, []byte, error) { + ptyOutFileName, err := scbase.PtyOutFile(screenId, lineId) if err != nil { return 0, nil, err } @@ -160,12 +162,16 @@ func FullSessionDiskSize() (map[string]SessionDiskSizeType, error) { return rtn, nil } -func DeletePtyOutFile(ctx context.Context, screenId string, cmdId string) error { - ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId) +func DeletePtyOutFile(ctx context.Context, screenId string, lineId string) error { + ptyOutFileName, err := scbase.PtyOutFile(screenId, lineId) if err != nil { return err } - return os.Remove(ptyOutFileName) + err = os.Remove(ptyOutFileName) + if errors.Is(err, fs.ErrNotExist) { + return nil + } + return err } func DeleteScreenDir(ctx context.Context, screenId string) error { diff --git a/pkg/sstore/migrate.go b/pkg/sstore/migrate.go index 72b6bc71..8578f945 100644 --- a/pkg/sstore/migrate.go +++ b/pkg/sstore/migrate.go @@ -17,9 +17,10 @@ import ( "github.com/golang-migrate/migrate/v4" ) -const MaxMigration = 19 +const MaxMigration = 20 const MigratePrimaryScreenVersion = 9 const CmdScreenSpecialMigration = 13 +const CmdLineSpecialMigration = 20 func MakeMigrate() (*migrate.Migrate, error) { fsVar, err := iofs.New(sh2db.MigrationFS, "migrations") @@ -64,9 +65,15 @@ func MigrateUpStep(m *migrate.Migrate, newVersion uint) error { return err } if newVersion == CmdScreenSpecialMigration { - mErr := RunCmdScreenMigration13() + mErr := RunMigration13() if mErr != nil { - return mErr + return fmt.Errorf("migrating to v%d: %w", newVersion, mErr) + } + } + if newVersion == CmdLineSpecialMigration { + mErr := RunMigration20() + if mErr != nil { + return fmt.Errorf("migrating to v%d: %w", newVersion, mErr) } } log.Printf("[db] migration v%d, elapsed %v\n", newVersion, time.Since(startTime)) diff --git a/pkg/sstore/quick.go b/pkg/sstore/quick.go index a887d23d..c19ce8a0 100644 --- a/pkg/sstore/quick.go +++ b/pkg/sstore/quick.go @@ -6,6 +6,7 @@ import ( var quickSetStr = dbutil.QuickSetStr var quickSetInt64 = dbutil.QuickSetInt64 +var quickSetInt = dbutil.QuickSetInt var quickSetBool = dbutil.QuickSetBool var quickSetBytes = dbutil.QuickSetBytes var quickSetJson = dbutil.QuickSetJson diff --git a/pkg/sstore/sstore.go b/pkg/sstore/sstore.go index c10a210a..47a4a164 100644 --- a/pkg/sstore/sstore.go +++ b/pkg/sstore/sstore.go @@ -17,6 +17,7 @@ import ( "sync" "time" + "github.com/commandlinedev/apishell/pkg/base" "github.com/commandlinedev/apishell/pkg/packet" "github.com/commandlinedev/apishell/pkg/shexec" "github.com/commandlinedev/prompt-server/pkg/dbutil" @@ -58,7 +59,6 @@ const ( CmdStatusError = "error" CmdStatusDone = "done" CmdStatusHangup = "hangup" - CmdStatusWaiting = "waiting" ) const ( @@ -115,7 +115,8 @@ const ( UpdateType_LineContentHeight = "line:contentheight" UpdateType_CmdStatus = "cmd:status" UpdateType_CmdTermOpts = "cmd:termopts" - UpdateType_CmdDoneInfo = "cmd:doneinfo" + UpdateType_CmdExitCode = "cmd:exitcode" + UpdateType_CmdDurationMs = "cmd:durationms" UpdateType_CmdRtnState = "cmd:rtnstate" UpdateType_PtyPos = "pty:pos" ) @@ -126,6 +127,10 @@ var globalDBLock = &sync.Mutex{} var globalDB *sqlx.DB var globalDBErr error +func lineIdFromCK(ck base.CommandKey) string { + return ck.GetCmdId() +} + func GetDBName() string { scHome := scbase.GetPromptHomeDir() return path.Join(scHome, DBFileName) @@ -174,7 +179,7 @@ func CloseDB() { type CmdPtr struct { ScreenId string - CmdId string + LineId string } type ClientWinSizeType struct { @@ -373,7 +378,6 @@ func (h *HistoryItemType) ToMap() map[string]interface{} { rtn["lineid"] = h.LineId rtn["linenum"] = h.LineNum rtn["haderror"] = h.HadError - rtn["cmdid"] = h.CmdId rtn["cmdstr"] = h.CmdStr rtn["remoteownerid"] = h.Remote.OwnerId rtn["remoteid"] = h.Remote.RemoteId @@ -391,7 +395,6 @@ func (h *HistoryItemType) FromMap(m map[string]interface{}) bool { quickSetStr(&h.ScreenId, m, "screenid") quickSetStr(&h.LineId, m, "lineid") quickSetBool(&h.HadError, m, "haderror") - quickSetStr(&h.CmdId, m, "cmdid") quickSetStr(&h.CmdStr, m, "cmdstr") quickSetStr(&h.Remote.OwnerId, m, "remoteownerid") quickSetStr(&h.Remote.RemoteId, m, "remoteid") @@ -535,7 +538,6 @@ type HistoryItemType struct { ScreenId string `json:"screenid"` LineId string `json:"lineid"` HadError bool `json:"haderror"` - CmdId string `json:"cmdid"` CmdStr string `json:"cmdstr"` Remote RemotePtrType `json:"remote"` IsMetaCmd bool `json:"ismetacmd"` @@ -715,7 +717,6 @@ type LineType struct { LineType string `json:"linetype"` Renderer string `json:"renderer,omitempty"` Text string `json:"text,omitempty"` - CmdId string `json:"cmdid,omitempty"` Ephemeral bool `json:"ephemeral,omitempty"` ContentHeight int64 `json:"contentheight,omitempty"` Star bool `json:"star,omitempty"` @@ -932,35 +933,26 @@ func (r *RemoteType) GetName() string { return r.RemoteCanonicalName } -type CmdDoneInfo struct { - Ts int64 `json:"ts"` - ExitCode int64 `json:"exitcode"` - DurationMs int64 `json:"durationms"` -} - -type CmdMapType struct { - SessionId string `json:"sessionid"` - ScreenId string `json:"screenid"` - CmdId string `json:"cmdid"` -} - type CmdType struct { - ScreenId string `json:"screenid"` - CmdId string `json:"cmdid"` - Remote RemotePtrType `json:"remote"` - CmdStr string `json:"cmdstr"` - RawCmdStr string `json:"rawcmdstr"` - FeState map[string]string `json:"festate"` - StatePtr ShellStatePtr `json:"state"` - TermOpts TermOpts `json:"termopts"` - OrigTermOpts TermOpts `json:"origtermopts"` - Status string `json:"status"` - StartPk *packet.CmdStartPacketType `json:"startpk,omitempty"` - DoneInfo *CmdDoneInfo `json:"doneinfo,omitempty"` - RunOut []packet.PacketType `json:"runout,omitempty"` - RtnState bool `json:"rtnstate,omitempty"` - RtnStatePtr ShellStatePtr `json:"rtnstateptr,omitempty"` - Remove bool `json:"remove,omitempty"` + ScreenId string `json:"screenid"` + LineId string `json:"lineid"` + Remote RemotePtrType `json:"remote"` + CmdStr string `json:"cmdstr"` + RawCmdStr string `json:"rawcmdstr"` + FeState map[string]string `json:"festate"` + StatePtr ShellStatePtr `json:"state"` + TermOpts TermOpts `json:"termopts"` + OrigTermOpts TermOpts `json:"origtermopts"` + Status string `json:"status"` + CmdPid int `json:"cmdpid"` + RemotePid int `json:"remotepid"` + DoneTs int64 `json:"donets"` + ExitCode int `json:"exitcode"` + DurationMs int `json:"durationms"` + RunOut []packet.PacketType `json:"runout,omitempty"` + RtnState bool `json:"rtnstate,omitempty"` + RtnStatePtr ShellStatePtr `json:"rtnstateptr,omitempty"` + Remove bool `json:"remove,omitempty"` } func (r *RemoteType) ToMap() map[string]interface{} { @@ -1007,7 +999,7 @@ func (r *RemoteType) FromMap(m map[string]interface{}) bool { func (cmd *CmdType) ToMap() map[string]interface{} { rtn := make(map[string]interface{}) rtn["screenid"] = cmd.ScreenId - rtn["cmdid"] = cmd.CmdId + rtn["lineid"] = cmd.LineId rtn["remoteownerid"] = cmd.Remote.OwnerId rtn["remoteid"] = cmd.Remote.RemoteId rtn["remotename"] = cmd.Remote.Name @@ -1019,8 +1011,11 @@ func (cmd *CmdType) ToMap() map[string]interface{} { rtn["termopts"] = quickJson(cmd.TermOpts) rtn["origtermopts"] = quickJson(cmd.OrigTermOpts) rtn["status"] = cmd.Status - rtn["startpk"] = quickJson(cmd.StartPk) - rtn["doneinfo"] = quickJson(cmd.DoneInfo) + rtn["cmdpid"] = cmd.CmdPid + rtn["remotepid"] = cmd.RemotePid + rtn["donets"] = cmd.DoneTs + rtn["exitcode"] = cmd.ExitCode + rtn["durationms"] = cmd.DurationMs rtn["runout"] = quickJson(cmd.RunOut) rtn["rtnstate"] = cmd.RtnState rtn["rtnbasehash"] = cmd.RtnStatePtr.BaseHash @@ -1030,7 +1025,7 @@ func (cmd *CmdType) ToMap() map[string]interface{} { func (cmd *CmdType) FromMap(m map[string]interface{}) bool { quickSetStr(&cmd.ScreenId, m, "screenid") - quickSetStr(&cmd.CmdId, m, "cmdid") + quickSetStr(&cmd.LineId, m, "lineid") quickSetStr(&cmd.Remote.OwnerId, m, "remoteownerid") quickSetStr(&cmd.Remote.RemoteId, m, "remoteid") quickSetStr(&cmd.Remote.Name, m, "remotename") @@ -1042,8 +1037,11 @@ func (cmd *CmdType) FromMap(m map[string]interface{}) bool { quickSetJson(&cmd.TermOpts, m, "termopts") quickSetJson(&cmd.OrigTermOpts, m, "origtermopts") quickSetStr(&cmd.Status, m, "status") - quickSetJson(&cmd.StartPk, m, "startpk") - quickSetJson(&cmd.DoneInfo, m, "doneinfo") + quickSetInt(&cmd.CmdPid, m, "cmdpid") + quickSetInt(&cmd.RemotePid, m, "remotepid") + quickSetInt64(&cmd.DoneTs, m, "donets") + quickSetInt(&cmd.ExitCode, m, "exitcode") + quickSetInt(&cmd.DurationMs, m, "durationms") quickSetJson(&cmd.RunOut, m, "runout") quickSetBool(&cmd.RtnState, m, "rtnstate") quickSetStr(&cmd.RtnStatePtr.BaseHash, m, "rtnbasehash") @@ -1051,15 +1049,19 @@ func (cmd *CmdType) FromMap(m map[string]interface{}) bool { return true } -func makeNewLineCmd(screenId string, userId string, cmdId string, renderer string) *LineType { +func (cmd *CmdType) IsRunning() bool { + return cmd.Status == CmdStatusRunning || cmd.Status == CmdStatusDetached +} + +func makeNewLineCmd(screenId string, userId string, lineId string, renderer string) *LineType { rtn := &LineType{} rtn.ScreenId = screenId rtn.UserId = userId - rtn.LineId = scbase.GenPromptUUID() + rtn.LineId = lineId rtn.Ts = time.Now().UnixMilli() rtn.LineLocal = true rtn.LineType = LineTypeCmd - rtn.CmdId = cmdId + rtn.LineId = lineId rtn.ContentHeight = LineNoHeight rtn.Renderer = renderer return rtn @@ -1078,12 +1080,11 @@ func makeNewLineText(screenId string, userId string, text string) *LineType { return rtn } -func makeNewLineOpenAI(screenId string, userId string, cmdId string) *LineType { +func makeNewLineOpenAI(screenId string, userId string, lineId string) *LineType { rtn := &LineType{} rtn.ScreenId = screenId rtn.UserId = userId - rtn.LineId = scbase.GenPromptUUID() - rtn.CmdId = cmdId + rtn.LineId = lineId rtn.Ts = time.Now().UnixMilli() rtn.LineLocal = true rtn.LineType = LineTypeOpenAI @@ -1102,7 +1103,7 @@ func AddCommentLine(ctx context.Context, screenId string, userId string, comment } func AddOpenAILine(ctx context.Context, screenId string, userId string, cmd *CmdType) (*LineType, error) { - rtnLine := makeNewLineOpenAI(screenId, userId, cmd.CmdId) + rtnLine := makeNewLineOpenAI(screenId, userId, cmd.LineId) err := InsertLine(ctx, rtnLine, cmd) if err != nil { return nil, err @@ -1111,7 +1112,7 @@ func AddOpenAILine(ctx context.Context, screenId string, userId string, cmd *Cmd } func AddCmdLine(ctx context.Context, screenId string, userId string, cmd *CmdType, renderer string) (*LineType, error) { - rtnLine := makeNewLineCmd(screenId, userId, cmd.CmdId, renderer) + rtnLine := makeNewLineCmd(screenId, userId, cmd.LineId, renderer) err := InsertLine(ctx, rtnLine, cmd) if err != nil { return nil, err diff --git a/pkg/sstore/sstore_migrate.go b/pkg/sstore/sstore_migrate.go index 909ff464..48872c57 100644 --- a/pkg/sstore/sstore_migrate.go +++ b/pkg/sstore/sstore_migrate.go @@ -10,12 +10,20 @@ import ( "github.com/commandlinedev/prompt-server/pkg/scbase" ) -type cmdMigrationType struct { +const MigrationChunkSize = 10 + +type cmdMigration13Type struct { SessionId string ScreenId string CmdId string } +type cmdMigration20Type struct { + ScreenId string + LineId string + CmdId string +} + func getSliceChunk[T any](slice []T, chunkSize int) ([]T, []T) { if chunkSize >= len(slice) { return slice, nil @@ -23,10 +31,65 @@ func getSliceChunk[T any](slice []T, chunkSize int) ([]T, []T) { return slice[0:chunkSize], slice[chunkSize:] } -func RunCmdScreenMigration13() error { +func RunMigration20() error { ctx := context.Background() startTime := time.Now() - var migrations []cmdMigrationType + var migrations []cmdMigration20Type + txErr := WithTx(ctx, func(tx *TxWrap) error { + tx.Select(&migrations, `SELECT * FROM cmd_migrate`) + return nil + }) + if txErr != nil { + return fmt.Errorf("trying to get cmd migrations: %w", txErr) + } + log.Printf("[db] got %d cmd-line migrations\n", len(migrations)) + for len(migrations) > 0 { + var mchunk []cmdMigration20Type + mchunk, migrations = getSliceChunk(migrations, MigrationChunkSize) + err := processMigration20Chunk(ctx, mchunk) + if err != nil { + return fmt.Errorf("cmd migration failed on chunk: %w", err) + } + } + log.Printf("[db] cmd line migration done: %v\n", time.Since(startTime)) + return nil +} + +func processMigration20Chunk(ctx context.Context, mchunk []cmdMigration20Type) error { + for _, mig := range mchunk { + newFile, err := scbase.PtyOutFile(mig.ScreenId, mig.LineId) + if err != nil { + log.Printf("ptyoutfile(lineid) error: %v\n", err) + continue + } + oldFile, err := scbase.PtyOutFile(mig.ScreenId, mig.CmdId) + if err != nil { + log.Printf("ptyoutfile(cmdid) error: %v\n", err) + continue + } + err = os.Rename(oldFile, newFile) + if err != nil { + log.Printf("error renaming %s => %s: %v\n", oldFile, newFile, err) + continue + } + } + txErr := WithTx(ctx, func(tx *TxWrap) error { + for _, mig := range mchunk { + query := `DELETE FROM cmd_migrate WHERE cmdid = ?` + tx.Exec(query, mig.CmdId) + } + return nil + }) + if txErr != nil { + return txErr + } + return nil +} + +func RunMigration13() error { + ctx := context.Background() + startTime := time.Now() + var migrations []cmdMigration13Type txErr := WithTx(ctx, func(tx *TxWrap) error { tx.Select(&migrations, `SELECT * FROM cmd_migrate`) return nil @@ -36,9 +99,9 @@ func RunCmdScreenMigration13() error { } log.Printf("[db] got %d cmd-screen migrations\n", len(migrations)) for len(migrations) > 0 { - var mchunk []cmdMigrationType - mchunk, migrations = getSliceChunk(migrations, 5) - err := processMigrationChunk(ctx, mchunk) + var mchunk []cmdMigration13Type + mchunk, migrations = getSliceChunk(migrations, MigrationChunkSize) + err := processMigration13Chunk(ctx, mchunk) if err != nil { return fmt.Errorf("cmd migration failed on chunk: %w", err) } @@ -59,7 +122,7 @@ func RunCmdScreenMigration13() error { return nil } -func processMigrationChunk(ctx context.Context, mchunk []cmdMigrationType) error { +func processMigration13Chunk(ctx context.Context, mchunk []cmdMigration13Type) error { for _, mig := range mchunk { newFile, err := scbase.PtyOutFile(mig.ScreenId, mig.CmdId) if err != nil { diff --git a/pkg/sstore/updatebus.go b/pkg/sstore/updatebus.go index e5382224..72316b3a 100644 --- a/pkg/sstore/updatebus.go +++ b/pkg/sstore/updatebus.go @@ -19,7 +19,7 @@ type UpdatePacket interface { type PtyDataUpdate struct { ScreenId string `json:"screenid,omitempty"` - CmdId string `json:"cmdid,omitempty"` + LineId string `json:"lineid,omitempty"` RemoteId string `json:"remoteid,omitempty"` PtyPos int64 `json:"ptypos"` PtyData64 string `json:"ptydata64"` @@ -72,21 +72,6 @@ type RemoteViewType struct { RemoteEdit *RemoteEditType `json:"remoteedit,omitempty"` } -func ReadHistoryDataFromUpdate(update UpdatePacket) (string, string, *RemotePtrType) { - modelUpdate, ok := update.(*ModelUpdate) - if !ok { - return "", "", nil - } - if modelUpdate.Line == nil { - return "", "", nil - } - var rptr *RemotePtrType - if modelUpdate.Cmd != nil { - rptr = &modelUpdate.Cmd.Remote - } - return modelUpdate.Line.LineId, modelUpdate.Line.CmdId, rptr -} - func InfoMsgUpdate(infoMsgFmt string, args ...interface{}) *ModelUpdate { msg := fmt.Sprintf(infoMsgFmt, args...) return &ModelUpdate{