Update method for deassetization (#213)

Author: AaronGBoswell

packages/data-sdk/package.json

@@ -49,6 +49,7 @@
     "@eslint/js": "^9.11.1",
     "@formant/realtime-sdk": "1.4.3",
     "@types/base-64": "^1.0.0",
+    "@types/fast-json-stable-stringify": "=2.0.0",
     "@types/node": "^18.16.3",
     "@types/pako": "^2.0.0",
     "@typescript-eslint/eslint-plugin": "^8.7.0",
@@ -75,7 +76,9 @@
     "base64-js": "^1.5.1",
     "date-fns": "^2.30.0",
     "eventemitter3": "^5.0.1",
+    "fast-json-stable-stringify": "=2.0.0",
     "google-protobuf": "^3.21.2",
+    "lru-cache": "^7.14.0",
     "lzfjs": "^1.0.1",
     "pako": "^2.1.0",
     "validator": "^13.11.0"

packages/data-sdk/src/connector/common/LruCache.ts

@@ -0,0 +1,70 @@
+import stringify from "fast-json-stable-stringify";
+import BaseLruCache from "lru-cache";
+
+export type CacheOptions<V> = BaseLruCache.Options<string, V> & {
+  name: string;
+  fastStringify?: boolean;
+};
+
+export class LruCache<K, V> {
+  private cache!: BaseLruCache<string, V>;
+  private stringify: (_: K) => string;
+
+  constructor(private options: CacheOptions<V>) {
+    this.cache = new BaseLruCache<string, V>({
+      // if using the dispose callback,
+      // by default automatically prune expired entries so
+      // they are delivered consistently and quickly
+      ...(options.dispose || options.disposeAfter
+        ? { ttlAutopurge: true }
+        : {}),
+      ...options,
+      dispose: (...args) => {
+        options.dispose?.(...args);
+      },
+      disposeAfter: (...args) => {
+        options.disposeAfter?.(...args);
+      },
+    });
+    this.stringify = options.fastStringify ? JSON.stringify : stringify;
+  }
+
+  public set(key: K, value: V, ttl?: number): void {
+    const keyString = this.stringify(key);
+    if (!this.cache.set(keyString, value, { ttl })) {
+      const size = this.cache.sizeCalculation
+        ? this.cache.sizeCalculation(value, keyString)
+        : "unknown";
+      throw Error(`Value too large (${size} > ${this.cache.max})`);
+    }
+  }
+
+  public get(key: K): V | undefined {
+    const keyString = this.stringify(key);
+    return this.cache.get(keyString);
+  }
+
+  public delete(key: K) {
+    this.cache.delete(this.stringify(key));
+  }
+
+  public peek(key: K): V | undefined {
+    return this.cache.peek(this.stringify(key));
+  }
+
+  public size() {
+    return this.cache.size;
+  }
+
+  public clear() {
+    this.cache.clear();
+  }
+
+  public forEach(callback: (value: V) => void) {
+    this.cache.forEach(callback);
+  }
+
+  public purgeStale(): boolean {
+    return this.cache.purgeStale();
+  }
+}

packages/data-sdk/src/connector/common/PromiseLruCache.ts

@@ -0,0 +1,45 @@
+import { duration } from "./duration";
+
+import { delay } from "./delay";
+import { CacheOptions, LruCache } from "./LruCache";
+
+export type PromiseCacheOptions<V> = CacheOptions<Promise<V>> & {
+  expireRejectedPromiseValues?: boolean;
+  rejectedPromiseValueTtl?: number;
+};
+
+export class PromiseLruCache<K, V> extends LruCache<K, Promise<V>> {
+  protected expireRejectedPromiseValues: boolean;
+  protected rejectedPromiseValueTtl: number;
+
+  constructor(options: PromiseCacheOptions<V>) {
+    super(options);
+
+    this.expireRejectedPromiseValues =
+      options.expireRejectedPromiseValues !== undefined
+        ? options.expireRejectedPromiseValues
+        : true;
+
+    this.rejectedPromiseValueTtl =
+      options.rejectedPromiseValueTtl !== undefined
+        ? options.rejectedPromiseValueTtl
+        : duration.second;
+
+    if (this.rejectedPromiseValueTtl < 0) {
+      throw new Error("rejectedPromiseValueTtl must not be negative");
+    }
+  }
+
+  public set(key: K, value: Promise<V>, ttl?: number): void {
+    super.set(key, value, ttl);
+
+    if (this.expireRejectedPromiseValues) {
+      value.catch(async () => {
+        await delay(this.rejectedPromiseValueTtl);
+        if (this.peek(key) === value) {
+          this.delete(key);
+        }
+      });
+    }
+  }
+}

packages/data-sdk/src/connector/common/errorToObject.ts

@@ -0,0 +1,22 @@
+import { toStringSafe } from "./toStringSafe";
+
+/**
+ * Convert an error object to plain javascript object, including any custom properties.
+ */
+export function errorToObject(
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  error: any
+) {
+  if (!error.stack) {
+    return { message: toStringSafe(error) };
+  }
+
+  const { name, message, stack, ...meta } = error;
+
+  return {
+    name,
+    message: message !== undefined ? message : toStringSafe(error),
+    stack,
+    meta: meta && Object.keys(meta).length === 0 ? undefined : meta,
+  };
+}

packages/data-sdk/src/connector/common/errorToString.ts

@@ -0,0 +1,13 @@
+import { errorToObject } from "./errorToObject";
+import { toStringSafe } from "./toStringSafe";
+
+/**
+ * Serialize an error object to a string, including any custom properties.
+ */
+export function errorToString(error: unknown): string {
+  const { message, stack, meta } = errorToObject(error);
+
+  return `${stack || message || ""}${
+    meta && Object.keys(meta).length > 0 ? ` -- ${toStringSafe(meta)}` : ""
+  }`;
+}

packages/data-sdk/src/connector/common/toStringSafe.ts

@@ -0,0 +1,12 @@
+import stringify from "fast-json-stable-stringify";
+
+/**
+ * Safely stringify input.
+ * Similar to `JSON.stringify` but
+ * doesn't throw an error if input includes circular properties.
+ *
+ * Warning: output is not guaranteed to be a valid JSON string.
+ */
+export function toStringSafe(input: unknown): string {
+    return stringify(input, { cycles: true });
+}

packages/data-sdk/src/connector/data/BaseUniverseDataConnector.ts

@@ -32,6 +32,7 @@ import {
   RealtimeButtonConfiguration,
   UniverseDataSource,
 } from "../model/IUniverseData";
+import { DataLoader } from "./loader/DataLoader";
 import { QueryStore } from "./queryStore";
 
 export type DeviceId = string;
@@ -68,6 +69,7 @@ export class BasicUniverseDataConnector {
   timeChangeListeners: ((time: Date | "live") => void)[] = [];
 
   queryStore: QueryStore = new QueryStore();
+  dataLoader: any;
 
   setTime(time: Date | "live"): void {
     if (time !== "live") {
@@ -86,6 +88,7 @@ export class BasicUniverseDataConnector {
 
   constructor() {
     this.time = "live";
+    this.dataLoader = DataLoader.get();
 
     const dataLoop = async () => {
       if (Array.from(this.subscriberLoaders.keys()).length > 0) {

packages/data-sdk/src/connector/data/LiveUniverseData.ts

@@ -303,7 +303,7 @@ export class LiveUniverseData
         deviceId,
         source,
         async (data: IStreamData[]): Promise<DataResult<T>> => {
-          let foundUrl: string | undefined;
+          let found: string | undefined;
           let jsonValue: T | undefined;
           for (let i = 0; i < data.length; i += 1) {
             const _ = data[i];
@@ -313,13 +313,14 @@ export class LiveUniverseData
               _.type === "json"
             ) {
               const [_time, value] = _.points[_.points.length - 1];
-              foundUrl = value as string;
+              found = value as string;
             }
           }
-          if (foundUrl) {
-            jsonValue = await fetch(foundUrl).then(
-              (r) => r.json() as Promise<T>
-            );
+          if (found?.startsWith("http")) {
+            const loaded = await this.dataLoader.load(found);
+            jsonValue = loaded.json;
+          } else {
+            jsonValue = JSON.parse(found || "{}");
           }
           return {
             deviceId,