waveterm/backup
0
0
Fork 0
mirror of https://github.com/wavetermdev/backup.git synced 2026-08-05 13:57:07 -07:00
Code Issues Packages Projects Releases Wiki Activity

checkpoint on big cmd screen migration

Browse Source
This commit is contained in:
sawka
2023-03-20 19:20:57 -07:00
parent cec75c0d5b
commit 02ae7ea972
14 changed files with 607 additions and 351 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -0,0 +1,3 @@
*~
bin/
*.out
cmd/main-server.go
+19 -12
View File
@@ -143,6 +143,12 @@ func HandleGetClientData(w http.ResponseWriter, r *http.Request) {
WriteJsonError(w, err)
return
}
mdata, err := sstore.GetCmdMigrationInfo(r.Context())
if err != nil {
WriteJsonError(w, err)
return
}
cdata.Migration = mdata
WriteJsonSuccess(w, cdata)
return
}
@@ -223,16 +229,16 @@ func HandleRtnState(w http.ResponseWriter, r *http.Request) {
return
}()
qvals := r.URL.Query()
sessionId := qvals.Get("sessionid")
screenId := qvals.Get("screenid")
cmdId := qvals.Get("cmdid")
if sessionId == "" || cmdId == "" {
if screenId == "" || cmdId == "" {
w.WriteHeader(500)
w.Write([]byte(fmt.Sprintf("must specify sessionid and cmdid")))
w.Write([]byte(fmt.Sprintf("must specify screenid and cmdid")))
return
}
if _, err := uuid.Parse(sessionId); err != nil {
if _, err := uuid.Parse(screenId); err != nil {
w.WriteHeader(500)
w.Write([]byte(fmt.Sprintf("invalid sessionid: %v", err)))
w.Write([]byte(fmt.Sprintf("invalid screenid: %v", err)))
return
}
if _, err := uuid.Parse(cmdId); err != nil {
@@ -240,7 +246,7 @@ func HandleRtnState(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(fmt.Sprintf("invalid cmdid: %v", err)))
return
}
data, err := cmdrunner.GetRtnStateDiff(r.Context(), sessionId, cmdId)
data, err := cmdrunner.GetRtnStateDiff(r.Context(), screenId, cmdId)
if err != nil {
w.WriteHeader(500)
w.Write([]byte(fmt.Sprintf("cannot get rtnstate diff: %v", err)))
@@ -278,16 +284,16 @@ func HandleRemotePty(w http.ResponseWriter, r *http.Request) {
func HandleGetPtyOut(w http.ResponseWriter, r *http.Request) {
qvals := r.URL.Query()
sessionId := qvals.Get("sessionid")
screenId := qvals.Get("screenid")
cmdId := qvals.Get("cmdid")
if sessionId == "" || cmdId == "" {
if screenId == "" || cmdId == "" {
w.WriteHeader(500)
w.Write([]byte(fmt.Sprintf("must specify sessionid and cmdid")))
w.Write([]byte(fmt.Sprintf("must specify screenid and cmdid")))
return
}
if _, err := uuid.Parse(sessionId); err != nil {
if _, err := uuid.Parse(screenId); err != nil {
w.WriteHeader(500)
w.Write([]byte(fmt.Sprintf("invalid sessionid: %v", err)))
w.Write([]byte(fmt.Sprintf("invalid screenid: %v", err)))
return
}
if _, err := uuid.Parse(cmdId); err != nil {
@@ -295,7 +301,7 @@ func HandleGetPtyOut(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(fmt.Sprintf("invalid cmdid: %v", err)))
return
}
realOffset, data, err := sstore.ReadFullPtyOutFile(r.Context(), sessionId, cmdId)
realOffset, data, err := sstore.ReadFullPtyOutFile(r.Context(), screenId, cmdId)
if err != nil {
if errors.Is(err, fs.ErrNotExist) {
w.WriteHeader(http.StatusOK)
@@ -558,6 +564,7 @@ func main() {
go telemetryLoop()
go stdinReadWatch()
go runWebSocketServer()
go sstore.RunCmdScreenMigration()
gr := mux.NewRouter()
gr.HandleFunc("/api/ptyout", AuthKeyWrap(HandleGetPtyOut))
gr.HandleFunc("/api/remote-pty", AuthKeyWrap(HandleRemotePty))
db/migrations/000013_cmdmigration.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- invalid, will throw an error, cannot migrate down
SELECT x;
db/migrations/000013_cmdmigration.up.sql
+123
View File
@@ -0,0 +1,123 @@
DELETE FROM cmd
WHERE screenid = '';
DELETE FROM line
WHERE screenid = '';
DELETE FROM cmd
WHERE cmdid NOT IN (SELECT cmdid FROM line);
DELETE FROM line
WHERE cmdid <> '' AND cmdid NOT IN (SELECT cmdid FROM cmd);
CREATE TABLE new_bookmark_cmd (
bookmarkid varchar(36) NOT NULL,
screenid varchar(36) NOT NULL,
cmdid varchar(36) NOT NULL,
PRIMARY KEY (bookmarkid, screenid, cmdid)
);
INSERT INTO new_bookmark_cmd
SELECT
b.bookmarkid,
c.screenid,
c.cmdid
FROM bookmark_cmd b, cmd c
WHERE b.cmdid = c.cmdid;
DROP TABLE bookmark_cmd;
ALTER TABLE new_bookmark_cmd RENAME TO bookmark_cmd;
ALTER TABLE client ADD COLUMN cmdstoretype varchar(20) DEFAULT 'session';
CREATE TABLE cmd_migrate (
sessionid varchar(36) NOT NULL,
screenid varchar(36) NOT NULL,
cmdid varchar(36) NOT NULL
);
INSERT INTO cmd_migrate
SELECT sessionid, screenid, cmdid
FROM cmd;
-- update primary key for screen
CREATE TABLE new_screen (
screenid varchar(36) NOT NULL,
sessionid varchar(36) NOT NULL,
name varchar(50) NOT NULL,
screenidx int NOT NULL,
screenopts json NOT NULL,
ownerid varchar(36) NOT NULL,
sharemode varchar(12) NOT NULL,
curremoteownerid varchar(36) NOT NULL,
curremoteid varchar(36) NOT NULL,
curremotename varchar(50) NOT NULL,
nextlinenum int NOT NULL,
selectedline int NOT NULL,
anchor json NOT NULL,
focustype varchar(12) NOT NULL,
archived boolean NOT NULL,
archivedts bigint NOT NULL,
PRIMARY KEY (screenid)
);
INSERT INTO new_screen
SELECT screenid, sessionid, name, screenidx, screenopts, ownerid, sharemode,
curremoteownerid, curremoteid, curremotename, nextlinenum, selectedline,
anchor, focustype, archived, archivedts
FROM screen;
DROP TABLE screen;
ALTER TABLE new_screen RENAME TO screen;
-- drop sessionid from line
CREATE TABLE new_line (
screenid varchar(36) NOT NULL,
userid varchar(36) NOT NULL,
lineid varchar(36) NOT NULL,
ts bigint NOT NULL,
linenum int NOT NULL,
linenumtemp boolean NOT NULL,
linetype varchar(10) NOT NULL,
linelocal boolean NOT NULL,
text text NOT NULL,
cmdid varchar(36) NOT NULL,
ephemeral boolean NOT NULL,
contentheight int NOT NULL,
star int NOT NULL,
archived boolean NOT NULL,
renderer varchar(50) NOT NULL,
bookmarked boolean NOT NULL,
PRIMARY KEY (screenid, lineid)
);
INSERT INTO new_line
SELECT screenid, userid, lineid, ts, linenum, linenumtemp, linetype, linelocal,
text, cmdid, ephemeral, contentheight, star, archived, renderer, bookmarked
FROM line;
DROP TABLE line;
ALTER TABLE new_line RENAME TO line;
-- drop sessionid from cmd
CREATE TABLE new_cmd (
screenid varchar(36) NOT NULL,
cmdid 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,
startpk json NOT NULL,
doneinfo json NOT NULL,
runout json NOT NULL,
rtnstate boolean NOT NULL,
rtnbasehash varchar(36) NOT NULL,
rtndiffhasharr json NOT NULL,
PRIMARY KEY (screenid, cmdid)
);
INSERT INTO new_cmd
SELECT screenid, cmdid, remoteownerid, remoteid, remotename, cmdstr, cmdstr,
festate, statebasehash, statediffhasharr, termopts, origtermopts, status, startpk, doneinfo, runout, rtnstate, rtnbasehash, rtndiffhasharr
FROM cmd;
DROP TABLE cmd;
ALTER TABLE new_cmd RENAME TO cmd;
pkg/cmdrunner/cmdrunner.go
+40 -41
View File
@@ -417,7 +417,7 @@ func RunCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sstore.U
// runPacket.State is set in remote.RunCommand()
runPacket := packet.MakeRunPacket()
runPacket.ReqId = uuid.New().String()
runPacket.CK = base.MakeCommandKey(ids.SessionId, scbase.GenPromptUUID())
runPacket.CK = base.MakeCommandKey(ids.ScreenId, scbase.GenPromptUUID())
runPacket.UsePty = true
ptermVal := defaultStr(pk.Kwargs["pterm"], DefaultPTERM)
runPacket.TermOpts, err = GetUITermOpts(pk.UIContext.WinSize, ptermVal)
@@ -433,12 +433,13 @@ func RunCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sstore.U
if err != nil {
return nil, err
}
cmd.RawCmdStr = pk.GetRawStr()
update, err := addLineForCmd(ctx, "/run", true, ids, cmd, renderer)
if err != nil {
return nil, err
}
update.Interactive = pk.Interactive
sstore.MainBus.SendUpdate(ids.SessionId, update)
sstore.MainBus.SendScreenUpdate(ids.ScreenId, update)
return nil, nil
}
@@ -559,7 +560,7 @@ func ScreenPurgeCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (
if err != nil {
return nil, fmt.Errorf("/screen:purge cannot purge screen: %w", err)
}
update, err := sstore.DeleteScreen(ctx, ids.ScreenId)
update, err := sstore.PurgeScreen(ctx, ids.ScreenId, false)
if err != nil {
return nil, err
}
@@ -1255,10 +1256,10 @@ func CrCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sstore.Up
func makeStaticCmd(ctx context.Context, metaCmd string, ids resolvedIds, cmdStr string, cmdOutput []byte) (*sstore.CmdType, error) {
cmd := &sstore.CmdType{
SessionId: ids.SessionId,
ScreenId: ids.ScreenId,
CmdId: 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,
@@ -1272,13 +1273,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.SessionId, cmd.CmdId, cmd.TermOpts.MaxPtySize)
err := sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.CmdId, 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, cmd.SessionId, ids.ScreenId, cmd.CmdId, cmdOutput, 0)
_, err = sstore.AppendToCmdPtyBlob(ctx, ids.ScreenId, cmd.CmdId, 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)
@@ -1287,7 +1288,7 @@ func makeStaticCmd(ctx context.Context, metaCmd string, ids resolvedIds, cmdStr
}
func addLineForCmd(ctx context.Context, metaCmd string, shouldFocus bool, ids resolvedIds, cmd *sstore.CmdType, renderer string) (*sstore.ModelUpdate, error) {
rtnLine, err := sstore.AddCmdLine(ctx, ids.SessionId, ids.ScreenId, DefaultUserId, cmd, renderer)
rtnLine, err := sstore.AddCmdLine(ctx, ids.ScreenId, DefaultUserId, cmd, renderer)
if err != nil {
return nil, err
}
@@ -1495,7 +1496,7 @@ func CommentCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (ssto
if strings.TrimSpace(text) == "" {
return nil, fmt.Errorf("cannot post empty comment")
}
rtnLine, err := sstore.AddCommentLine(ctx, ids.SessionId, ids.ScreenId, DefaultUserId, text)
rtnLine, err := sstore.AddCommentLine(ctx, ids.ScreenId, DefaultUserId, text)
if err != nil {
return nil, err
}
@@ -1637,7 +1638,7 @@ func SessionDeleteCommand(ctx context.Context, pk *scpacket.FeCommandPacketType)
if err != nil {
return nil, err
}
update, err := sstore.DeleteSession(ctx, ids.SessionId)
update, err := sstore.PurgeSession(ctx, ids.SessionId)
if err != nil {
return nil, fmt.Errorf("cannot delete session: %v", err)
}
@@ -1907,10 +1908,9 @@ func HistoryPurgeCommand(ctx context.Context, pk *scpacket.FeCommandPacketType)
continue
}
lineObj := &sstore.LineType{
SessionId: historyItem.SessionId,
ScreenId: historyItem.ScreenId,
LineId: historyItem.LineId,
Remove: true,
ScreenId: historyItem.ScreenId,
LineId: historyItem.LineId,
Remove: true,
}
update.Lines = append(update.Lines, lineObj)
}
@@ -2076,7 +2076,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.SessionId, cmd.CmdId)
siPk.CK = base.MakeCommandKey(cmd.ScreenId, cmd.CmdId)
siPk.WinSize = &packet.WinSize{Rows: int(cmd.TermOpts.Rows), Cols: newCols}
msh := remote.GetRemoteById(cmd.Remote.RemoteId)
if msh == nil {
@@ -2088,7 +2088,7 @@ func resizeRunningCommand(ctx context.Context, cmd *sstore.CmdType, newCols int)
}
newTermOpts := cmd.TermOpts
newTermOpts.Cols = int64(newCols)
err = sstore.UpdateCmdTermOpts(ctx, cmd.SessionId, cmd.CmdId, newTermOpts)
err = sstore.UpdateCmdTermOpts(ctx, cmd.ScreenId, cmd.CmdId, newTermOpts)
if err != nil {
return err
}
@@ -2112,7 +2112,7 @@ func ScreenResizeCommand(ctx context.Context, pk *scpacket.FeCommandPacketType)
return nil, fmt.Errorf("/screen:resize invalid zero/negative 'cols' argument")
}
cols = base.BoundInt(cols, shexec.MinTermCols, shexec.MaxTermCols)
runningCmds, err := sstore.GetRunningScreenCmds(ctx, ids.SessionId, ids.ScreenId)
runningCmds, err := sstore.GetRunningScreenCmds(ctx, ids.ScreenId)
if err != nil {
return nil, fmt.Errorf("/screen:resize cannot get running commands: %v", err)
}
@@ -2140,7 +2140,7 @@ func LineSetHeightCommand(ctx context.Context, pk *scpacket.FeCommandPacketType)
return nil, fmt.Errorf("/line:setheight requires 2 arguments (linearg and height)")
}
lineArg := pk.Args[0]
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
@@ -2168,7 +2168,7 @@ func LineSetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (ssto
return nil, fmt.Errorf("/line:set requires 1 argument (linearg)")
}
lineArg := pk.Args[0]
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
@@ -2186,7 +2186,7 @@ func LineSetCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (ssto
if len(varsUpdated) == 0 {
return nil, fmt.Errorf("/line:set requires a value to set: %s", formatStrs([]string{"renderer"}, "or", false))
}
updatedLine, err := sstore.GetLineById(ctx, ids.SessionId, ids.ScreenId, lineId)
updatedLine, err := sstore.GetLineById(ctx, ids.ScreenId, lineId)
if err != nil {
return nil, fmt.Errorf("/line:set cannot retrieve updated line: %v", err)
}
@@ -2332,21 +2332,21 @@ func LineBookmarkCommand(ctx context.Context, pk *scpacket.FeCommandPacketType)
return nil, fmt.Errorf("/line:bookmark requires an argument (line number or id)")
}
lineArg := pk.Args[0]
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
if lineId == "" {
return nil, fmt.Errorf("line %q not found", lineArg)
}
lineObj, cmdObj, err := sstore.GetLineCmdByLineId(ctx, ids.SessionId, ids.ScreenId, lineId)
lineObj, cmdObj, err := sstore.GetLineCmdByLineId(ctx, ids.ScreenId, lineId)
if err != nil {
return nil, fmt.Errorf("/line:bookmark error getting line: %v", err)
}
if cmdObj == nil {
return nil, fmt.Errorf("cannot bookmark non-cmd line")
}
ck := base.MakeCommandKey(lineObj.SessionId, cmdObj.CmdId)
ck := base.MakeCommandKey(lineObj.ScreenId, cmdObj.CmdId)
bm := &sstore.BookmarkType{
BookmarkId: uuid.New().String(),
CreatedTs: time.Now().UnixMilli(),
@@ -2360,7 +2360,7 @@ func LineBookmarkCommand(ctx context.Context, pk *scpacket.FeCommandPacketType)
if err != nil {
return nil, fmt.Errorf("cannot insert bookmark: %v", err)
}
newLineObj, err := sstore.GetLineById(ctx, ids.SessionId, ids.ScreenId, lineId)
newLineObj, err := sstore.GetLineById(ctx, ids.ScreenId, lineId)
if err != nil {
return nil, fmt.Errorf("/line:bookmark error getting line: %v", err)
}
@@ -2387,7 +2387,7 @@ func LineStarCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sst
return nil, fmt.Errorf("/line:star only takes up to 2 arguments (line-number and star-value)")
}
lineArg := pk.Args[0]
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
@@ -2405,7 +2405,7 @@ func LineStarCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sst
if err != nil {
return nil, fmt.Errorf("/line:star error updating star value: %v", err)
}
lineObj, err := sstore.GetLineById(ctx, ids.SessionId, ids.ScreenId, lineId)
lineObj, err := sstore.GetLineById(ctx, ids.ScreenId, lineId)
if err != nil {
return nil, fmt.Errorf("/line:star error getting line: %v", err)
}
@@ -2425,7 +2425,7 @@ func LineArchiveCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (
return nil, fmt.Errorf("/line:archive requires an argument (line number or id)")
}
lineArg := pk.Args[0]
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
@@ -2440,7 +2440,7 @@ func LineArchiveCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (
if err != nil {
return nil, fmt.Errorf("/line:archive error updating hidden status: %v", err)
}
lineObj, err := sstore.GetLineById(ctx, ids.SessionId, ids.ScreenId, lineId)
lineObj, err := sstore.GetLineById(ctx, ids.ScreenId, lineId)
if err != nil {
return nil, fmt.Errorf("/line:archive error getting line: %v", err)
}
@@ -2461,7 +2461,7 @@ func LinePurgeCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (ss
}
var lineIds []string
for _, lineArg := range pk.Args {
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
@@ -2470,17 +2470,16 @@ func LinePurgeCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (ss
}
lineIds = append(lineIds, lineId)
}
err = sstore.PurgeLinesByIds(ctx, ids.SessionId, lineIds)
err = sstore.PurgeLinesByIds(ctx, ids.ScreenId, lineIds)
if err != nil {
return nil, fmt.Errorf("/line:purge error purging lines: %v", err)
}
update := sstore.ModelUpdate{}
for _, lineId := range lineIds {
lineObj := &sstore.LineType{
SessionId: ids.SessionId,
ScreenId: ids.ScreenId,
LineId: lineId,
Remove: true,
ScreenId: ids.ScreenId,
LineId: lineId,
Remove: true,
}
update.Lines = append(update.Lines, lineObj)
}
@@ -2496,14 +2495,14 @@ func LineShowCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sst
return nil, fmt.Errorf("/line:show requires an argument (line number or id)")
}
lineArg := pk.Args[0]
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
if lineId == "" {
return nil, fmt.Errorf("line %q not found", lineArg)
}
line, cmd, err := sstore.GetLineCmdByLineId(ctx, ids.SessionId, ids.ScreenId, lineId)
line, cmd, err := sstore.GetLineCmdByLineId(ctx, ids.ScreenId, lineId)
if err != nil {
return nil, fmt.Errorf("error getting line: %v", err)
}
@@ -2542,7 +2541,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.SessionId, cmd.CmdId)
stat, _ := sstore.StatCmdPtyFile(ctx, cmd.ScreenId, cmd.CmdId)
if stat == nil {
buf.WriteString(fmt.Sprintf(" %-15s %s\n", "file", "-"))
} else {
@@ -2599,11 +2598,11 @@ func SignalCommand(ctx context.Context, pk *scpacket.FeCommandPacketType) (sstor
return nil, fmt.Errorf("/signal requires a second argument (signal name)")
}
lineArg := pk.Args[0]
lineId, err := sstore.FindLineIdByArg(ctx, ids.SessionId, ids.ScreenId, lineArg)
lineId, err := sstore.FindLineIdByArg(ctx, ids.ScreenId, lineArg)
if err != nil {
return nil, fmt.Errorf("error looking up lineid: %v", err)
}
line, cmd, err := sstore.GetLineCmdByLineId(ctx, ids.SessionId, ids.ScreenId, lineId)
line, cmd, err := sstore.GetLineCmdByLineId(ctx, ids.ScreenId, lineId)
if err != nil {
return nil, fmt.Errorf("error getting line: %v", err)
}
@@ -2640,7 +2639,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.SessionId, cmd.CmdId)
siPk.CK = base.MakeCommandKey(cmd.ScreenId, cmd.CmdId)
siPk.SigName = sigArg
err = msh.SendSpecialInput(siPk)
if err != nil {
@@ -2950,8 +2949,8 @@ func displayStateUpdateDiff(buf *bytes.Buffer, oldState packet.ShellState, newSt
}
}
func GetRtnStateDiff(ctx context.Context, sessionId string, cmdId string) ([]byte, error) {
cmd, err := sstore.GetCmdById(ctx, sessionId, cmdId)
func GetRtnStateDiff(ctx context.Context, screenId string, cmdId string) ([]byte, error) {
cmd, err := sstore.GetCmdByScreenId(ctx, screenId, cmdId)
if err != nil {
return nil, err
}
pkg/cmdrunner/resolver.go
+1 -1
View File
@@ -298,7 +298,7 @@ func resolveSession(ctx context.Context, sessionArg string, curSessionArg string
}
func resolveLine(ctx context.Context, sessionId string, screenId string, lineArg string, curLineArg string) (*ResolveItem, error) {
lines, err := sstore.GetLineResolveItems(ctx, sessionId, screenId)
lines, err := sstore.GetLineResolveItems(ctx, screenId)
if err != nil {
return nil, fmt.Errorf("could not get lines: %v", err)
}
pkg/remote/remote.go
+18 -18
View File
@@ -578,7 +578,7 @@ func (msh *MShellProc) GetRemoteRuntimeState() RemoteRuntimeState {
func (msh *MShellProc) NotifyRemoteUpdate() {
rstate := msh.GetRemoteRuntimeState()
update := &sstore.ModelUpdate{Remotes: []interface{}{rstate}}
sstore.MainBus.SendUpdate("", update)
sstore.MainBus.SendUpdate(update)
}
func GetAllRemoteRuntimeState() []RemoteRuntimeState {
@@ -811,7 +811,7 @@ func sendRemotePtyUpdate(remoteId string, dataOffset int64, data []byte) {
PtyData64: data64,
PtyDataLen: int64(len(data)),
}
sstore.MainBus.SendUpdate("", update)
sstore.MainBus.SendUpdate(update)
}
func (msh *MShellProc) isWaitingForPassword_nolock() bool {
@@ -1265,8 +1265,8 @@ func RunCommand(ctx context.Context, sessionId string, screenId string, remotePt
if remotePtr.OwnerId != "" {
return nil, nil, fmt.Errorf("cannot run command against another user's remote '%s'", remotePtr.MakeFullRemoteRef())
}
if sessionId != runPacket.CK.GetSessionId() {
return nil, nil, fmt.Errorf("run commands sessionids do not match")
if screenId != runPacket.CK.GetGroupId() {
return nil, nil, fmt.Errorf("run commands screenids do not match")
}
msh := GetRemoteById(remotePtr.RemoteId)
if msh == nil {
@@ -1284,7 +1284,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, sessionId, screenId, existingPSC.GetCmdId())
line, _, err := sstore.GetLineCmdByCmdId(ctx, screenId, existingPSC.GetCmdId())
if err != nil {
return nil, nil, fmt.Errorf("cannot run command while a stateful command is still running: %v", err)
}
@@ -1344,10 +1344,10 @@ func RunCommand(ctx context.Context, sessionId string, screenId string, remotePt
status = sstore.CmdStatusDetached
}
cmd := &sstore.CmdType{
SessionId: runPacket.CK.GetSessionId(),
ScreenId: screenId,
ScreenId: runPacket.CK.GetGroupId(),
CmdId: runPacket.CK.GetCmdId(),
CmdStr: runPacket.Command,
RawCmdStr: runPacket.Command,
Remote: remotePtr,
FeState: *sstore.FeStateFromShellState(currentState),
StatePtr: *statePtr,
@@ -1358,7 +1358,7 @@ func RunCommand(ctx context.Context, sessionId string, screenId string, remotePt
RunOut: nil,
RtnState: runPacket.ReturnState,
}
err = sstore.CreateCmdPtyFile(ctx, cmd.SessionId, cmd.CmdId, cmd.TermOpts.MaxPtySize)
err = sstore.CreateCmdPtyFile(ctx, cmd.ScreenId, cmd.CmdId, 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)
@@ -1476,12 +1476,12 @@ func makeDataAckPacket(ck base.CommandKey, fdNum int, ackLen int, err error) *pa
func (msh *MShellProc) notifyHangups_nolock() {
for ck, _ := range msh.RunningCmds {
cmd, err := sstore.GetCmdById(context.Background(), ck.GetSessionId(), ck.GetCmdId())
cmd, err := sstore.GetCmdByScreenId(context.Background(), ck.GetGroupId(), ck.GetCmdId())
if err != nil {
continue
}
update := sstore.ModelUpdate{Cmd: cmd}
sstore.MainBus.SendUpdate(ck.GetSessionId(), update)
sstore.MainBus.SendScreenUpdate(ck.GetGroupId(), update)
}
msh.RunningCmds = make(map[base.CommandKey]RunCmdType)
}
@@ -1505,12 +1505,12 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
msh.WriteToPtyBuffer("*error updating cmddone: %v\n", err)
return
}
screens, err := sstore.UpdateScreensWithCmdFg(context.Background(), donePk.CK.GetSessionId(), donePk.CK.GetCmdId())
screen, err := sstore.UpdateScreenWithCmdFg(context.Background(), donePk.CK.GetGroupId(), donePk.CK.GetCmdId())
if err != nil {
msh.WriteToPtyBuffer("*error trying to update cmd-fg screens: %v\n", err)
// fall-through (nothing to do)
}
update.Screens = screens
update.Screens = []*sstore.ScreenType{screen}
rct := msh.GetRunningCmd(donePk.CK)
var statePtr *sstore.ShellStatePtr
if donePk.FinalState != nil && rct != nil {
@@ -1550,13 +1550,13 @@ func (msh *MShellProc) handleCmdDonePacket(donePk *packet.CmdDonePacketType) {
// fall-through (nothing to do)
}
}
sstore.MainBus.SendUpdate(donePk.CK.GetSessionId(), update)
sstore.MainBus.SendScreenUpdate(donePk.CK.GetGroupId(), update)
return
}
func (msh *MShellProc) handleCmdFinalPacket(finalPk *packet.CmdFinalPacketType) {
defer msh.RemoveRunningCmd(finalPk.CK)
rtnCmd, err := sstore.GetCmdById(context.Background(), finalPk.CK.GetSessionId(), finalPk.CK.GetCmdId())
rtnCmd, err := sstore.GetCmdByScreenId(context.Background(), finalPk.CK.GetGroupId(), finalPk.CK.GetCmdId())
if err != nil {
log.Printf("error calling GetCmdById in handleCmdFinalPacket: %v\n", err)
return
@@ -1566,7 +1566,7 @@ func (msh *MShellProc) handleCmdFinalPacket(finalPk *packet.CmdFinalPacketType)
}
log.Printf("finalpk %s (hangup): %s\n", finalPk.CK, finalPk.Error)
sstore.HangupCmd(context.Background(), finalPk.CK)
rtnCmd, err = sstore.GetCmdById(context.Background(), finalPk.CK.GetSessionId(), finalPk.CK.GetCmdId())
rtnCmd, err = sstore.GetCmdByScreenId(context.Background(), finalPk.CK.GetGroupId(), finalPk.CK.GetCmdId())
if err != nil {
log.Printf("error getting cmd(2) in handleCmdFinalPacket: %v\n", err)
return
@@ -1576,7 +1576,7 @@ func (msh *MShellProc) handleCmdFinalPacket(finalPk *packet.CmdFinalPacketType)
return
}
update := &sstore.ModelUpdate{Cmd: rtnCmd}
sstore.MainBus.SendUpdate(finalPk.CK.GetSessionId(), update)
sstore.MainBus.SendScreenUpdate(finalPk.CK.GetGroupId(), update)
}
// TODO notify FE about cmd errors
@@ -1600,7 +1600,7 @@ func (msh *MShellProc) handleDataPacket(dataPk *packet.DataPacketType, dataPosMa
if len(realData) > 0 {
dataPos := dataPosMap[dataPk.CK]
rcmd := msh.GetRunningCmd(dataPk.CK)
update, err := sstore.AppendToCmdPtyBlob(context.Background(), dataPk.CK.GetSessionId(), rcmd.ScreenId, dataPk.CK.GetCmdId(), realData, dataPos)
update, err := sstore.AppendToCmdPtyBlob(context.Background(), rcmd.ScreenId, dataPk.CK.GetCmdId(), realData, dataPos)
if err != nil {
ack = makeDataAckPacket(dataPk.CK, dataPk.FdNum, 0, err)
} else {
@@ -1608,7 +1608,7 @@ func (msh *MShellProc) handleDataPacket(dataPk *packet.DataPacketType, dataPosMa
}
dataPosMap[dataPk.CK] += int64(len(realData))
if update != nil {
sstore.MainBus.SendUpdate(dataPk.CK.GetSessionId(), update)
sstore.MainBus.SendScreenUpdate(dataPk.CK.GetGroupId(), update)
}
}
if ack != nil {
pkg/scbase/scbase.go
+40 -19
View File
@@ -27,6 +27,7 @@ const HomeVarName = "HOME"
const PromptHomeVarName = "PROMPT_HOME"
const PromptDevVarName = "PROMPT_DEV"
const SessionsDirBaseName = "sessions"
const ScreensDirBaseName = "screens"
const PromptLockFile = "prompt.lock"
const PromptDirName = "prompt"
const PromptDevDirName = "prompt-dev"
@@ -36,6 +37,7 @@ const PromptAuthKeyFileName = "prompt.authkey"
const MShellVersion = "v0.2.0"
var SessionDirCache = make(map[string]string)
var ScreenDirCache = make(map[string]string)
var BaseLock = &sync.Mutex{}
var BuildTime = "-"
@@ -164,6 +166,7 @@ func AcquirePromptLock() (*os.File, error) {
return fd, nil
}
// deprecated (v0.1.8)
func EnsureSessionDir(sessionId string) (string, error) {
if sessionId == "" {
return "", fmt.Errorf("cannot get session dir for blank sessionid")
@@ -186,12 +189,41 @@ func EnsureSessionDir(sessionId string) (string, error) {
return sdir, nil
}
// deprecated (v0.1.8)
func GetSessionsDir() string {
promptHome := GetPromptHomeDir()
sdir := path.Join(promptHome, SessionsDirBaseName)
return sdir
}
func EnsureScreenDir(screenId string) (string, error) {
if screenId == "" {
return "", fmt.Errorf("cannot get screen dir for blank sessionid")
}
BaseLock.Lock()
sdir, ok := ScreenDirCache[screenId]
BaseLock.Unlock()
if ok {
return sdir, nil
}
scHome := GetPromptHomeDir()
sdir = path.Join(scHome, ScreensDirBaseName, screenId)
err := ensureDir(sdir)
if err != nil {
return "", err
}
BaseLock.Lock()
ScreenDirCache[screenId] = sdir
BaseLock.Unlock()
return sdir, nil
}
func GetScreensDir() string {
promptHome := GetPromptHomeDir()
sdir := path.Join(promptHome, ScreensDirBaseName)
return sdir
}
func ensureDir(dirName string) error {
info, err := os.Stat(dirName)
if errors.Is(err, fs.ErrNotExist) {
@@ -211,7 +243,8 @@ func ensureDir(dirName string) error {
return nil
}
func PtyOutFile(sessionId string, cmdId string) (string, error) {
// deprecated (v0.1.8)
func PtyOutFile_Sessions(sessionId string, cmdId string) (string, error) {
sdir, err := EnsureSessionDir(sessionId)
if err != nil {
return "", err
@@ -225,30 +258,18 @@ func PtyOutFile(sessionId string, cmdId string) (string, error) {
return fmt.Sprintf("%s/%s.ptyout.cf", sdir, cmdId), nil
}
func RunOutFile(sessionId string, cmdId string) (string, error) {
sdir, err := EnsureSessionDir(sessionId)
func PtyOutFile(screenId string, cmdId string) (string, error) {
sdir, err := EnsureScreenDir(screenId)
if err != nil {
return "", err
}
if sessionId == "" {
return "", fmt.Errorf("cannot get runout file for blank sessionid")
if screenId == "" {
return "", fmt.Errorf("cannot get ptyout file for blank screenid")
}
if cmdId == "" {
return "", fmt.Errorf("cannot get runout file for blank cmdid")
return "", fmt.Errorf("cannot get ptyout file for blank cmdid")
}
return fmt.Sprintf("%s/%s.runout", sdir, cmdId), nil
}
type PromptFileNameGenerator struct {
PromptHome string
}
func (g PromptFileNameGenerator) PtyOutFile(ck base.CommandKey) string {
return path.Join(g.PromptHome, SessionsDirBaseName, ck.GetSessionId(), ck.GetCmdId()+".ptyout")
}
func (g PromptFileNameGenerator) RunOutFile(ck base.CommandKey) string {
return path.Join(g.PromptHome, SessionsDirBaseName, ck.GetSessionId(), ck.GetCmdId()+".runout")
return fmt.Sprintf("%s/%s.ptyout.cf", sdir, cmdId), nil
}
func GenPromptUUID() string {
pkg/scws/scws.go
+1 -1
View File
@@ -99,7 +99,7 @@ func (ws *WSState) WatchScreen(sessionId string, screenId string) {
}
ws.SessionId = sessionId
ws.ScreenId = screenId
ws.UpdateCh = sstore.MainBus.RegisterChannel(ws.ClientId, ws.SessionId)
ws.UpdateCh = sstore.MainBus.RegisterChannel(ws.ClientId, ws.ScreenId)
go ws.RunUpdates(ws.UpdateCh)
}
pkg/sstore/dbops.go
+160 -210
View File
File diff suppressed because it is too large Load Diff
pkg/sstore/fileops.go
+19 -11
View File
@@ -12,8 +12,8 @@ import (
"github.com/scripthaus-dev/sh2-server/pkg/scbase"
)
func CreateCmdPtyFile(ctx context.Context, sessionId string, cmdId string, maxSize int64) error {
ptyOutFileName, err := scbase.PtyOutFile(sessionId, cmdId)
func CreateCmdPtyFile(ctx context.Context, screenId string, cmdId string, maxSize int64) error {
ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId)
if err != nil {
return err
}
@@ -24,22 +24,22 @@ func CreateCmdPtyFile(ctx context.Context, sessionId string, cmdId string, maxSi
return f.Close()
}
func StatCmdPtyFile(ctx context.Context, sessionId string, cmdId string) (*cirfile.Stat, error) {
ptyOutFileName, err := scbase.PtyOutFile(sessionId, cmdId)
func StatCmdPtyFile(ctx context.Context, screenId string, cmdId string) (*cirfile.Stat, error) {
ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId)
if err != nil {
return nil, err
}
return cirfile.StatCirFile(ctx, ptyOutFileName)
}
func AppendToCmdPtyBlob(ctx context.Context, sessionId string, screenId string, cmdId string, data []byte, pos int64) (*PtyDataUpdate, error) {
func AppendToCmdPtyBlob(ctx context.Context, screenId string, cmdId 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(sessionId, cmdId)
ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId)
if err != nil {
return nil, err
}
@@ -54,7 +54,6 @@ func AppendToCmdPtyBlob(ctx context.Context, sessionId string, screenId string,
}
data64 := base64.StdEncoding.EncodeToString(data)
update := &PtyDataUpdate{
SessionId: sessionId,
ScreenId: screenId,
CmdId: cmdId,
PtyPos: pos,
@@ -65,8 +64,8 @@ func AppendToCmdPtyBlob(ctx context.Context, sessionId string, screenId string,
}
// returns (offset, data, err)
func ReadFullPtyOutFile(ctx context.Context, sessionId string, cmdId string) (int64, []byte, error) {
ptyOutFileName, err := scbase.PtyOutFile(sessionId, cmdId)
func ReadFullPtyOutFile(ctx context.Context, screenId string, cmdId string) (int64, []byte, error) {
ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId)
if err != nil {
return 0, nil, err
}
@@ -141,8 +140,8 @@ func FullSessionDiskSize() (map[string]SessionDiskSizeType, error) {
return rtn, nil
}
func DeletePtyOutFile(ctx context.Context, sessionId string, cmdId string) error {
ptyOutFileName, err := scbase.PtyOutFile(sessionId, cmdId)
func DeletePtyOutFile(ctx context.Context, screenId string, cmdId string) error {
ptyOutFileName, err := scbase.PtyOutFile(screenId, cmdId)
if err != nil {
return err
}
@@ -157,3 +156,12 @@ func DeleteSessionDir(ctx context.Context, sessionId string) error {
fmt.Printf("remove-all %s\n", sessionDir)
return os.RemoveAll(sessionDir)
}
func DeleteScreenDir(ctx context.Context, screenId string) error {
screenDir, err := scbase.EnsureScreenDir(screenId)
if err != nil {
return fmt.Errorf("error getting screendir: %w", err)
}
fmt.Printf("remove-all %s\n", screenDir)
return os.RemoveAll(screenDir)
}
pkg/sstore/migrate.go
+3 -1
View File
@@ -17,7 +17,7 @@ import (
"github.com/golang-migrate/migrate/v4"
)
const MaxMigration = 12
const MaxMigration = 13
const MigratePrimaryScreenVersion = 9
func MakeMigrate() (*migrate.Migrate, error) {
@@ -81,7 +81,9 @@ func MigrateUp() error {
if err != nil {
return fmt.Errorf("error creating database backup: %v", err)
}
startTime := time.Now()
err = m.Migrate(MaxMigration)
log.Printf("[db] migration took %v\n", time.Since(startTime))
if err != nil {
return err
}
pkg/sstore/sstore.go
+152 -23
View File
@@ -76,6 +76,11 @@ const (
ScreenFocusCmdFg = "cmd-fg"
)
const (
CmdStoreTypeSession = "session"
CmdStoreTypeScreen = "screen"
)
const MaxTzNameLen = 50
var globalDBLock = &sync.Mutex{}
@@ -187,17 +192,26 @@ type FeOptsType struct {
TermFontSize int `json:"termfontsize,omitempty"`
}
type ClientMigrationData struct {
MigrationType string `json:"migrationtype"`
MigrationPos int `json:"migrationpos"`
MigrationTotal int `json:"migrationtotal"`
MigrationDone bool `json:"migrationdone"`
}
type ClientData struct {
ClientId string `json:"clientid"`
UserId string `json:"userid"`
UserPrivateKeyBytes []byte `json:"-"`
UserPublicKeyBytes []byte `json:"-"`
UserPrivateKey *ecdsa.PrivateKey `json:"-" dbmap:"-"`
UserPublicKey *ecdsa.PublicKey `json:"-" dbmap:"-"`
ActiveSessionId string `json:"activesessionid"`
WinSize ClientWinSizeType `json:"winsize"`
ClientOpts ClientOptsType `json:"clientopts"`
FeOpts FeOptsType `json:"feopts"`
ClientId string `json:"clientid"`
UserId string `json:"userid"`
UserPrivateKeyBytes []byte `json:"-"`
UserPublicKeyBytes []byte `json:"-"`
UserPrivateKey *ecdsa.PrivateKey `json:"-" dbmap:"-"`
UserPublicKey *ecdsa.PublicKey `json:"-" dbmap:"-"`
ActiveSessionId string `json:"activesessionid"`
WinSize ClientWinSizeType `json:"winsize"`
ClientOpts ClientOptsType `json:"clientopts"`
FeOpts FeOptsType `json:"feopts"`
CmdStoreType string `json:"cmdstoretype"`
Migration *ClientMigrationData `json:"migration,omitempty" dbmap:"-"`
}
func (ClientData) UseDBMap() {}
@@ -641,7 +655,6 @@ func (ri *RemoteInstance) ToMap() map[string]interface{} {
}
type LineType struct {
SessionId string `json:"sessionid"`
ScreenId string `json:"screenid"`
UserId string `json:"userid"`
LineId string `json:"lineid"`
@@ -657,7 +670,6 @@ type LineType struct {
ContentHeight int64 `json:"contentheight,omitempty"`
Star bool `json:"star,omitempty"`
Bookmarked bool `json:"bookmarked,omitempty"`
Pinned bool `json:"pinned,omitempty"`
Archived bool `json:"archived,omitempty"`
Remove bool `json:"remove,omitempty"`
}
@@ -832,12 +844,18 @@ type CmdDoneInfo struct {
DurationMs int64 `json:"durationms"`
}
type CmdMapType struct {
SessionId string `json:"sessionid"`
ScreenId string `json:"screenid"`
CmdId string `json:"cmdid"`
}
type CmdType struct {
SessionId string `json:"sessionid"`
ScreenId string `json:"screenid"`
CmdId string `json:"cmdid"`
Remote RemotePtrType `json:"remote"`
CmdStr string `json:"cmdstr"`
RawCmdStr string `json:"rawcmdstr"`
FeState FeStateType `json:"festate"`
StatePtr ShellStatePtr `json:"state"`
TermOpts TermOpts `json:"termopts"`
@@ -894,13 +912,13 @@ func (r *RemoteType) FromMap(m map[string]interface{}) bool {
func (cmd *CmdType) ToMap() map[string]interface{} {
rtn := make(map[string]interface{})
rtn["sessionid"] = cmd.SessionId
rtn["screenid"] = cmd.ScreenId
rtn["cmdid"] = cmd.CmdId
rtn["remoteownerid"] = cmd.Remote.OwnerId
rtn["remoteid"] = cmd.Remote.RemoteId
rtn["remotename"] = cmd.Remote.Name
rtn["cmdstr"] = cmd.CmdStr
rtn["rawcmdstr"] = cmd.RawCmdStr
rtn["festate"] = quickJson(cmd.FeState)
rtn["statebasehash"] = cmd.StatePtr.BaseHash
rtn["statediffhasharr"] = quickJsonArr(cmd.StatePtr.DiffHashArr)
@@ -917,13 +935,13 @@ func (cmd *CmdType) ToMap() map[string]interface{} {
}
func (cmd *CmdType) FromMap(m map[string]interface{}) bool {
quickSetStr(&cmd.SessionId, m, "sessionid")
quickSetStr(&cmd.ScreenId, m, "screenid")
quickSetStr(&cmd.CmdId, m, "cmdid")
quickSetStr(&cmd.Remote.OwnerId, m, "remoteownerid")
quickSetStr(&cmd.Remote.RemoteId, m, "remoteid")
quickSetStr(&cmd.Remote.Name, m, "remotename")
quickSetStr(&cmd.CmdStr, m, "cmdstr")
quickSetStr(&cmd.RawCmdStr, m, "rawcmdstr")
quickSetJson(&cmd.FeState, m, "festate")
quickSetStr(&cmd.StatePtr.BaseHash, m, "statebasehash")
quickSetJsonArr(&cmd.StatePtr.DiffHashArr, m, "statediffhasharr")
@@ -939,9 +957,8 @@ func (cmd *CmdType) FromMap(m map[string]interface{}) bool {
return true
}
func makeNewLineCmd(sessionId string, screenId string, userId string, cmdId string, renderer string) *LineType {
func makeNewLineCmd(screenId string, userId string, cmdId string, renderer string) *LineType {
rtn := &LineType{}
rtn.SessionId = sessionId
rtn.ScreenId = screenId
rtn.UserId = userId
rtn.LineId = scbase.GenPromptUUID()
@@ -954,9 +971,8 @@ func makeNewLineCmd(sessionId string, screenId string, userId string, cmdId stri
return rtn
}
func makeNewLineText(sessionId string, screenId string, userId string, text string) *LineType {
func makeNewLineText(screenId string, userId string, text string) *LineType {
rtn := &LineType{}
rtn.SessionId = sessionId
rtn.ScreenId = screenId
rtn.UserId = userId
rtn.LineId = scbase.GenPromptUUID()
@@ -968,8 +984,8 @@ func makeNewLineText(sessionId string, screenId string, userId string, text stri
return rtn
}
func AddCommentLine(ctx context.Context, sessionId string, screenId string, userId string, commentText string) (*LineType, error) {
rtnLine := makeNewLineText(sessionId, screenId, userId, commentText)
func AddCommentLine(ctx context.Context, screenId string, userId string, commentText string) (*LineType, error) {
rtnLine := makeNewLineText(screenId, userId, commentText)
err := InsertLine(ctx, rtnLine, nil)
if err != nil {
return nil, err
@@ -977,8 +993,8 @@ func AddCommentLine(ctx context.Context, sessionId string, screenId string, user
return rtnLine, nil
}
func AddCmdLine(ctx context.Context, sessionId string, screenId string, userId string, cmd *CmdType, renderer string) (*LineType, error) {
rtnLine := makeNewLineCmd(sessionId, screenId, userId, cmd.CmdId, renderer)
func AddCmdLine(ctx context.Context, screenId string, userId string, cmd *CmdType, renderer string) (*LineType, error) {
rtnLine := makeNewLineCmd(screenId, userId, cmd.CmdId, renderer)
err := InsertLine(ctx, rtnLine, cmd)
if err != nil {
return nil, err
@@ -1085,6 +1101,7 @@ func createClientData(tx *TxWrap) error {
UserPublicKeyBytes: pubBytes,
ActiveSessionId: "",
WinSize: ClientWinSizeType{},
CmdStoreType: CmdStoreTypeScreen,
}
query := `INSERT INTO client ( clientid, userid, activesessionid, userpublickeybytes, userprivatekeybytes, winsize)
VALUES (:clientid,:userid,:activesessionid,:userpublickeybytes,:userprivatekeybytes,:winsize)`
@@ -1137,6 +1154,28 @@ func EnsureClientData(ctx context.Context) (*ClientData, error) {
return rtn, nil
}
func GetCmdMigrationInfo(ctx context.Context) (*ClientMigrationData, error) {
return WithTxRtn(ctx, func(tx *TxWrap) (*ClientMigrationData, error) {
cdata := GetMappable[*ClientData](tx, `SELECT * FROM client`)
if cdata == nil {
return nil, fmt.Errorf("no client data found")
}
if cdata.CmdStoreType == "session" {
total := tx.GetInt(`SELECT count(*) FROM cmd`)
posInv := tx.GetInt(`SELECT count(*) FROM cmd_migrate`)
mdata := &ClientMigrationData{
MigrationType: "cmdscreen",
MigrationPos: total - posInv,
MigrationTotal: total,
MigrationDone: false,
}
return mdata, nil
}
// no migration info
return nil, nil
})
}
func SetClientOpts(ctx context.Context, clientOpts ClientOptsType) error {
txErr := WithTx(ctx, func(tx *TxWrap) error {
query := `UPDATE client SET clientopts = ?`
@@ -1145,3 +1184,93 @@ func SetClientOpts(ctx context.Context, clientOpts ClientOptsType) error {
})
return txErr
}
type cmdMigrationType struct {
SessionId string
ScreenId string
CmdId string
}
func getSliceChunk[T any](slice []T, chunkSize int) ([]T, []T) {
if chunkSize >= len(slice) {
return slice, nil
}
return slice[0:chunkSize], slice[chunkSize:]
}
func processChunk(ctx context.Context, mchunk []cmdMigrationType) error {
for _, mig := range mchunk {
newFile, err := scbase.PtyOutFile(mig.ScreenId, mig.CmdId)
if err != nil {
log.Printf("ptyoutfile error: %v\n", err)
continue
}
oldFile, err := scbase.PtyOutFile_Sessions(mig.SessionId, mig.CmdId)
if err != nil {
log.Printf("ptyoutfile_sessions 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 RunCmdScreenMigration() {
ctx := context.Background()
startTime := time.Now()
mdata, err := GetCmdMigrationInfo(ctx)
if err != nil {
log.Printf("[prompt] error trying to run cmd migration: %v\n", err)
return
}
if mdata == nil || mdata.MigrationType != "cmdscreen" {
return
}
var migrations []cmdMigrationType
txErr := WithTx(ctx, func(tx *TxWrap) error {
tx.Select(&migrations, `SELECT * FROM cmd_migrate`)
return nil
})
if txErr != nil {
log.Printf("[prompt] error trying to get cmd migrations: %v\n", txErr)
return
}
log.Printf("[db] got %d cmd migrations\n", len(migrations))
for len(migrations) > 0 {
var mchunk []cmdMigrationType
mchunk, migrations = getSliceChunk(migrations, 5)
err = processChunk(ctx, mchunk)
if err != nil {
log.Printf("[prompt] cmd migration failed on chunk: %v\n%#v\n", err, mchunk)
return
}
}
err = os.RemoveAll(scbase.GetSessionsDir())
if err != nil {
log.Printf("[db] cannot remove old sessions dir %s: %v\n", scbase.GetSessionsDir(), err)
}
txErr = WithTx(ctx, func(tx *TxWrap) error {
query := `UPDATE client SET cmdstoretype = 'screen'`
tx.Exec(query)
return nil
})
if txErr != nil {
log.Printf("[db] cannot change client cmdstoretype: %v\n", err)
}
log.Printf("[db] cmd screen migration done: %v\n", time.Since(startTime))
return
}
pkg/sstore/updatebus.go
+26 -14
View File
@@ -17,7 +17,6 @@ type UpdatePacket interface {
}
type PtyDataUpdate struct {
SessionId string `json:"sessionid,omitempty"`
ScreenId string `json:"screenid,omitempty"`
CmdId string `json:"cmdid,omitempty"`
RemoteId string `json:"remoteid,omitempty"`
@@ -123,16 +122,16 @@ type CmdLineType struct {
}
type UpdateChannel struct {
SessionId string
ClientId string
Ch chan interface{}
ScreenId string
ClientId string
Ch chan interface{}
}
func (uch UpdateChannel) Match(sessionId string) bool {
if sessionId == "" {
func (uch UpdateChannel) Match(screenId string) bool {
if screenId == "" {
return true
}
return sessionId == uch.SessionId
return screenId == uch.ScreenId
}
type UpdateBus struct {
@@ -148,19 +147,19 @@ func MakeUpdateBus() *UpdateBus {
}
// always returns a new channel
func (bus *UpdateBus) RegisterChannel(clientId string, sessionId string) chan interface{} {
func (bus *UpdateBus) RegisterChannel(clientId string, screenId string) chan interface{} {
bus.Lock.Lock()
defer bus.Lock.Unlock()
uch, found := bus.Channels[clientId]
if found {
close(uch.Ch)
uch.SessionId = sessionId
uch.ScreenId = screenId
uch.Ch = make(chan interface{}, UpdateChSize)
} else {
uch = UpdateChannel{
ClientId: clientId,
SessionId: sessionId,
Ch: make(chan interface{}, UpdateChSize),
ClientId: clientId,
ScreenId: screenId,
Ch: make(chan interface{}, UpdateChSize),
}
}
bus.Channels[clientId] = uch
@@ -177,11 +176,24 @@ func (bus *UpdateBus) UnregisterChannel(clientId string) {
}
}
func (bus *UpdateBus) SendUpdate(sessionId string, update interface{}) {
func (bus *UpdateBus) SendUpdate(update interface{}) {
bus.Lock.Lock()
defer bus.Lock.Unlock()
for _, uch := range bus.Channels {
if uch.Match(sessionId) {
select {
case uch.Ch <- update:
default:
log.Printf("[error] dropped update on updatebus uch clientid=%s\n", uch.ClientId)
}
}
}
func (bus *UpdateBus) SendScreenUpdate(screenId string, update interface{}) {
bus.Lock.Lock()
defer bus.Lock.Unlock()
for _, uch := range bus.Channels {
if uch.Match(screenId) {
select {
case uch.Ch <- update: