2 files changed, 49 insertions, 0 deletions
diff --git a/src/lib/async/accumulate.ts b/src/lib/async/accumulate.ts
new file mode 100644
index 000000000..99226418e
--- /dev/null
+++ b/src/lib/async/accumulate.ts
@@ -0,0 +1,25 @@
+export interface AccumulateResponse<T> {
+  cursor?: string
+  items: T[]
+}
+
+export type AccumulateFetchFn<T> = (
+  cursor: string | undefined,
+) => Promise<AccumulateResponse<T>>
+
+export async function accumulate<T>(
+  fn: AccumulateFetchFn<T>,
+  pageLimit = 100,
+): Promise<T[]> {
+  let cursor: string | undefined
+  let acc: T[] = []
+  for (let i = 0; i < pageLimit; i++) {
+    const res = await fn(cursor)
+    cursor = res.cursor
+    acc = acc.concat(res.items)
+    if (!cursor) {
+      break
+    }
+  }
+  return acc
+}
diff --git a/src/lib/async/until.ts b/src/lib/async/until.ts
new file mode 100644
index 000000000..db53c9218
--- /dev/null
+++ b/src/lib/async/until.ts
@@ -0,0 +1,24 @@
+import {timeout} from './timeout'
+
+export async function until(
+  retries: number,
+  delay: number,
+  cond: (v: any, err: any) => boolean,
+  fn: () => Promise<any>,
+): Promise<boolean> {
+  while (retries > 0) {
+    try {
+      const v = await fn()
+      if (cond(v, undefined)) {
+        return true
+      }
+    } catch (e: any) {
+      if (cond(undefined, e)) {
+        return true
+      }
+    }
+    await timeout(delay)
+    retries--
+  }
+  return false
+}