diff --git a/pkg/cmdrunner/cmdrunner.go b/pkg/cmdrunner/cmdrunner.go index 9ee78210..af5029f4 100644 --- a/pkg/cmdrunner/cmdrunner.go +++ b/pkg/cmdrunner/cmdrunner.go @@ -1657,7 +1657,6 @@ func ScreenWebShareCommand(ctx context.Context, pk *scpacket.FeCommandPacketType if err != nil { return nil, fmt.Errorf("cannot web-share screen: %v", err) } - pcloud.NotifyUpdateWriter() infoMsg = fmt.Sprintf("screen is now shared to the web at %s", "[screen-share-url]") } else { err = sstore.ScreenWebShareStop(ctx, ids.ScreenId) diff --git a/pkg/pcloud/pcloud.go b/pkg/pcloud/pcloud.go index 2830feba..dca98837 100644 --- a/pkg/pcloud/pcloud.go +++ b/pkg/pcloud/pcloud.go @@ -30,11 +30,9 @@ const PCloudDefaultTimeout = 5 * time.Second const TelemetryUrl = "/telemetry" const NoTelemetryUrl = "/no-telemetry" -const CreateWebScreenUrl = "/auth/create-web-screen" const WebShareUpdateUrl = "/auth/web-share-update" -var updateWriterCVar = sync.NewCond(&sync.Mutex{}) -var updateWriterMoreData = false +var updateWriterLock = &sync.Mutex{} var updateWriterRunning = false type AuthInfo struct { @@ -101,7 +99,7 @@ func makeAnonPostReq(ctx context.Context, apiUrl string, data interface{}) (*htt func doRequest(req *http.Request, outputObj interface{}) (*http.Response, error) { apiUrl := req.Header.Get("X-PromptAPIUrl") - log.Printf("sending request %v\n", req.URL) + log.Printf("[pcloud] sending request %v\n", req.URL) resp, err := http.DefaultClient.Do(req) if err != nil { return nil, fmt.Errorf("error contacting pcloud %q service: %v", apiUrl, err) @@ -138,7 +136,7 @@ func SendTelemetry(ctx context.Context, force bool) error { if len(activity) == 0 { return nil } - log.Printf("sending telemetry data\n") + log.Printf("[pcloud] sending telemetry data\n") dayStr := sstore.GetCurDayStr() input := TelemetryInputType{UserId: clientData.UserId, ClientId: clientData.ClientId, CurDay: dayStr, Activity: activity} req, err := makeAnonPostReq(ctx, TelemetryUrl, input) @@ -369,49 +367,21 @@ func DoWebScreenUpdates(authInfo AuthInfo, updateArr []*sstore.ScreenUpdateType) return nil } -func CreateWebScreen(ctx context.Context, screen *WebShareScreenType) error { - authInfo, err := getAuthInfo(ctx) - if err != nil { - return err - } - req, err := makeAuthPostReq(ctx, CreateWebScreenUrl, authInfo, screen) - if err != nil { - return err - } - _, err = doRequest(req, nil) - if err != nil { - return err - } - return nil -} - -func updateWriterCheckMoreData() { - updateWriterCVar.L.Lock() - defer updateWriterCVar.L.Unlock() - for { - if updateWriterMoreData { - updateWriterMoreData = false - break - } - updateWriterCVar.Wait() - } -} - func setUpdateWriterRunning(running bool) { - updateWriterCVar.L.Lock() - defer updateWriterCVar.L.Unlock() + updateWriterLock.Lock() + defer updateWriterLock.Unlock() updateWriterRunning = running } func GetUpdateWriterRunning() bool { - updateWriterCVar.L.Lock() - defer updateWriterCVar.L.Unlock() + updateWriterLock.Lock() + defer updateWriterLock.Unlock() return updateWriterRunning } func StartUpdateWriter() { - updateWriterCVar.L.Lock() - defer updateWriterCVar.L.Unlock() + updateWriterLock.Lock() + defer updateWriterLock.Unlock() if updateWriterRunning { return } @@ -449,6 +419,7 @@ func runWebShareUpdateWriter() { numErrors := 0 numSendErrors := 0 for { + time.Sleep(100 * time.Millisecond) updateArr, err := sstore.GetScreenUpdates(context.Background(), MaxUpdatesPerReq) if err != nil { log.Printf("[pcloud] error retrieving updates: %v", err) @@ -460,7 +431,8 @@ func runWebShareUpdateWriter() { } } if len(updateArr) == 0 { - updateWriterCheckMoreData() + sstore.UpdateWriterCheckMoreData() + continue } numErrors = 0 authInfo, err := getAuthInfo(context.Background()) @@ -472,13 +444,7 @@ func runWebShareUpdateWriter() { time.Sleep(backoffTime) continue } + log.Printf("[pcloud] sent %d web-updates\n", len(updateArr)) numSendErrors = 0 } } - -func NotifyUpdateWriter() { - updateWriterCVar.L.Lock() - defer updateWriterCVar.L.Unlock() - updateWriterMoreData = true - updateWriterCVar.Signal() -} diff --git a/pkg/sstore/dbops.go b/pkg/sstore/dbops.go index bb4b4344..852a6ef0 100644 --- a/pkg/sstore/dbops.go +++ b/pkg/sstore/dbops.go @@ -21,6 +21,9 @@ import ( 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 DefaultMaxHistoryItems = 1000 +var updateWriterCVar = sync.NewCond(&sync.Mutex{}) +var updateWriterMoreData = false + type SingleConnDBGetter struct { SingleConnLock *sync.Mutex } @@ -50,6 +53,25 @@ func WithTx(ctx context.Context, fn func(tx *TxWrap) error) error { return txwrap.DBGWithTx(ctx, dbWrap, fn) } +func NotifyUpdateWriter() { + updateWriterCVar.L.Lock() + defer updateWriterCVar.L.Unlock() + updateWriterMoreData = true + updateWriterCVar.Signal() +} + +func UpdateWriterCheckMoreData() { + updateWriterCVar.L.Lock() + defer updateWriterCVar.L.Unlock() + for { + if updateWriterMoreData { + updateWriterMoreData = false + break + } + updateWriterCVar.Wait() + } +} + func NumSessions(ctx context.Context) (int, error) { var numSessions int txErr := WithTx(ctx, func(tx *TxWrap) error { @@ -2383,6 +2405,7 @@ func ScreenWebShareStart(ctx context.Context, screenId string, shareOpts ScreenW query = `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets) SELECT screenid, lineid, ?, ? FROM line WHERE screenid = ? ORDER BY linenum` tx.Exec(query, UpdateType_LineNew, time.Now().UnixMilli(), screenId) + NotifyUpdateWriter() return nil }) } @@ -2417,6 +2440,7 @@ func insertScreenUpdate(tx *TxWrap, screenId string, updateType string) { } query := `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets) VALUES (?, ?, ?, ?)` tx.Exec(query, screenId, "", updateType, time.Now().UnixMilli()) + NotifyUpdateWriter() } func insertScreenLineUpdate(tx *TxWrap, screenId string, lineId string, updateType string) { @@ -2428,8 +2452,12 @@ func insertScreenLineUpdate(tx *TxWrap, screenId string, lineId string, updateTy tx.SetErr(errors.New("invalid screen-update, lineid is empty")) return } - query := `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets) VALUES (?, ?, ?, ?)` - tx.Exec(query, screenId, lineId, updateType, time.Now().UnixMilli()) + query := `SELECT updateid FROM screenupdate WHERE screenid = ? AND lineid = ? AND (updatetype = ? OR updatetype = ?)` + if !tx.Exists(query, screenId, lineId, updateType, UpdateType_LineNew) { + query = `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets) VALUES (?, ?, ?, ?)` + tx.Exec(query, screenId, lineId, updateType, time.Now().UnixMilli()) + NotifyUpdateWriter() + } } func insertScreenCmdUpdate(tx *TxWrap, screenId string, cmdId string, updateType string) { @@ -2474,7 +2502,9 @@ func InsertPtyPosUpdate(ctx context.Context, screenId string, cmdId string) erro } query = `SELECT updateid FROM screenupdate WHERE screenid = ? AND lineid = ? AND updatetype = ?` if !tx.Exists(query, screenId, lineId, UpdateType_PtyPos) { - insertScreenLineUpdate(tx, screenId, lineId, UpdateType_PtyPos) + query := `INSERT INTO screenupdate (screenid, lineid, updatetype, updatets) VALUES (?, ?, ?, ?)` + tx.Exec(query, screenId, lineId, UpdateType_PtyPos, time.Now().UnixMilli()) + NotifyUpdateWriter() } return nil })