feat: create custom Readable to share stores logic (#239)

* feat: create custom Writable to share stores logic

Signed-off-by: lstocchi <[email protected]>

* fix: rename test file and store to new modelsInfo

Signed-off-by: lstocchi <[email protected]>

---------

Signed-off-by: lstocchi <[email protected]>

packages/frontend/src/stores/modelsInfo.spec.ts

@@ -0,0 +1,77 @@
+/**********************************************************************
+ * Copyright (C) 2024 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ***********************************************************************/
+
+import { afterEach, beforeEach, expect, test, vi } from 'vitest';
+import { MSG_NEW_MODELS_STATE } from '@shared/Messages';
+import { rpcBrowser } from '../utils/client';
+import type { Unsubscriber } from 'svelte/store';
+import { modelsInfo } from './modelsInfo';
+
+const mocks = vi.hoisted(() => {
+  return {
+    getModelsInfoMock: vi.fn().mockResolvedValue([]),
+  };
+});
+
+vi.mock('../utils/client', async () => {
+  const subscriber = new Map();
+  const rpcBrowser = {
+    invoke: (msgId: string, _: unknown[]) => {
+      const f = subscriber.get(msgId);
+      f();
+    },
+    subscribe: (msgId: string, f: (msg: any) => void) => {
+      subscriber.set(msgId, f);
+      return {
+        unsubscribe: () => {
+          subscriber.clear();
+        },
+      };
+    },
+  };
+  return {
+    rpcBrowser,
+    studioClient: {
+      getModelsInfo: mocks.getModelsInfoMock,
+    },
+  };
+});
+
+let unsubscriber: Unsubscriber | undefined;
+beforeEach(() => {
+  vi.clearAllMocks();
+  unsubscriber = modelsInfo.subscribe(_ => {});
+});
+
+afterEach(() => {
+  if (unsubscriber) {
+    unsubscriber();
+    unsubscriber = undefined;
+  }
+});
+
+test('check getLocalModels is called at subscription', async () => {
+  expect(mocks.getModelsInfoMock).toHaveBeenCalledOnce();
+});
+
+test('check getLocalModels is called twice if event is fired (one at init, one for the event)', async () => {
+  rpcBrowser.invoke(MSG_NEW_MODELS_STATE);
+  // wait for the timeout in the debouncer
+  await new Promise(resolve => setTimeout(resolve, 600));
+  expect(mocks.getModelsInfoMock).toHaveBeenCalledTimes(2);
+});

packages/frontend/src/stores/modelsInfo.ts

@@ -1,18 +1,24 @@
+/**********************************************************************
+ * Copyright (C) 2024 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ***********************************************************************/
+
 import type { ModelInfo } from '@shared/src/models/IModelInfo';
-import type { Readable } from 'svelte/store';
-import { readable } from 'svelte/store';
-import { rpcBrowser, studioClient } from '/@/utils/client';
+import { studioClient } from '/@/utils/client';
 import { MSG_NEW_MODELS_STATE } from '@shared/Messages';
+import { RPCReadable } from './rpcReadable';
 
-export const modelsInfo: Readable<ModelInfo[]> = readable<ModelInfo[]>([], set => {
-  const sub = rpcBrowser.subscribe(MSG_NEW_MODELS_STATE, msg => {
-    set(msg);
-  });
-  // Initialize the store manually
-  studioClient.getModelsInfo().then(v => {
-    set(v);
-  });
-  return () => {
-    sub.unsubscribe();
-  };
-});
+export const modelsInfo = RPCReadable<ModelInfo[]>([], [MSG_NEW_MODELS_STATE], studioClient.getModelsInfo);

packages/frontend/src/stores/rpcReadable.spec.ts

@@ -0,0 +1,92 @@
+/**********************************************************************
+ * Copyright (C) 2024 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ***********************************************************************/
+
+import { beforeEach, expect, test, vi } from 'vitest';
+import { RpcBrowser } from '@shared/src/messages/MessageProxy';
+import { RPCReadable } from './rpcReadable';
+import { studioClient, rpcBrowser } from '../utils/client';
+import type { ModelInfo } from '@shared/src/models/IModelInfo';
+
+const mocks = vi.hoisted(() => {
+  return {
+    getModelsInfoMock: vi.fn().mockResolvedValue([]),
+  };
+});
+
+vi.mock('../utils/client', async () => {
+  const window = {
+    addEventListener: (_: string, _f: (message: unknown) => void) => {},
+  } as unknown as Window;
+
+  const api = {
+    postMessage: (message: unknown) => {
+      if (message && typeof message === 'object' && 'channel' in message) {
+        const f = rpcBrowser.subscribers.get(message.channel as string);
+        f?.('');
+      }
+    },
+  } as unknown as PodmanDesktopApi;
+
+  const rpcBrowser = new RpcBrowser(window, api);
+
+  return {
+    rpcBrowser: rpcBrowser,
+    studioClient: {
+      getModelsInfo: mocks.getModelsInfoMock,
+    },
+  };
+});
+
+beforeEach(() => {
+  vi.clearAllMocks();
+});
+
+test('check updater is called once at subscription', async () => {
+  const rpcWritable = RPCReadable<string[]>([], [], () => {
+    studioClient.getModelsInfo();
+    return Promise.resolve(['']);
+  });
+  rpcWritable.subscribe(_ => {});
+  expect(mocks.getModelsInfoMock).toHaveBeenCalledOnce();
+});
+
+test('check updater is called twice if there is one event fired', async () => {
+  const rpcWritable = RPCReadable<string[]>([], ['event'], () => {
+    studioClient.getModelsInfo();
+    return Promise.resolve(['']);
+  });
+  rpcWritable.subscribe(_ => {});
+  rpcBrowser.invoke('event');
+  // wait for the timeout in the debouncer
+  await new Promise(resolve => setTimeout(resolve, 600));
+  expect(mocks.getModelsInfoMock).toHaveBeenCalledTimes(2);
+});
+
+test('check updater is called only twice because of the debouncer if there is more than one event in a row', async () => {
+  const rpcWritable = RPCReadable<ModelInfo[]>([], ['event2'], () => {
+    return studioClient.getModelsInfo();
+  });
+  rpcWritable.subscribe(_ => {});
+  rpcBrowser.invoke('event2');
+  rpcBrowser.invoke('event2');
+  rpcBrowser.invoke('event2');
+  rpcBrowser.invoke('event2');
+  // wait for the timeout in the debouncer
+  await new Promise(resolve => setTimeout(resolve, 600));
+  expect(mocks.getModelsInfoMock).toHaveBeenCalledTimes(2);
+});

packages/frontend/src/stores/rpcReadable.ts

@@ -0,0 +1,93 @@
+/**********************************************************************
+ * Copyright (C) 2024 Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ ***********************************************************************/
+
+import { writable, type Invalidator, type Subscriber, type Unsubscriber, type Readable } from 'svelte/store';
+import { rpcBrowser } from '../utils/client';
+import type { Subscriber as SharedSubscriber } from '@shared/src/messages/MessageProxy';
+
+export function RPCReadable<T>(
+  value: T,
+  // The event used to subscribe to a webview postMessage event
+  subscriptionEvents: string[],
+  // The initialization function that will be called to update the store at creation.
+  // For example, you can pass in a custom function such as "getPullingStatuses".
+  updater: () => Promise<T>,
+): Readable<T> {
+  let timeoutId: NodeJS.Timeout | undefined;
+  let timeoutThrottle: NodeJS.Timeout | undefined;
+
+  const debouncedUpdater = debounce(updater);
+  const origWritable = writable(value);
+
+  function subscribe(this: void, run: Subscriber<T>, invalidate?: Invalidator<T>): Unsubscriber {
+    const rcpSubscribes: SharedSubscriber[] = [];
+
+    for (const subscriptionEvent of subscriptionEvents) {
+      const rcpSubscribe = rpcBrowser.subscribe(subscriptionEvent, (_: unknown) => {
+        debouncedUpdater()
+          .then(v => origWritable.set(v))
+          .catch((e: unknown) => console.error('failed at updating store', String(e)));
+      });
+      rcpSubscribes.push(rcpSubscribe);
+    }
+
+    updater()
+      .then(v => origWritable.set(v))
+      .catch((e: unknown) => console.error('failed at init store', String(e)));
+
+    const unsubscribe = origWritable.subscribe(run, invalidate);
+    return () => {
+      rcpSubscribes.forEach(r => r.unsubscribe());
+      unsubscribe();
+    };
+  }
+
+  function debounce(func: () => Promise<T>): () => Promise<T> {
+    return () =>
+      new Promise<T>(resolve => {
+        if (timeoutId) {
+          clearTimeout(timeoutId);
+          timeoutId = undefined;
+        }
+
+        // throttle timeout, ask after 5s to update anyway to have at least UI being refreshed every 5s if there is a lot of events
+        // because debounce will defer all the events until the end so it's not so nice from UI side.
+        if (!timeoutThrottle) {
+          timeoutThrottle = setTimeout(() => {
+            if (timeoutId) {
+              clearTimeout(timeoutId);
+              timeoutId = undefined;
+            }
+            resolve(func());
+          }, 5000);
+        }
+
+        timeoutId = setTimeout(() => {
+          if (timeoutThrottle) {
+            clearTimeout(timeoutThrottle);
+            timeoutThrottle = undefined;
+          }
+          resolve(func());
+        }, 500);
+      });
+  }
+
+  return {
+    subscribe,
+  };
+}