diff --git a/frontend/app/app.tsx b/frontend/app/app.tsx index eb8d5a8a..67681faa 100644 --- a/frontend/app/app.tsx +++ b/frontend/app/app.tsx @@ -9,6 +9,7 @@ import { Workspace } from "@/app/workspace/workspace"; import { globalStore, atoms } from "@/store/global"; import "../../public/style.less"; +import { CenteredDiv } from "./element/quickelems"; const App = () => { return ( @@ -19,16 +20,16 @@ const App = () => { }; const AppInner = () => { - const client = jotai.useAtomValue(atoms.clientAtom); - const windowData = jotai.useAtomValue(atoms.windowData); + const client = jotai.useAtomValue(atoms.client); + const windowData = jotai.useAtomValue(atoms.waveWindow); if (client == null || windowData == null) { return (
-
invalid configuration, client or window was not loaded
+
+ invalid configuration, client or window was not loaded
); } - return (
diff --git a/frontend/app/store/global.ts b/frontend/app/store/global.ts index eed83efe..cf2dc10d 100644 --- a/frontend/app/store/global.ts +++ b/frontend/app/store/global.ts @@ -1,54 +1,66 @@ // Copyright 2024, Command Line Inc. // SPDX-License-Identifier: Apache-2.0 -import * as React from "react"; import * as jotai from "jotai"; -import * as jotaiUtils from "jotai/utils"; -import { v4 as uuidv4 } from "uuid"; import * as rxjs from "rxjs"; -import type { WailsEvent } from "@wailsio/runtime/types/events"; import { Events } from "@wailsio/runtime"; import { produce } from "immer"; import { BlockService } from "@/bindings/blockservice"; -import { ObjectService } from "@/bindings/objectservice"; import * as wstore from "@/gopkg/wstore"; -import { Call as $Call } from "@wailsio/runtime"; +import * as WOS from "./wos"; const globalStore = jotai.createStore(); const blockDataMap = new Map>(); +const urlParams = new URLSearchParams(window.location.search); +const globalWindowId = urlParams.get("windowid"); +const globalClientId = urlParams.get("clientid"); +const windowIdAtom = jotai.atom(null) as jotai.PrimitiveAtom; +const clientIdAtom = jotai.atom(null) as jotai.PrimitiveAtom; +globalStore.set(windowIdAtom, globalWindowId); +globalStore.set(clientIdAtom, globalClientId); +const uiContextAtom = jotai.atom((get) => { + const uiContext: UIContext = { + windowid: get(atoms.windowId), + }; + return uiContext; +}) as jotai.Atom; +const clientAtom: jotai.Atom = jotai.atom((get) => { + const clientId = get(clientIdAtom); + if (clientId == null) { + return null; + } + return WOS.getStaticObjectValue(WOS.makeORef("client", clientId), get); +}); +const windowDataAtom: jotai.Atom = jotai.atom((get) => { + const windowId = get(windowIdAtom); + if (windowId == null) { + return null; + } + return WOS.getStaticObjectValue(WOS.makeORef("window", windowId), get); +}); +const workspaceAtom: jotai.Atom = jotai.atom((get) => { + const windowData = get(windowDataAtom); + if (windowData == null) { + return null; + } + return WOS.getStaticObjectValue(WOS.makeORef("workspace", windowData.workspaceid), get); +}); const atoms = { - blockDataMap: blockDataMap, - clientAtom: jotai.atom(null) as jotai.PrimitiveAtom, - // initialized in wave.ts (will not be null inside of application) - windowId: jotai.atom(null) as jotai.PrimitiveAtom, - windowData: jotai.atom(null) as jotai.PrimitiveAtom, + windowId: windowIdAtom, + clientId: clientIdAtom, + uiContext: uiContextAtom, + client: clientAtom, + waveWindow: windowDataAtom, + workspace: workspaceAtom, + blockDataMap: blockDataMap, }; type SubjectWithRef = rxjs.Subject & { refCount: number; release: () => void }; const blockSubjects = new Map>(); -function isBlank(str: string): boolean { - return str == null || str == ""; -} - -function makeORef(otype: string, oid: string): string { - if (isBlank(otype) || isBlank(oid)) { - return null; - } - return `${otype}:${oid}`; -} - -function splitORef(oref: string): [string, string] { - let parts = oref.split(":"); - if (parts.length != 2) { - throw new Error("invalid oref"); - } - return [parts[0], parts[1]]; -} - function getBlockSubject(blockId: string): SubjectWithRef { let subject = blockSubjects.get(blockId); if (subject == null) { @@ -95,94 +107,4 @@ function useBlockAtom(blockId: string, name: string, makeFn: () => jotai.Atom return atom as jotai.Atom; } -function GetObject(oref: string): Promise { - let prtn = $Call.ByName( - "github.com/wavetermdev/thenextwave/pkg/service/objectservice.ObjectService.GetObject", - oref - ); - return prtn; -} - -function GetClientObject(): Promise { - let prtn = $Call.ByName( - "github.com/wavetermdev/thenextwave/pkg/service/objectservice.ObjectService.GetClientObject" - ); - return prtn; -} - -type WaveObjectValue = { - pendingPromise: Promise; - dataAtom: jotai.PrimitiveAtom<{ value: T; loading: boolean }>; -}; - -const waveObjectValueCache = new Map>(); - -function clearWaveObjectCache() { - waveObjectValueCache.clear(); -} - -function createWaveValueObject(oref: string): WaveObjectValue { - const wov = { pendingPromise: null, dataAtom: null }; - wov.dataAtom = jotai.atom({ value: null, loading: true }); - let startTs = Date.now(); - let localPromise = GetObject(oref); - wov.pendingPromise = localPromise; - localPromise.then((val) => { - if (wov.pendingPromise != localPromise) { - return; - } - const [otype, oid] = splitORef(oref); - if (val != null) { - if (val["otype"] != otype) { - throw new Error("GetObject returned wrong type"); - } - if (val["oid"] != oid) { - throw new Error("GetObject returned wrong id"); - } - } - wov.pendingPromise = null; - globalStore.set(wov.dataAtom, { value: val, loading: false }); - console.log("GetObject resolved", oref, val, Date.now() - startTs + "ms"); - }); - return wov; -} - -function useWaveObjectValue(oref: string): [T, boolean] { - console.log("useWaveObjectValue", oref); - let wov = waveObjectValueCache.get(oref); - if (wov == null) { - console.log("creating new wov", oref); - wov = createWaveValueObject(oref); - waveObjectValueCache.set(oref, wov); - } - const atomVal = jotai.useAtomValue(wov.dataAtom); - return [atomVal.value, atomVal.loading]; -} - -function useWaveObject(oref: string): [T, boolean, (T) => void] { - console.log("useWaveObject", oref); - let wov = waveObjectValueCache.get(oref); - if (wov == null) { - wov = createWaveValueObject(oref); - waveObjectValueCache.set(oref, wov); - } - const [atomVal, setAtomVal] = jotai.useAtom(wov.dataAtom); - const simpleSet = (val: T) => { - setAtomVal({ value: val, loading: false }); - }; - return [atomVal.value, atomVal.loading, simpleSet]; -} - -export { - globalStore, - makeORef, - atoms, - getBlockSubject, - blockDataMap, - useBlockAtom, - GetObject, - GetClientObject, - useWaveObject, - useWaveObjectValue, - clearWaveObjectCache, -}; +export { globalStore, atoms, getBlockSubject, blockDataMap, useBlockAtom, WOS }; diff --git a/frontend/app/store/wos.ts b/frontend/app/store/wos.ts new file mode 100644 index 00000000..bd845f60 --- /dev/null +++ b/frontend/app/store/wos.ts @@ -0,0 +1,251 @@ +// Copyright 2024, Command Line Inc. +// SPDX-License-Identifier: Apache-2.0 + +// WaveObjectStore + +import * as React from "react"; +import * as jotai from "jotai"; +import { Events } from "@wailsio/runtime"; +import { Call as $Call } from "@wailsio/runtime"; +import { globalStore, atoms } from "./global"; + +type WaveObjectDataItemType = { + value: T; + loading: boolean; +}; + +type WaveObjectValue = { + pendingPromise: Promise; + dataAtom: jotai.PrimitiveAtom>; + refCount: number; + holdTime: number; +}; + +function splitORef(oref: string): [string, string] { + let parts = oref.split(":"); + if (parts.length != 2) { + throw new Error("invalid oref"); + } + return [parts[0], parts[1]]; +} + +function isBlank(str: string): boolean { + return str == null || str == ""; +} + +function isBlankNum(num: number): boolean { + return num == null || isNaN(num) || num == 0; +} + +function isValidWaveObj(val: WaveObj): boolean { + if (val == null) { + return false; + } + if (isBlank(val.otype) || isBlank(val.oid)) { + return false; + } + if (!val.deleted && isBlankNum(val.version)) { + return false; + } + return true; +} + +function makeORef(otype: string, oid: string): string { + if (isBlank(otype) || isBlank(oid)) { + return null; + } + return `${otype}:${oid}`; +} + +function GetObject(oref: string): Promise { + let prtn = $Call.ByName( + "github.com/wavetermdev/thenextwave/pkg/service/objectservice.ObjectService.GetObject", + oref + ); + return prtn; +} + +const waveObjectValueCache = new Map>(); + +function clearWaveObjectCache() { + waveObjectValueCache.clear(); +} + +const defaultHoldTime = 5000; // 5-seconds + +function createWaveValueObject(oref: string, shouldFetch: boolean): WaveObjectValue { + const wov = { pendingPromise: null, dataAtom: null, refCount: 0, holdTime: Date.now() + 5000 }; + wov.dataAtom = jotai.atom({ value: null, loading: true }); + if (!shouldFetch) { + return wov; + } + let startTs = Date.now(); + let localPromise = GetObject(oref); + wov.pendingPromise = localPromise; + localPromise.then((val) => { + if (wov.pendingPromise != localPromise) { + return; + } + const [otype, oid] = splitORef(oref); + if (val != null) { + if (val["otype"] != otype) { + throw new Error("GetObject returned wrong type"); + } + if (val["oid"] != oid) { + throw new Error("GetObject returned wrong id"); + } + } + wov.pendingPromise = null; + globalStore.set(wov.dataAtom, { value: val, loading: false }); + console.log("GetObject resolved", oref, Date.now() - startTs + "ms"); + }); + return wov; +} + +function loadAndPinWaveObject(oref: string): Promise { + let wov = waveObjectValueCache.get(oref); + if (wov == null) { + wov = createWaveValueObject(oref, true); + waveObjectValueCache.set(oref, wov); + } + wov.refCount++; + if (wov.pendingPromise == null) { + const dataValue = globalStore.get(wov.dataAtom); + return Promise.resolve(dataValue.value); + } + return wov.pendingPromise; +} + +function useWaveObjectValue(oref: string): [T, boolean] { + let wov = waveObjectValueCache.get(oref); + if (wov == null) { + wov = createWaveValueObject(oref, true); + waveObjectValueCache.set(oref, wov); + } + React.useEffect(() => { + wov.refCount++; + return () => { + wov.refCount--; + }; + }, [oref]); + const atomVal = jotai.useAtomValue(wov.dataAtom); + return [atomVal.value, atomVal.loading]; +} + +function useWaveObject(oref: string): [T, boolean, (T) => void] { + let wov = waveObjectValueCache.get(oref); + if (wov == null) { + wov = createWaveValueObject(oref, true); + waveObjectValueCache.set(oref, wov); + } + React.useEffect(() => { + wov.refCount++; + return () => { + wov.refCount--; + }; + }, [oref]); + const [atomVal, setAtomVal] = jotai.useAtom(wov.dataAtom); + const simpleSet = (val: T) => { + setAtomVal({ value: val, loading: false }); + }; + return [atomVal.value, atomVal.loading, simpleSet]; +} + +function updateWaveObject(val: WaveObj) { + if (val == null) { + return; + } + if (!isValidWaveObj(val)) { + console.log("invalid wave object", val); + return; + } + let oref = makeORef(val.otype, val.oid); + let wov = waveObjectValueCache.get(oref); + if (wov == null) { + wov = createWaveValueObject(oref, false); + waveObjectValueCache.set(oref, wov); + } + if (val.deleted) { + globalStore.set(wov.dataAtom, { value: null, loading: false }); + } else { + let curValue: WaveObjectDataItemType = globalStore.get(wov.dataAtom); + if (curValue.value != null && curValue.value.version >= val.version) { + return; + } + globalStore.set(wov.dataAtom, { value: val, loading: false }); + } + wov.holdTime = Date.now() + defaultHoldTime; + return; +} + +function updateWaveObjects(vals: WaveObj[]) { + for (let val of vals) { + updateWaveObject(val); + } +} + +function cleanWaveObjectCache() { + let now = Date.now(); + for (let [oref, wov] of waveObjectValueCache) { + if (wov.refCount == 0 && wov.holdTime < now) { + waveObjectValueCache.delete(oref); + } + } +} + +Events.On("waveobj:update", (event: any) => { + const data: WaveObj[] = event?.data; + if (data == null) { + return; + } + if (!Array.isArray(data)) { + console.log("invalid waveobj:update, not an array", data); + return; + } + if (data.length == 0) { + return; + } + updateWaveObjects(data); +}); + +function wrapObjectServiceCall(fnName: string, ...args: any[]): Promise { + const uiContext = globalStore.get(atoms.uiContext); + let prtn = $Call.ByName( + "github.com/wavetermdev/thenextwave/pkg/service/objectservice.ObjectService." + fnName, + uiContext, + ...args + ); + prtn = prtn.then((val) => { + if (val.updates) { + updateWaveObjects(val.updates); + } + return val; + }); + return prtn; +} + +function AddTabToWorkspace(tabName: string, activateTab: boolean): Promise<{ tabId: string }> { + return wrapObjectServiceCall("AddTabToWorkspace", tabName, activateTab); +} + +function getStaticObjectValue(oref: string, getFn: jotai.Getter): T { + let wov = waveObjectValueCache.get(oref); + if (wov == null) { + return null; + } + const atomVal = getFn(wov.dataAtom); + return atomVal.value; +} + +export { + makeORef, + useWaveObject, + useWaveObjectValue, + loadAndPinWaveObject, + clearWaveObjectCache, + updateWaveObject, + updateWaveObjects, + cleanWaveObjectCache, + getStaticObjectValue, + AddTabToWorkspace, +}; diff --git a/frontend/app/tab/tab.tsx b/frontend/app/tab/tab.tsx index 9799be16..97aa653f 100644 --- a/frontend/app/tab/tab.tsx +++ b/frontend/app/tab/tab.tsx @@ -5,13 +5,13 @@ import * as React from "react"; import * as jotai from "jotai"; import { Block } from "@/app/block/block"; import { atoms } from "@/store/global"; -import * as gdata from "@/store/global"; +import * as WOS from "@/store/wos"; import "./tab.less"; import { CenteredLoadingDiv } from "../element/quickelems"; const TabContent = ({ tabId }: { tabId: string }) => { - const [tabData, tabLoading] = gdata.useWaveObjectValue(gdata.makeORef("tab", tabId)); + const [tabData, tabLoading] = WOS.useWaveObjectValue(WOS.makeORef("tab", tabId)); if (tabLoading) { return ; } diff --git a/frontend/app/workspace/workspace.tsx b/frontend/app/workspace/workspace.tsx index 69f76909..b00428c9 100644 --- a/frontend/app/workspace/workspace.tsx +++ b/frontend/app/workspace/workspace.tsx @@ -12,22 +12,20 @@ import { ClientService } from "@/bindings/clientservice"; import { Workspace } from "@/gopkg/wstore"; import * as wstore from "@/gopkg/wstore"; import * as jotaiUtil from "jotai/utils"; -import * as gdata from "@/store/global"; - -import "./workspace.less"; +import * as WOS from "@/store/wos"; import { CenteredLoadingDiv, CenteredDiv } from "../element/quickelems"; -function Tab({ tabId }: { tabId: string }) { - const windowData = jotai.useAtomValue(atoms.windowData); - const [tabData, tabLoading] = gdata.useWaveObjectValue(gdata.makeORef("tab", tabId)); +import "./workspace.less"; +function Tab({ tabId }: { tabId: string }) { + const windowData = jotai.useAtomValue(atoms.waveWindow); + const [tabData, tabLoading] = WOS.useWaveObjectValue(WOS.makeORef("tab", tabId)); function setActiveTab(tabId: string) { if (tabId == null) { return; } // TODO } - return (
@@ -60,7 +55,7 @@ function TabBar({ workspace, waveWindow }: { workspace: Workspace; waveWindow: W } function Widgets() { - const windowData = jotai.useAtomValue(atoms.windowData); + const windowData = jotai.useAtomValue(atoms.waveWindow); const activeTabId = windowData.activetabid; async function createBlock(blockDef: wstore.BlockDef) { @@ -122,18 +117,14 @@ function Widgets() { } function WorkspaceElem() { - const windowData = jotai.useAtomValue(atoms.windowData); - const workspaceId = windowData?.workspaceid; + const windowData = jotai.useAtomValue(atoms.waveWindow); const activeTabId = windowData?.activetabid; - const [ws, wsLoading] = gdata.useWaveObjectValue(gdata.makeORef("workspace", workspaceId)); - if (wsLoading) { - return ; - } + const ws = jotai.useAtomValue(atoms.workspace); return (
- +
diff --git a/frontend/types/custom.d.ts b/frontend/types/custom.d.ts index aae4f6e2..da6b0c94 100644 --- a/frontend/types/custom.d.ts +++ b/frontend/types/custom.d.ts @@ -2,6 +2,10 @@ // SPDX-License-Identifier: Apache-2.0 declare global { + type UIContext = { + windowid: string; + }; + type ORef = { otype: string; oid: string; @@ -79,6 +83,8 @@ declare global { type WaveObj = { otype: string; oid: string; + version: number; + deleted?: boolean; }; type WaveWindow = { diff --git a/frontend/wave.ts b/frontend/wave.ts index c06689bc..c9d116ac 100644 --- a/frontend/wave.ts +++ b/frontend/wave.ts @@ -7,14 +7,16 @@ import { App } from "./app/app"; import { loadFonts } from "./util/fontutil"; import { ClientService } from "@/bindings/clientservice"; import { Client } from "@/gopkg/wstore"; -import { globalStore, atoms, GetClientObject, GetObject, makeORef } from "@/store/global"; +import { globalStore, atoms } from "@/store/global"; +import * as WOS from "@/store/wos"; import * as wailsRuntime from "@wailsio/runtime"; import * as wstore from "@/gopkg/wstore"; +import * as gdata from "@/store/global"; import { immerable } from "immer"; const urlParams = new URLSearchParams(window.location.search); const windowId = urlParams.get("windowid"); -globalStore.set(atoms.windowId, windowId); +const clientId = urlParams.get("clientid"); wstore.Block.prototype[immerable] = true; wstore.Tab.prototype[immerable] = true; @@ -30,10 +32,10 @@ wstore.WinSize.prototype[immerable] = true; loadFonts(); document.addEventListener("DOMContentLoaded", async () => { - const client = await GetClientObject(); - globalStore.set(atoms.clientAtom, client); - const window = await GetObject(makeORef("window", windowId)); - globalStore.set(atoms.windowData, window); + // ensures client/window are loaded into the cache before rendering + await WOS.loadAndPinWaveObject(WOS.makeORef("client", clientId)); + const waveWindow = await WOS.loadAndPinWaveObject(WOS.makeORef("window", windowId)); + await WOS.loadAndPinWaveObject(WOS.makeORef("workspace", waveWindow.workspaceid)); let reactElem = React.createElement(App, null, null); let elem = document.getElementById("main"); let root = createRoot(elem); diff --git a/main.go b/main.go index 7b976f80..007106e0 100644 --- a/main.go +++ b/main.go @@ -8,6 +8,7 @@ package main import ( "context" "embed" + "fmt" "log" "net/http" "runtime" @@ -53,6 +54,10 @@ func createAppMenu(app *application.App) *application.Menu { } func createWindow(windowData *wstore.Window, app *application.App) { + client, err := wstore.DBGetSingleton[*wstore.Client](context.Background()) + if err != nil { + panic(fmt.Errorf("error getting client data: %w", err)) + } window := app.NewWebviewWindowWithOptions(application.WebviewWindowOptions{ Title: "Wave Terminal", Mac: application.MacWindow{ @@ -61,7 +66,7 @@ func createWindow(windowData *wstore.Window, app *application.App) { TitleBar: application.MacTitleBarHiddenInset, }, BackgroundColour: application.NewRGB(0, 0, 0), - URL: "/public/index.html?windowid=" + windowData.OID, + URL: "/public/index.html?windowid=" + windowData.OID + "&clientid=" + client.OID, X: windowData.Pos.X, Y: windowData.Pos.Y, Width: windowData.WinSize.Width, diff --git a/pkg/blockcontroller/blockcontroller.go b/pkg/blockcontroller/blockcontroller.go index bb8ade23..70d03c7a 100644 --- a/pkg/blockcontroller/blockcontroller.go +++ b/pkg/blockcontroller/blockcontroller.go @@ -62,6 +62,7 @@ func jsonDeepCopy(val map[string]any) (map[string]any, error) { } func CreateBlock(ctx context.Context, bdef *wstore.BlockDef, rtOpts *wstore.RuntimeOpts) (*wstore.Block, error) { + // TODO blockId := uuid.New().String() blockData := &wstore.Block{ OID: blockId, @@ -86,13 +87,13 @@ func CreateBlock(ctx context.Context, bdef *wstore.BlockDef, rtOpts *wstore.Runt } func CloseBlock(blockId string) { + // TODO bc := GetBlockController(blockId) if bc == nil { return } bc.Close() close(bc.InputCh) - wstore.BlockMap.Delete(blockId) } func (bc *BlockController) setShellProc(shellProc *shellexec.ShellProc) error { diff --git a/pkg/service/objectservice/objectservice.go b/pkg/service/objectservice/objectservice.go index 64476bec..e0eae455 100644 --- a/pkg/service/objectservice/objectservice.go +++ b/pkg/service/objectservice/objectservice.go @@ -25,16 +25,6 @@ func parseORef(oref string) (*waveobj.ORef, error) { return &waveobj.ORef{OType: fields[0], OID: fields[1]}, nil } -func (svc *ObjectService) GetClientObject() (any, error) { - ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout) - defer cancelFn() - client, err := wstore.DBGetSingleton[*wstore.Client](ctx) - if err != nil { - return nil, fmt.Errorf("error getting client: %w", err) - } - return waveobj.ToJsonMap(client) -} - func (svc *ObjectService) GetObject(orefStr string) (any, error) { oref, err := parseORef(orefStr) if err != nil { @@ -46,7 +36,8 @@ func (svc *ObjectService) GetObject(orefStr string) (any, error) { if err != nil { return nil, fmt.Errorf("error getting object: %w", err) } - return waveobj.ToJsonMap(obj) + rtn, err := waveobj.ToJsonMap(obj) + return rtn, err } func (svc *ObjectService) GetObjects(orefStrArr []string) (any, error) { @@ -63,3 +54,60 @@ func (svc *ObjectService) GetObjects(orefStrArr []string) (any, error) { } return wstore.DBSelectORefs(ctx, orefArr) } + +func updatesRtn(ctx context.Context, rtnVal map[string]any) (any, error) { + updates := wstore.ContextGetUpdates(ctx) + if len(updates) == 0 { + return nil, nil + } + var rtn []any + for _, obj := range updates { + if obj == nil { + continue + } + jmap, err := waveobj.ToJsonMap(obj) + if err != nil { + return nil, fmt.Errorf("error converting object to JSON: %w", err) + } + rtn = append(rtn, jmap) + } + if rtnVal == nil { + rtnVal = make(map[string]any) + } + rtnVal["updates"] = rtn + return rtnVal, nil +} + +func (svc *ObjectService) AddTabToWorkspace(uiContext wstore.UIContext, tabName string, activateTab bool) (any, error) { + ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout) + defer cancelFn() + ctx = wstore.ContextWithUpdates(ctx) + windowData, err := wstore.DBMustGet[*wstore.Window](ctx, uiContext.WindowId) + if err != nil { + return nil, fmt.Errorf("error getting window: %w", err) + } + tab, err := wstore.CreateTab(ctx, windowData.WorkspaceId, tabName) + if err != nil { + return nil, fmt.Errorf("error creating tab: %w", err) + } + if activateTab { + err = wstore.SetActiveTab(ctx, uiContext.WindowId, tab.OID) + if err != nil { + return nil, fmt.Errorf("error setting active tab: %w", err) + } + } + rtn := make(map[string]any) + rtn["tabid"] = waveobj.GetOID(tab) + return updatesRtn(ctx, rtn) +} + +func (svc *ObjectService) SetActiveTab(uiContext wstore.UIContext, tabId string) (any, error) { + ctx, cancelFn := context.WithTimeout(context.Background(), DefaultTimeout) + defer cancelFn() + ctx = wstore.ContextWithUpdates(ctx) + err := wstore.SetActiveTab(ctx, uiContext.WindowId, tabId) + if err != nil { + return nil, fmt.Errorf("error setting active tab: %w", err) + } + return updatesRtn(ctx, nil) +} diff --git a/pkg/waveobj/waveobj.go b/pkg/waveobj/waveobj.go index 0b965a9e..38245773 100644 --- a/pkg/waveobj/waveobj.go +++ b/pkg/waveobj/waveobj.go @@ -18,6 +18,7 @@ const ( OTypeKeyName = "otype" OIDKeyName = "oid" VersionKeyName = "version" + DeletedKeyName = "deleted" OIDGoFieldName = "OID" VersionGoFieldName = "Version" @@ -32,6 +33,15 @@ type WaveObj interface { GetOType() string // should not depend on object state (should work with nil value) } +type WaveObjTombstone struct { + OType string `json:"otype"` + OID string `json:"oid"` +} + +func (w *WaveObjTombstone) GetOType() string { + return w.OType +} + type waveObjDesc struct { RType reflect.Type OIDField reflect.StructField @@ -93,6 +103,9 @@ func getWaveObjDesc(otype string) *waveObjDesc { } func GetOID(waveObj WaveObj) string { + if tomb, ok := waveObj.(*WaveObjTombstone); ok { + return tomb.OID + } desc := getWaveObjDesc(waveObj.GetOType()) if desc == nil { return "" @@ -101,6 +114,10 @@ func GetOID(waveObj WaveObj) string { } func SetOID(waveObj WaveObj, oid string) { + if tomb, ok := waveObj.(*WaveObjTombstone); ok { + tomb.OID = oid + return + } desc := getWaveObjDesc(waveObj.GetOType()) if desc == nil { return @@ -109,6 +126,9 @@ func SetOID(waveObj WaveObj, oid string) { } func GetVersion(waveObj WaveObj) int { + if _, ok := waveObj.(*WaveObjTombstone); ok { + return 0 + } desc := getWaveObjDesc(waveObj.GetOType()) if desc == nil { return 0 @@ -117,6 +137,9 @@ func GetVersion(waveObj WaveObj) int { } func SetVersion(waveObj WaveObj, version int) { + if _, ok := waveObj.(*WaveObjTombstone); ok { + return + } desc := getWaveObjDesc(waveObj.GetOType()) if desc == nil { return @@ -138,6 +161,10 @@ func ToJsonMap(w WaveObj) (map[string]any, error) { if err != nil { return nil, err } + if _, ok := w.(*WaveObjTombstone); ok { + m[DeletedKeyName] = true + return m, nil + } m[OTypeKeyName] = w.GetOType() m[OIDKeyName] = GetOID(w) m[VersionKeyName] = GetVersion(w) @@ -152,12 +179,39 @@ func ToJson(w WaveObj) ([]byte, error) { return json.Marshal(m) } +func getMapBoolVal(m map[string]any, key string) bool { + val, ok := m[key].(bool) + if !ok { + return false + } + return val +} + +func getMapStringVal(m map[string]any, key string) string { + val, ok := m[key].(string) + if !ok { + return "" + } + return val +} + +func IsTombstone(w WaveObj) bool { + _, ok := w.(*WaveObjTombstone) + return ok +} + func FromJson(data []byte) (WaveObj, error) { var m map[string]any err := json.Unmarshal(data, &m) if err != nil { return nil, err } + if getMapBoolVal(m, DeletedKeyName) { + return &WaveObjTombstone{ + OType: getMapStringVal(m, OTypeKeyName), + OID: getMapStringVal(m, OIDKeyName), + }, nil + } otype, ok := m[OTypeKeyName].(string) if !ok { return nil, fmt.Errorf("missing otype") @@ -320,6 +374,7 @@ func GenerateWaveObjTSType() string { buf.WriteString("type WaveObj = {\n") buf.WriteString(" otype: string;\n") buf.WriteString(" oid: string;\n") + buf.WriteString(" version: number;\n") buf.WriteString("};\n") return buf.String() } diff --git a/pkg/wstore/wstore.go b/pkg/wstore/wstore.go index 163316df..8a9d7ec5 100644 --- a/pkg/wstore/wstore.go +++ b/pkg/wstore/wstore.go @@ -4,20 +4,19 @@ package wstore import ( + "bytes" "context" "fmt" + "log" "reflect" "time" "github.com/google/uuid" "github.com/wavetermdev/thenextwave/pkg/shellexec" - "github.com/wavetermdev/thenextwave/pkg/util/ds" "github.com/wavetermdev/thenextwave/pkg/waveobj" ) -var WorkspaceMap = ds.NewSyncMap[*Workspace]() -var TabMap = ds.NewSyncMap[*Tab]() -var BlockMap = ds.NewSyncMap[*Block]() +var waveObjUpdateKey = struct{}{} func init() { for _, rtype := range AllWaveObjTypes() { @@ -25,6 +24,122 @@ func init() { } } +type contextUpdatesType struct { + UpdatesStack []map[waveobj.ORef]waveobj.WaveObj +} + +func dumpUpdateStack(updates *contextUpdatesType) { + log.Printf("dumpUpdateStack len:%d\n", len(updates.UpdatesStack)) + for idx, update := range updates.UpdatesStack { + var buf bytes.Buffer + buf.WriteString(fmt.Sprintf(" [%d]:", idx)) + for k := range update { + buf.WriteString(fmt.Sprintf(" %s:%s", k.OType, k.OID)) + } + buf.WriteString("\n") + log.Print(buf.String()) + } +} + +func ContextWithUpdates(ctx context.Context) context.Context { + updatesVal := ctx.Value(waveObjUpdateKey) + if updatesVal != nil { + return ctx + } + return context.WithValue(ctx, waveObjUpdateKey, &contextUpdatesType{ + UpdatesStack: []map[waveobj.ORef]waveobj.WaveObj{make(map[waveobj.ORef]waveobj.WaveObj)}, + }) +} + +func ContextGetUpdates(ctx context.Context) map[waveobj.ORef]waveobj.WaveObj { + updatesVal := ctx.Value(waveObjUpdateKey) + if updatesVal == nil { + return nil + } + updates := updatesVal.(*contextUpdatesType) + if len(updates.UpdatesStack) == 1 { + return updates.UpdatesStack[0] + } + rtn := make(map[waveobj.ORef]waveobj.WaveObj) + for _, update := range updates.UpdatesStack { + for k, v := range update { + rtn[k] = v + } + } + return rtn +} + +func ContextGetUpdate(ctx context.Context, oref waveobj.ORef) waveobj.WaveObj { + updatesVal := ctx.Value(waveObjUpdateKey) + if updatesVal == nil { + return nil + } + updates := updatesVal.(*contextUpdatesType) + for idx := len(updates.UpdatesStack) - 1; idx >= 0; idx-- { + if obj, ok := updates.UpdatesStack[idx][oref]; ok { + return obj + } + } + return nil +} + +func ContextAddUpdate(ctx context.Context, obj waveobj.WaveObj) { + updatesVal := ctx.Value(waveObjUpdateKey) + if updatesVal == nil { + return + } + updates := updatesVal.(*contextUpdatesType) + oref := waveobj.ORef{ + OType: obj.GetOType(), + OID: waveobj.GetOID(obj), + } + updates.UpdatesStack[len(updates.UpdatesStack)-1][oref] = obj +} + +func ContextUpdatesBeginTx(ctx context.Context) context.Context { + updatesVal := ctx.Value(waveObjUpdateKey) + if updatesVal == nil { + return ctx + } + updates := updatesVal.(*contextUpdatesType) + updates.UpdatesStack = append(updates.UpdatesStack, make(map[waveobj.ORef]waveobj.WaveObj)) + return ctx +} + +func ContextUpdatesCommitTx(ctx context.Context) { + updatesVal := ctx.Value(waveObjUpdateKey) + if updatesVal == nil { + return + } + updates := updatesVal.(*contextUpdatesType) + if len(updates.UpdatesStack) <= 1 { + panic(fmt.Errorf("no updates transaction to commit")) + } + // merge the last two updates + curUpdateMap := updates.UpdatesStack[len(updates.UpdatesStack)-1] + prevUpdateMap := updates.UpdatesStack[len(updates.UpdatesStack)-2] + for k, v := range curUpdateMap { + prevUpdateMap[k] = v + } + updates.UpdatesStack = updates.UpdatesStack[:len(updates.UpdatesStack)-1] +} + +func ContextUpdatesRollbackTx(ctx context.Context) { + updatesVal := ctx.Value(waveObjUpdateKey) + if updatesVal == nil { + return + } + updates := updatesVal.(*contextUpdatesType) + if len(updates.UpdatesStack) <= 1 { + panic(fmt.Errorf("no updates transaction to rollback")) + } + updates.UpdatesStack = updates.UpdatesStack[:len(updates.UpdatesStack)-1] +} + +type UIContext struct { + WindowId string `json:"windowid"` +} + type Client struct { OID string `json:"oid"` Version int `json:"version"` @@ -128,39 +243,51 @@ func (*Block) GetOType() string { return "block" } -func CreateTab(workspaceId string, name string) (*Tab, error) { - tab := &Tab{ - OID: uuid.New().String(), - Name: name, - BlockIds: []string{}, - } - TabMap.Set(tab.OID, tab) - ws := WorkspaceMap.Get(workspaceId) - if ws == nil { - return nil, fmt.Errorf("workspace not found: %q", workspaceId) - } - ws.TabIds = append(ws.TabIds, tab.OID) - return tab, nil +func CreateTab(ctx context.Context, workspaceId string, name string) (*Tab, error) { + return WithTxRtn(ctx, func(tx *TxWrap) (*Tab, error) { + ws, _ := DBGet[*Workspace](tx.Context(), workspaceId) + if ws == nil { + return nil, fmt.Errorf("workspace not found: %q", workspaceId) + } + tab := &Tab{ + OID: uuid.New().String(), + Name: name, + BlockIds: []string{}, + } + ws.TabIds = append(ws.TabIds, tab.OID) + DBInsert(tx.Context(), tab) + DBUpdate(tx.Context(), ws) + return tab, nil + }) } -func CreateWorkspace() (*Workspace, error) { +func CreateWorkspace(ctx context.Context) (*Workspace, error) { ws := &Workspace{ OID: uuid.New().String(), TabIds: []string{}, } - WorkspaceMap.Set(ws.OID, ws) - _, err := CreateTab(ws.OID, "Tab 1") - if err != nil { - return nil, err - } + DBInsert(ctx, ws) return ws, nil } -func GetObject(otype string, oid string) (waveobj.WaveObj, error) { - return nil, nil +func SetActiveTab(ctx context.Context, windowId string, tabId string) error { + return WithTx(ctx, func(tx *TxWrap) error { + window, _ := DBGet[*Window](tx.Context(), windowId) + if window == nil { + return fmt.Errorf("window not found: %q", windowId) + } + tab, _ := DBGet[*Tab](tx.Context(), tabId) + if tab == nil { + return fmt.Errorf("tab not found: %q", tabId) + } + window.ActiveTabId = tabId + DBUpdate(tx.Context(), window) + return nil + }) } func EnsureInitialData() error { + // does not need to run in a transaction since it is called on startup ctx, cancelFn := context.WithTimeout(context.Background(), 2*time.Second) defer cancelFn() clientCount, err := DBGetCount[*Client](ctx) diff --git a/pkg/wstore/wstore_dbops.go b/pkg/wstore/wstore_dbops.go index 393ac628..ec7a6d3e 100644 --- a/pkg/wstore/wstore_dbops.go +++ b/pkg/wstore/wstore_dbops.go @@ -10,6 +10,8 @@ import ( "github.com/wavetermdev/thenextwave/pkg/waveobj" ) +var ErrNotFound = fmt.Errorf("not found") + func waveObjTableName(w waveobj.WaveObj) string { return "db_" + w.GetOType() } @@ -75,6 +77,19 @@ func DBGet[T waveobj.WaveObj](ctx context.Context, id string) (T, error) { return genericCastWithErr[T](rtn, err) } +func DBMustGet[T waveobj.WaveObj](ctx context.Context, id string) (T, error) { + rtn, err := DBGetORef(ctx, waveobj.ORef{OType: getOTypeGen[T](), OID: id}) + if err != nil { + var zeroVal T + return zeroVal, err + } + if rtn == nil { + var zeroVal T + return zeroVal, ErrNotFound + } + return rtn.(T), nil +} + func DBGetORef(ctx context.Context, oref waveobj.ORef) (waveobj.WaveObj, error) { return WithTxRtn(ctx, func(tx *TxWrap) (waveobj.WaveObj, error) { table := tableNameFromOType(oref.OType) @@ -144,11 +159,15 @@ func DBDelete(ctx context.Context, otype string, id string) error { table := tableNameFromOType(otype) query := fmt.Sprintf("DELETE FROM %s WHERE oid = ?", table) tx.Exec(query, id) + ContextAddUpdate(ctx, &waveobj.WaveObjTombstone{OType: otype, OID: id}) return nil }) } func DBUpdate(ctx context.Context, val waveobj.WaveObj) error { + if waveobj.IsTombstone(val) { + return fmt.Errorf("cannot update deleted object") + } oid := waveobj.GetOID(val) if oid == "" { return fmt.Errorf("cannot update %T value with empty id", val) @@ -159,13 +178,18 @@ func DBUpdate(ctx context.Context, val waveobj.WaveObj) error { } return WithTx(ctx, func(tx *TxWrap) error { table := waveObjTableName(val) - query := fmt.Sprintf("UPDATE %s SET data = ?, version = version+1 WHERE oid = ?", table) - tx.Exec(query, jsonData, oid) + query := fmt.Sprintf("UPDATE %s SET data = ?, version = version+1 WHERE oid = ? RETURNING version", table) + newVersion := tx.GetInt(query, jsonData, oid) + waveobj.SetVersion(val, newVersion) + ContextAddUpdate(ctx, val) return nil }) } func DBInsert(ctx context.Context, val waveobj.WaveObj) error { + if waveobj.IsTombstone(val) { + return fmt.Errorf("cannot insert deleted object") + } oid := waveobj.GetOID(val) if oid == "" { return fmt.Errorf("cannot insert %T value with empty id", val) @@ -176,8 +200,10 @@ func DBInsert(ctx context.Context, val waveobj.WaveObj) error { } return WithTx(ctx, func(tx *TxWrap) error { table := waveObjTableName(val) + waveobj.SetVersion(val, 1) query := fmt.Sprintf("INSERT INTO %s (oid, version, data) VALUES (?, ?, ?)", table) tx.Exec(query, oid, 1, jsonData) + ContextAddUpdate(ctx, val) return nil }) } diff --git a/pkg/wstore/wstore_dbsetup.go b/pkg/wstore/wstore_dbsetup.go index d1141490..f79e2cea 100644 --- a/pkg/wstore/wstore_dbsetup.go +++ b/pkg/wstore/wstore_dbsetup.go @@ -55,10 +55,26 @@ func MakeDB(ctx context.Context) (*sqlx.DB, error) { return rtn, nil } -func WithTx(ctx context.Context, fn func(tx *TxWrap) error) error { +func WithTx(ctx context.Context, fn func(tx *TxWrap) error) (rtnErr error) { + ContextUpdatesBeginTx(ctx) + defer func() { + if rtnErr != nil { + ContextUpdatesRollbackTx(ctx) + } else { + ContextUpdatesCommitTx(ctx) + } + }() return txwrap.WithTx(ctx, globalDB, fn) } -func WithTxRtn[RT any](ctx context.Context, fn func(tx *TxWrap) (RT, error)) (RT, error) { +func WithTxRtn[RT any](ctx context.Context, fn func(tx *TxWrap) (RT, error)) (rtnVal RT, rtnErr error) { + ContextUpdatesBeginTx(ctx) + defer func() { + if rtnErr != nil { + ContextUpdatesRollbackTx(ctx) + } else { + ContextUpdatesCommitTx(ctx) + } + }() return txwrap.WithTxRtn(ctx, globalDB, fn) }