Schedules for stores (#21)

Add IScheduleStore and local and remote implementations

Author: vaibhawvipul

lib/core/promise.ts

@@ -134,3 +134,15 @@ export function isCompletedPromise(
 ): p is ResolvedPromise | RejectedPromise | CanceledPromise | TimedoutPromise {
   return isDurablePromise(p) && ["RESOLVED", "REJECTED", "REJECTED_CANCELED", "REJECTED_TIMEDOUT"].includes(p.state);
 }
+
+export function searchStates(state: string | undefined): string[] {
+  if (state?.toLowerCase() == "pending") {
+    return ["PENDING"];
+  } else if (state?.toLowerCase() == "resolved") {
+    return ["RESOLVED"];
+  } else if (state?.toLowerCase() == "rejected") {
+    return ["REJECTED", "REJECTED_CANCELED", "REJECTED_TIMEDOUT"];
+  } else {
+    return ["PENDING", "RESOLVED", "REJECTED", "REJECTED_CANCELED", "REJECTED_TIMEDOUT"];
+  }
+}

lib/core/schedule.ts

@@ -0,0 +1,29 @@
+export type Schedule = {
+  id: string;
+  description?: string;
+  cron: string;
+  tags?: Record<string, string>;
+  promiseId: string;
+  promiseTimeout: number;
+  promiseParam?: {
+    data?: string;
+    headers: Record<string, string>;
+  };
+  promiseTags?: Record<string, string>;
+  lastRunTime?: number;
+  nextRunTime?: number;
+  idempotencyKey?: string;
+  createdOn?: number;
+};
+
+// Function to check if the response matches the Schedule type
+export function isSchedule(obj: any): obj is Schedule {
+  // You may need to adjust this based on the actual structure of your Schedule type
+  return (
+    obj !== undefined &&
+    typeof obj.id === "string" &&
+    typeof obj.cron === "string" &&
+    typeof obj.promiseId === "string" &&
+    typeof obj.promiseTimeout === "number"
+  );
+}

lib/core/storage.ts

@@ -1,6 +1,7 @@
-import { DurablePromise, TimedoutPromise, isPendingPromise } from "./promise";
+import { DurablePromise } from "./promise";
+import { Schedule } from "./schedule";
 
-export interface IStorage {
+export interface IPromiseStorage {
   rmw<P extends DurablePromise | undefined>(id: string, f: (promise: DurablePromise | undefined) => P): Promise<P>;
   search(
     id: string,
@@ -10,61 +11,13 @@ export interface IStorage {
   ): AsyncGenerator<DurablePromise[], void>;
 }
 
-export class WithTimeout implements IStorage {
-  constructor(private storage: IStorage) {}
-
-  rmw<P extends DurablePromise | undefined>(id: string, f: (promise: DurablePromise | undefined) => P): Promise<P> {
-    return this.storage.rmw(id, (promise) => f(timeout(promise)));
-  }
-
-  async *search(
+export interface IScheduleStorage {
+  rmw<S extends Schedule | undefined>(id: string, f: (schedule: Schedule | undefined) => S): Promise<S>;
+  search(
     id: string,
-    state: string | undefined,
     tags: Record<string, string> | undefined,
     limit: number | undefined,
-  ): AsyncGenerator<DurablePromise[], void, unknown> {
-    const regex = new RegExp(id.replaceAll("*", ".*"));
-    const tagEntries = Object.entries(tags ?? {});
-
-    let states: string[] = [];
-    if (state?.toLowerCase() == "pending") {
-      states = ["PENDING"];
-    } else if (state?.toLowerCase() == "resolved") {
-      states = ["RESOLVED"];
-    } else if (state?.toLowerCase() == "rejected") {
-      states = ["REJECTED", "REJECTED_CANCELED", "REJECTED_TIMEDOUT"];
-    } else {
-      states = ["PENDING", "RESOLVED", "REJECTED", "REJECTED_CANCELED", "REJECTED_TIMEDOUT"];
-    }
-
-    for await (const res of this.storage.search("*", undefined, undefined, limit)) {
-      yield res
-        .map(timeout)
-        .filter((promise) => regex.test(promise.id))
-        .filter((promise) => states.includes(promise.state))
-        .filter((promise) => tagEntries.every(([k, v]) => promise.tags?.[k] == v));
-    }
-  }
-}
-
-function timeout<P extends DurablePromise | undefined>(promise: P): P | TimedoutPromise {
-  if (isPendingPromise(promise) && Date.now() >= promise.timeout) {
-    return {
-      state: "REJECTED_TIMEDOUT",
-      id: promise.id,
-      timeout: promise.timeout,
-      param: promise.param,
-      value: {
-        headers: undefined,
-        data: undefined,
-      },
-      createdOn: promise.createdOn,
-      completedOn: promise.timeout,
-      idempotencyKeyForCreate: promise.idempotencyKeyForCreate,
-      idempotencyKeyForComplete: undefined,
-      tags: promise.tags,
-    };
-  }
+  ): AsyncGenerator<Schedule[], void>;
 
-  return promise;
+  delete(id: string): Promise<boolean>;
 }

lib/core/storages/indexdb.ts

@@ -1,8 +1,9 @@
-import { IStorage } from "../storage";
+import { IPromiseStorage } from "../storage";
 import { DurablePromise, isDurablePromise } from "../promise";
 import { ResonateError, ErrorCodes } from "../error";
+import { Schedule } from "../schedule";
 
-export class IndexedDbStorage implements IStorage {
+export class IndexedDbStorage implements IPromiseStorage {
   private dbName = "resonateDB";
   private readonly storeName = "promises";
   private db: Promise<IDBDatabase>;
@@ -32,7 +33,7 @@ export class IndexedDbStorage implements IStorage {
     }
   }
 
-  async rmw<P extends DurablePromise | undefined>(
+  async rmw<P extends DurablePromise | Schedule | undefined>(
     id: string,
     f: (promise: DurablePromise | undefined) => P,
   ): Promise<P> {
@@ -43,7 +44,7 @@ export class IndexedDbStorage implements IStorage {
     const storedPromise: DurablePromise | undefined = await this.getPromiseById(objectStore, id);
     const resultPromise = f(storedPromise);
 
-    if (resultPromise) {
+    if (isDurablePromise(resultPromise)) {
       await this.savePromise(objectStore, resultPromise);
     }
 
@@ -65,6 +66,11 @@ export class IndexedDbStorage implements IStorage {
     yield this.getAllPromises(objectStore);
   }
 
+  async deleteSchedule(id: string): Promise<boolean> {
+    // TODO: To be implemented
+    return true;
+  }
+
   private async getDb(): Promise<IDBDatabase> {
     return new Promise<IDBDatabase>((resolve, reject) => {
       const request = indexedDB.open(this.dbName, 1);

lib/core/storages/memory.ts

@@ -1,21 +1,18 @@
-import { IStorage } from "../storage";
-import { DurablePromise } from "../promise";
+import { IPromiseStorage, IScheduleStorage } from "../storage";
+import { DurablePromise, searchStates } from "../promise";
+import { Schedule } from "../schedule";
 
-export class MemoryStorage implements IStorage {
+export class MemoryPromiseStorage implements IPromiseStorage {
   private promises: Record<string, DurablePromise> = {};
-
   constructor() {}
 
-  async rmw<P extends DurablePromise | undefined>(
-    id: string,
-    f: (promise: DurablePromise | undefined) => P,
-  ): Promise<P> {
-    const promise = f(this.promises[id]);
-    if (promise) {
-      this.promises[id] = promise;
+  async rmw<P extends DurablePromise | undefined>(id: string, f: (item: DurablePromise | undefined) => P): Promise<P> {
+    const item = f(this.promises[id]);
+    if (item) {
+      this.promises[id] = item;
     }
 
-    return promise;
+    return item;
   }
 
   async *search(
@@ -24,8 +21,53 @@ export class MemoryStorage implements IStorage {
     tags: Record<string, string> | undefined,
     limit: number | undefined,
   ): AsyncGenerator<DurablePromise[], void> {
-    // for now WithTimeout will implement
-    // search logic
-    yield Object.values(this.promises);
+    // for the memory storage, we will ignore limit and return all
+    // promises that match the search criteria
+    const regex = new RegExp(id.replaceAll("*", ".*"));
+    const states = searchStates(state);
+    const tagEntries = Object.entries(tags ?? {});
+
+    yield Object.values(this.promises)
+      .filter((promise) => states.includes(promise.state))
+      .filter((promise) => regex.test(promise.id))
+      .filter((promise) => tagEntries.every(([k, v]) => promise.tags?.[k] == v));
+  }
+}
+
+export class MemoryScheduleStorage implements IScheduleStorage {
+  private schedules: Record<string, Schedule> = {};
+
+  constructor() {}
+
+  async rmw<S extends Schedule | undefined>(id: string, f: (item: Schedule | undefined) => S): Promise<S> {
+    const item = f(this.schedules[id]);
+    if (item) {
+      this.schedules[id] = item;
+    }
+
+    return item;
+  }
+
+  async *search(
+    id: string,
+    tags: Record<string, string> | undefined,
+    limit: number | undefined,
+  ): AsyncGenerator<Schedule[], void> {
+    // for the memory storage, we will ignore limit and return all
+    // schedules that match the search criteria
+    const regex = new RegExp(id.replaceAll("*", ".*"));
+    const tagEntries = Object.entries(tags ?? {});
+
+    yield Object.values(this.schedules)
+      .filter((schedule) => regex.test(schedule.id))
+      .filter((schedule) => tagEntries.every(([k, v]) => schedule.tags?.[k] == v));
+  }
+
+  async delete(id: string): Promise<boolean> {
+    if (this.schedules[id]) {
+      delete this.schedules[id];
+      return true;
+    }
+    return false;
   }
 }

lib/core/storages/withTimeout.ts

@@ -0,0 +1,49 @@
+import { DurablePromise, TimedoutPromise, isPendingPromise, searchStates } from "../promise";
+import { IPromiseStorage } from "../storage";
+
+export class WithTimeout implements IPromiseStorage {
+  constructor(private storage: IPromiseStorage) {}
+
+  rmw<P extends DurablePromise | undefined>(id: string, f: (promise: DurablePromise | undefined) => P): Promise<P> {
+    return this.storage.rmw(id, (promise) => f(promise ? timeout(promise) : undefined));
+  }
+
+  async *search(
+    id: string,
+    state: string | undefined,
+    tags: Record<string, string> | undefined,
+    limit: number | undefined,
+  ): AsyncGenerator<DurablePromise[], void, unknown> {
+    // search promises against all provided criteria except state,
+    // then timeout any pending promises that have exceeded the
+    // timout, and finally apply the state filter
+
+    const states = searchStates(state);
+
+    for await (const promises of this.storage.search(id, undefined, tags, limit)) {
+      yield promises.map(timeout).filter((promise) => states.includes(promise.state));
+    }
+  }
+}
+
+function timeout<P extends DurablePromise>(promise: P): P | TimedoutPromise {
+  if (isPendingPromise(promise) && Date.now() >= promise.timeout) {
+    return {
+      state: "REJECTED_TIMEDOUT",
+      id: promise.id,
+      timeout: promise.timeout,
+      param: promise.param,
+      value: {
+        headers: undefined,
+        data: undefined,
+      },
+      createdOn: promise.createdOn,
+      completedOn: promise.timeout,
+      idempotencyKeyForCreate: promise.idempotencyKeyForCreate,
+      idempotencyKeyForComplete: undefined,
+      tags: promise.tags,
+    };
+  }
+
+  return promise;
+}

lib/core/store.ts

@@ -8,6 +8,8 @@ import {
   isDurablePromise,
 } from "./promise";
 
+import { Schedule, isSchedule } from "./schedule";
+
 export interface SearchPromiseParams {
   id: string;
   state?: string;
@@ -31,6 +33,17 @@ export function isSearchPromiseResult(obj: any): obj is SearchPromiseResult {
   );
 }
 
+export function isSearchSchedulesResult(obj: any): obj is { cursor: string; schedules: Schedule[] } {
+  return (
+    obj !== undefined &&
+    obj.cursor !== undefined &&
+    (obj.cursor === null || typeof obj.cursor === "string") &&
+    obj.schedules !== undefined &&
+    Array.isArray(obj.schedules) &&
+    obj.schedules.every(isSchedule)
+  );
+}
+
 /**
  * Promise Store API
  */
@@ -133,3 +146,57 @@ export interface IPromiseStore {
     limit?: number,
   ): AsyncGenerator<DurablePromise[], void>;
 }
+
+export interface IScheduleStore {
+  /**
+   * Creates a new schedule.
+   *
+   * @param id Unique identifier for the schedule.
+   * @param ikey Idempotency key associated with the create operation.
+   * @param description Description of the schedule.
+   * @param cron CRON expression defining the schedule's execution time.
+   * @param tags Key-value pairs associated with the schedule.
+   * @param promiseId Unique identifier for the associated promise.
+   * @param promiseTimeout Timeout for the associated promise in milliseconds.
+   * @param promiseHeaders Headers associated with the promise data.
+   * @param promiseData Encoded data for the promise of type string.
+   * @param promiseTags Key-value pairs associated with the promise.
+   * @returns A Promise resolving to the created schedule.
+   */
+  create(
+    id: string,
+    ikey: string | undefined,
+    description: string | undefined,
+    cron: string,
+    tags: Record<string, string> | undefined,
+    promiseId: string,
+    promiseTimeout: number,
+    promiseHeaders: Record<string, string> | undefined,
+    promiseData: string | undefined,
+    promiseTags: Record<string, string> | undefined,
+  ): Promise<Schedule>;
+
+  /**
+   * Retrieves a schedule based on its id.
+   *
+   * @param id Unique identifier for the promise to be retrieved.
+   * @returns A promise schedule that is pending, canceled, resolved, or rejected.
+   */
+  get(id: string): Promise<Schedule>;
+
+  /**
+   * Deletes a schedule based on its id.
+   * @param id Unique identifier for the promise to be deleted.
+   * @returns A promise schedule that is pending, canceled, resolved, or rejected.
+   */
+  delete(id: string): Promise<boolean>;
+
+  /**
+   * Search for schedules.
+   *
+   * @param id Ids to match, can include wildcards.
+   * @param tags Tags to match.
+   * @returns A list of promise schedules.
+   */
+  search(id: string, tags?: Record<string, string>, limit?: number): AsyncGenerator<Schedule[], void>;
+}