From 710f63977deeeb791d6c68f4303706585be30cea Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kevin=20Gr=C3=BCneberg?= <k.grueneberg1994@gmail.com>
Date: Thu, 2 May 2024 23:30:17 +0800
Subject: [PATCH] feat: support manual sync/backfill (#6)

---
 .github/workflows/ci.yml                      |   2 +-
 apps/node-fastify/.env.sample                 |  14 +-
 apps/node-fastify/package.json                |   3 +-
 apps/node-fastify/src/app.ts                  |  15 +-
 apps/node-fastify/src/routes/sync.ts          | 163 ++++++++++++++
 apps/node-fastify/src/routes/webhooks.ts      |   2 +-
 apps/node-fastify/src/server.ts               |   9 +-
 apps/node-fastify/src/utils/config.ts         |  50 +++++
 apps/node-fastify/src/utils/verifyApiKey.ts   |  15 ++
 package-lock.json                             | 202 +++++++++++-------
 package.json                                  |   6 +-
 packages/orb-sync-lib/package.json            |   2 +-
 .../orb-sync-lib/src/database/postgres.ts     |   2 +
 packages/orb-sync-lib/src/orb-sync.ts         |  42 +++-
 .../orb-sync-lib/src/sync/credit_notes.ts     |  18 ++
 packages/orb-sync-lib/src/sync/customers.ts   |  24 +++
 packages/orb-sync-lib/src/sync/invoices.ts    |  24 +++
 .../orb-sync-lib/src/sync/subscriptions.ts    |  24 +++
 packages/orb-sync-lib/src/types.ts            |  28 +++
 19 files changed, 531 insertions(+), 114 deletions(-)
 create mode 100644 apps/node-fastify/src/routes/sync.ts
 create mode 100644 apps/node-fastify/src/utils/config.ts
 create mode 100644 apps/node-fastify/src/utils/verifyApiKey.ts

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index c69d0ad..dbd728b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -26,7 +26,7 @@ jobs:
           echo ORB_WEBHOOK_SECRET=whsec_ >> .env
           echo SCHEMA=orb >> .env
           echo PORT=8080 >> .env
-          echo API_KEY=api_key_test >> .env
+          echo API_KEY_SYNC=api_key_test >> .env
 
       - name: Install dependencies
         run: |
diff --git a/apps/node-fastify/.env.sample b/apps/node-fastify/.env.sample
index d4c7359..7702a94 100644
--- a/apps/node-fastify/.env.sample
+++ b/apps/node-fastify/.env.sample
@@ -1,16 +1,18 @@
+# Postgres database URL including auth and search path
 DATABASE_URL=postgres://postgres:postgres@host:5432/postgres?sslmode=disable&search_path=orb
 
-# optional
+# Optional
 SCHEMA=orb
 
+# Secret to validate signatures of Orb webhooks
 ORB_WEBHOOK_SECRET=whsec_
 
-# optional, only needed when you want to actively sync data and call the Orb API, not needed for webhook processing
+# Access the Orb API
 ORB_API_KEY=test_
 
-# optional
-PORT=8080
+# Optional, API key to authorize requests against the sync endpoints. When calling a sync endpoint this value has to be provided via HTTP header "Authorization".
+#API_KEY_SYNC=
 
-# API key to access the endpoints of the Node.js app
-API_KEY=api_key_test
+# Optional
+PORT=8080
 
diff --git a/apps/node-fastify/package.json b/apps/node-fastify/package.json
index b3ff222..3a83791 100644
--- a/apps/node-fastify/package.json
+++ b/apps/node-fastify/package.json
@@ -18,9 +18,10 @@
   "license": "MIT",
   "dependencies": {
     "@fastify/autoload": "^5.8.0",
+    "@fastify/type-provider-typebox": "^4.0.0",
     "dotenv": "^16.4.5",
     "fastify": "^4.26.2",
-    "pino": "^8.20.0",
+    "pino": "^9.0.0",
     "orb-sync-lib": "*"
   },
   "devDependencies": {
diff --git a/apps/node-fastify/src/app.ts b/apps/node-fastify/src/app.ts
index 90d8f1d..54aabb2 100644
--- a/apps/node-fastify/src/app.ts
+++ b/apps/node-fastify/src/app.ts
@@ -2,7 +2,7 @@ import fastify, { FastifyInstance, FastifyServerOptions } from 'fastify';
 import autoload from '@fastify/autoload';
 import path from 'node:path';
 import { OrbSync } from 'orb-sync-lib';
-import assert from 'node:assert';
+import { getConfig } from './utils/config';
 
 export async function createApp(opts: FastifyServerOptions = {}): Promise<FastifyInstance> {
   const app = fastify(opts);
@@ -35,16 +35,13 @@ export async function createApp(opts: FastifyServerOptions = {}): Promise<Fastif
     dir: path.join(__dirname, 'routes'),
   });
 
-  const { DATABASE_URL, ORB_WEBHOOK_SECRET, DATABASE_SCHEMA, ORB_API_KEY } = process.env;
-
-  assert(DATABASE_URL, 'DATABASE_URL is required');
-  assert(ORB_WEBHOOK_SECRET, 'ORB_WEBHOOK_SECRET is required');
+  const config = getConfig();
 
   const orbSync = new OrbSync({
-    databaseUrl: DATABASE_URL,
-    orbWebhookSecret: ORB_WEBHOOK_SECRET,
-    databaseSchema: DATABASE_SCHEMA || 'orb',
-    orbApiKey: ORB_API_KEY,
+    databaseUrl: config.DATABASE_URL,
+    orbWebhookSecret: config.ORB_WEBHOOK_SECRET,
+    databaseSchema: config.DATABASE_SCHEMA,
+    orbApiKey: config.ORB_API_KEY,
   });
 
   app.decorate('orbSync', orbSync);
diff --git a/apps/node-fastify/src/routes/sync.ts b/apps/node-fastify/src/routes/sync.ts
new file mode 100644
index 0000000..4776b07
--- /dev/null
+++ b/apps/node-fastify/src/routes/sync.ts
@@ -0,0 +1,163 @@
+import { Static, Type } from '@fastify/type-provider-typebox';
+import type { FastifyInstance } from 'fastify';
+import { verifyApiKey } from '../utils/verifyApiKey';
+
+const SchemaRequestParamsSyncCreditNotes = Type.Object({
+  limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })),
+});
+
+const SchemaRequestParamsSyncCustomers = Type.Object({
+  limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })),
+  createdAtGt: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtGte: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtLt: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtLte: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+});
+
+const SchemaRequestParamsSyncSubscriptions = Type.Object({
+  limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })),
+  createdAtGt: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtGte: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtLt: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtLte: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+});
+
+const SchemaRequestParamsSyncInvoices = Type.Object({
+  limit: Type.Optional(Type.Number({ minimum: 1, maximum: 100 })),
+  createdAtGt: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtGte: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtLt: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+  createdAtLte: Type.Optional(
+    Type.String({
+      format: 'date-time',
+    })
+  ),
+});
+
+export default async function routes(fastify: FastifyInstance) {
+  fastify.post<{
+    Querystring: Static<typeof SchemaRequestParamsSyncCreditNotes>;
+  }>('/sync/credit_notes', {
+    preHandler: [verifyApiKey],
+    schema: {
+      querystring: SchemaRequestParamsSyncCreditNotes,
+    },
+    handler: async (request, reply) => {
+      const query = request.query;
+
+      const count = await fastify.orbSync.sync('credit_notes', { limit: query.limit });
+
+      return reply.send({ count });
+    },
+  });
+
+  fastify.post<{
+    Querystring: Static<typeof SchemaRequestParamsSyncCustomers>;
+  }>('/sync/customers', {
+    preHandler: [verifyApiKey],
+    schema: {
+      querystring: SchemaRequestParamsSyncCustomers,
+    },
+    handler: async (request, reply) => {
+      const query = request.query;
+
+      const count = await fastify.orbSync.sync('customers', {
+        limit: query.limit,
+        createdAtGt: query.createdAtGt,
+        createdAtGte: query.createdAtGte,
+        createdAtLt: query.createdAtLt,
+        createdAtLte: query.createdAtLte,
+      });
+
+      return reply.send({ count });
+    },
+  });
+
+  fastify.post<{
+    Querystring: Static<typeof SchemaRequestParamsSyncSubscriptions>;
+  }>('/sync/subscriptions', {
+    preHandler: [verifyApiKey],
+    schema: {
+      querystring: SchemaRequestParamsSyncSubscriptions,
+    },
+    handler: async (request, reply) => {
+      const query = request.query;
+
+      const count = await fastify.orbSync.sync('subscriptions', {
+        limit: query.limit,
+        createdAtGt: query.createdAtGt,
+        createdAtGte: query.createdAtGte,
+        createdAtLt: query.createdAtLt,
+        createdAtLte: query.createdAtLte,
+      });
+
+      return reply.send({ count });
+    },
+  });
+
+  fastify.post<{
+    Querystring: Static<typeof SchemaRequestParamsSyncInvoices>;
+  }>('/sync/invoices', {
+    preHandler: [verifyApiKey],
+    schema: {
+      querystring: SchemaRequestParamsSyncInvoices,
+    },
+    handler: async (request, reply) => {
+      const query = request.query;
+
+      const count = await fastify.orbSync.sync('invoices', {
+        limit: query.limit,
+        createdAtGt: query.createdAtGt,
+        createdAtGte: query.createdAtGte,
+        createdAtLt: query.createdAtLt,
+        createdAtLte: query.createdAtLte,
+      });
+
+      return reply.send({ count });
+    },
+  });
+}
diff --git a/apps/node-fastify/src/routes/webhooks.ts b/apps/node-fastify/src/routes/webhooks.ts
index 8aa4d49..37b12e8 100644
--- a/apps/node-fastify/src/routes/webhooks.ts
+++ b/apps/node-fastify/src/routes/webhooks.ts
@@ -6,7 +6,7 @@ export default async function routes(fastify: FastifyInstance) {
       const headers = request.headers;
       const body: { raw: Buffer } = request.body as { raw: Buffer };
 
-      await fastify.orbSync.sync(body.raw.toString(), headers);
+      await fastify.orbSync.processWebhook(body.raw.toString(), headers);
 
       return reply.send({ received: true });
     },
diff --git a/apps/node-fastify/src/server.ts b/apps/node-fastify/src/server.ts
index 68d5e9e..9b31a05 100644
--- a/apps/node-fastify/src/server.ts
+++ b/apps/node-fastify/src/server.ts
@@ -3,6 +3,8 @@ import type { FastifyInstance } from 'fastify';
 import type { Server, IncomingMessage, ServerResponse } from 'node:http';
 import { createApp } from './app';
 import pino from 'pino';
+import { TypeBoxTypeProvider } from '@fastify/type-provider-typebox';
+import { getConfig } from './utils/config';
 
 const logger = pino({
   formatters: {
@@ -20,11 +22,12 @@ const main = async () => {
     requestIdHeader: 'Request-Id',
   });
 
-  // Init config
-  const port = process.env.PORT ? Number(process.env.PORT) : 8080;
+  app.withTypeProvider<TypeBoxTypeProvider>();
+
+  const config = getConfig();
 
   // Start the server
-  app.listen({ port: Number(port), host: '0.0.0.0' }, (err, address) => {
+  app.listen({ port: config.PORT, host: '0.0.0.0' }, (err, address) => {
     if (err) {
       console.error(err);
       process.exit(1);
diff --git a/apps/node-fastify/src/utils/config.ts b/apps/node-fastify/src/utils/config.ts
new file mode 100644
index 0000000..8f4330a
--- /dev/null
+++ b/apps/node-fastify/src/utils/config.ts
@@ -0,0 +1,50 @@
+import dotenv from 'dotenv';
+import assert from 'assert';
+
+type configType = {
+  /** Optional, API key to authorize requests against the sync endpoints */
+  API_KEY_SYNC?: string;
+
+  /** Port number the API is running on, defaults to 8080 */
+  PORT: number;
+
+  /** Postgres database URL including auth and search path */
+  DATABASE_URL: string
+
+  /** Secret to validate signatures of Orb webhooks */
+  ORB_WEBHOOK_SECRET: string
+
+  /** Defaults to Orb */
+  DATABASE_SCHEMA: string
+
+  /** Access the Orb API */
+  ORB_API_KEY?: string
+};
+
+function getConfigFromEnv(key: string, defaultValue?: string): string {
+  const value = process.env[key];
+  return value || defaultValue!;
+}
+
+let config: configType;
+
+export function getConfig(): configType {
+  if (config) return config;
+
+  dotenv.config();
+
+  config = {
+    API_KEY_SYNC: getConfigFromEnv('API_KEY_SYNC'),
+    ORB_API_KEY: getConfigFromEnv('ORB_API_KEY'),
+    DATABASE_SCHEMA: getConfigFromEnv('DATABASE_SCHEMA', 'orb'),
+    DATABASE_URL: getConfigFromEnv('DATABASE_URL'),
+    ORB_WEBHOOK_SECRET: getConfigFromEnv('ORB_WEBHOOK_SECRET'),
+    PORT: Number(getConfigFromEnv('PORT', '8080')),
+  };
+
+  assert(!Number.isNaN(config.PORT), 'PORT must be a number');
+  assert(config.DATABASE_URL, 'DATABASE_URL is required');
+  assert(config.ORB_WEBHOOK_SECRET, 'ORB_WEBHOOK_SECRET is required');
+
+  return config;
+}
diff --git a/apps/node-fastify/src/utils/verifyApiKey.ts b/apps/node-fastify/src/utils/verifyApiKey.ts
new file mode 100644
index 0000000..4856cfc
--- /dev/null
+++ b/apps/node-fastify/src/utils/verifyApiKey.ts
@@ -0,0 +1,15 @@
+import { FastifyReply, FastifyRequest, HookHandlerDoneFunction } from 'fastify';
+import { getConfig } from './config';
+
+export const verifyApiKey = (request: FastifyRequest, reply: FastifyReply, done: HookHandlerDoneFunction): unknown => {
+  const config = getConfig();
+
+  if (!request.headers || !request.headers.authorization) {
+    return reply.code(401).send('Unauthorized');
+  }
+  const { authorization } = request.headers;
+  if (authorization !== config.API_KEY_SYNC) {
+    return reply.code(401).send('Unauthorized');
+  }
+  done();
+};
diff --git a/package-lock.json b/package-lock.json
index 9112b1f..11c2653 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -13,12 +13,12 @@
         "apps/*"
       ],
       "devDependencies": {
-        "@typescript-eslint/eslint-plugin": "^7.6.0",
-        "@typescript-eslint/parser": "^7.6.0",
+        "@typescript-eslint/eslint-plugin": "7.8.0",
+        "@typescript-eslint/parser": "^7.8.0",
         "eslint": "^8.57.0",
         "eslint-config-prettier": "^9.1.0",
         "prettier": "^4.0.0-alpha.8",
-        "turbo": "^1.13.2",
+        "turbo": "^1.13.3",
         "typescript": "^5.4.5"
       }
     },
@@ -28,16 +28,38 @@
       "license": "MIT",
       "dependencies": {
         "@fastify/autoload": "^5.8.0",
+        "@fastify/type-provider-typebox": "^4.0.0",
         "dotenv": "^16.4.5",
         "fastify": "^4.26.2",
         "orb-sync-lib": "*",
-        "pino": "^8.20.0"
+        "pino": "^9.0.0"
       },
       "devDependencies": {
         "pino-pretty": "^11.0.0",
         "tsx": "^4.7.2"
       }
     },
+    "apps/node-fastify/node_modules/pino": {
+      "version": "9.0.0",
+      "resolved": "https://registry.npmjs.org/pino/-/pino-9.0.0.tgz",
+      "integrity": "sha512-uI1ThkzTShNSwvsUM6b4ND8ANzWURk9zTELMztFkmnCQeR/4wkomJ+echHee5GMWGovoSfjwdeu80DsFIt7mbA==",
+      "dependencies": {
+        "atomic-sleep": "^1.0.0",
+        "fast-redact": "^3.1.1",
+        "on-exit-leak-free": "^2.1.0",
+        "pino-abstract-transport": "^1.2.0",
+        "pino-std-serializers": "^6.0.0",
+        "process-warning": "^3.0.0",
+        "quick-format-unescaped": "^4.0.3",
+        "real-require": "^0.2.0",
+        "safe-stable-stringify": "^2.3.1",
+        "sonic-boom": "^3.7.0",
+        "thread-stream": "^2.6.0"
+      },
+      "bin": {
+        "pino": "bin.js"
+      }
+    },
     "node_modules/@aashutoshrathi/word-wrap": {
       "version": "1.2.6",
       "resolved": "https://registry.npmjs.org/@aashutoshrathi/word-wrap/-/word-wrap-1.2.6.tgz",
@@ -527,6 +549,14 @@
         "fast-deep-equal": "^3.1.3"
       }
     },
+    "node_modules/@fastify/type-provider-typebox": {
+      "version": "4.0.0",
+      "resolved": "https://registry.npmjs.org/@fastify/type-provider-typebox/-/type-provider-typebox-4.0.0.tgz",
+      "integrity": "sha512-kTlN0saC/+xhcQPyBjb3YONQAMjiD/EHlCRjQjsr5E3NFjS5K8ZX5LGzXYDRjSa+sV4y8gTL5Q7FlObePv4iTA==",
+      "peerDependencies": {
+        "@sinclair/typebox": ">=0.26 <=0.32"
+      }
+    },
     "node_modules/@humanwhocodes/config-array": {
       "version": "0.11.14",
       "resolved": "https://registry.npmjs.org/@humanwhocodes/config-array/-/config-array-0.11.14.tgz",
@@ -601,6 +631,12 @@
         "node": ">= 8"
       }
     },
+    "node_modules/@sinclair/typebox": {
+      "version": "0.32.27",
+      "resolved": "https://registry.npmjs.org/@sinclair/typebox/-/typebox-0.32.27.tgz",
+      "integrity": "sha512-JHRrubCKiXi6VKlbBTpTQnExkUFasPMIaXCJYJhqVBGLliQVt1yBZZgiZo3/uSmvAdXlIIdGoTAT6RB09L0QqA==",
+      "peer": true
+    },
     "node_modules/@types/json-schema": {
       "version": "7.0.15",
       "resolved": "https://registry.npmjs.org/@types/json-schema/-/json-schema-7.0.15.tgz",
@@ -710,16 +746,16 @@
       "dev": true
     },
     "node_modules/@typescript-eslint/eslint-plugin": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/eslint-plugin/-/eslint-plugin-7.6.0.tgz",
-      "integrity": "sha512-gKmTNwZnblUdnTIJu3e9kmeRRzV2j1a/LUO27KNNAnIC5zjy1aSvXSRp4rVNlmAoHlQ7HzX42NbKpcSr4jF80A==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/eslint-plugin/-/eslint-plugin-7.8.0.tgz",
+      "integrity": "sha512-gFTT+ezJmkwutUPmB0skOj3GZJtlEGnlssems4AjkVweUPGj7jRwwqg0Hhg7++kPGJqKtTYx+R05Ftww372aIg==",
       "dev": true,
       "dependencies": {
         "@eslint-community/regexpp": "^4.10.0",
-        "@typescript-eslint/scope-manager": "7.6.0",
-        "@typescript-eslint/type-utils": "7.6.0",
-        "@typescript-eslint/utils": "7.6.0",
-        "@typescript-eslint/visitor-keys": "7.6.0",
+        "@typescript-eslint/scope-manager": "7.8.0",
+        "@typescript-eslint/type-utils": "7.8.0",
+        "@typescript-eslint/utils": "7.8.0",
+        "@typescript-eslint/visitor-keys": "7.8.0",
         "debug": "^4.3.4",
         "graphemer": "^1.4.0",
         "ignore": "^5.3.1",
@@ -745,15 +781,15 @@
       }
     },
     "node_modules/@typescript-eslint/parser": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/parser/-/parser-7.6.0.tgz",
-      "integrity": "sha512-usPMPHcwX3ZoPWnBnhhorc14NJw9J4HpSXQX4urF2TPKG0au0XhJoZyX62fmvdHONUkmyUe74Hzm1//XA+BoYg==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/parser/-/parser-7.8.0.tgz",
+      "integrity": "sha512-KgKQly1pv0l4ltcftP59uQZCi4HUYswCLbTqVZEJu7uLX8CTLyswqMLqLN+2QFz4jCptqWVV4SB7vdxcH2+0kQ==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/scope-manager": "7.6.0",
-        "@typescript-eslint/types": "7.6.0",
-        "@typescript-eslint/typescript-estree": "7.6.0",
-        "@typescript-eslint/visitor-keys": "7.6.0",
+        "@typescript-eslint/scope-manager": "7.8.0",
+        "@typescript-eslint/types": "7.8.0",
+        "@typescript-eslint/typescript-estree": "7.8.0",
+        "@typescript-eslint/visitor-keys": "7.8.0",
         "debug": "^4.3.4"
       },
       "engines": {
@@ -773,13 +809,13 @@
       }
     },
     "node_modules/@typescript-eslint/scope-manager": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/scope-manager/-/scope-manager-7.6.0.tgz",
-      "integrity": "sha512-ngttyfExA5PsHSx0rdFgnADMYQi+Zkeiv4/ZxGYUWd0nLs63Ha0ksmp8VMxAIC0wtCFxMos7Lt3PszJssG/E6w==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/scope-manager/-/scope-manager-7.8.0.tgz",
+      "integrity": "sha512-viEmZ1LmwsGcnr85gIq+FCYI7nO90DVbE37/ll51hjv9aG+YZMb4WDE2fyWpUR4O/UrhGRpYXK/XajcGTk2B8g==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/types": "7.6.0",
-        "@typescript-eslint/visitor-keys": "7.6.0"
+        "@typescript-eslint/types": "7.8.0",
+        "@typescript-eslint/visitor-keys": "7.8.0"
       },
       "engines": {
         "node": "^18.18.0 || >=20.0.0"
@@ -790,13 +826,13 @@
       }
     },
     "node_modules/@typescript-eslint/type-utils": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/type-utils/-/type-utils-7.6.0.tgz",
-      "integrity": "sha512-NxAfqAPNLG6LTmy7uZgpK8KcuiS2NZD/HlThPXQRGwz6u7MDBWRVliEEl1Gj6U7++kVJTpehkhZzCJLMK66Scw==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/type-utils/-/type-utils-7.8.0.tgz",
+      "integrity": "sha512-H70R3AefQDQpz9mGv13Uhi121FNMh+WEaRqcXTX09YEDky21km4dV1ZXJIp8QjXc4ZaVkXVdohvWDzbnbHDS+A==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/typescript-estree": "7.6.0",
-        "@typescript-eslint/utils": "7.6.0",
+        "@typescript-eslint/typescript-estree": "7.8.0",
+        "@typescript-eslint/utils": "7.8.0",
         "debug": "^4.3.4",
         "ts-api-utils": "^1.3.0"
       },
@@ -817,9 +853,9 @@
       }
     },
     "node_modules/@typescript-eslint/types": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-7.6.0.tgz",
-      "integrity": "sha512-h02rYQn8J+MureCvHVVzhl69/GAfQGPQZmOMjG1KfCl7o3HtMSlPaPUAPu6lLctXI5ySRGIYk94clD/AUMCUgQ==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/types/-/types-7.8.0.tgz",
+      "integrity": "sha512-wf0peJ+ZGlcH+2ZS23aJbOv+ztjeeP8uQ9GgwMJGVLx/Nj9CJt17GWgWWoSmoRVKAX2X+7fzEnAjxdvK2gqCLw==",
       "dev": true,
       "engines": {
         "node": "^18.18.0 || >=20.0.0"
@@ -830,13 +866,13 @@
       }
     },
     "node_modules/@typescript-eslint/typescript-estree": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/typescript-estree/-/typescript-estree-7.6.0.tgz",
-      "integrity": "sha512-+7Y/GP9VuYibecrCQWSKgl3GvUM5cILRttpWtnAu8GNL9j11e4tbuGZmZjJ8ejnKYyBRb2ddGQ3rEFCq3QjMJw==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/typescript-estree/-/typescript-estree-7.8.0.tgz",
+      "integrity": "sha512-5pfUCOwK5yjPaJQNy44prjCwtr981dO8Qo9J9PwYXZ0MosgAbfEMB008dJ5sNo3+/BN6ytBPuSvXUg9SAqB0dg==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/types": "7.6.0",
-        "@typescript-eslint/visitor-keys": "7.6.0",
+        "@typescript-eslint/types": "7.8.0",
+        "@typescript-eslint/visitor-keys": "7.8.0",
         "debug": "^4.3.4",
         "globby": "^11.1.0",
         "is-glob": "^4.0.3",
@@ -882,17 +918,17 @@
       }
     },
     "node_modules/@typescript-eslint/utils": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/utils/-/utils-7.6.0.tgz",
-      "integrity": "sha512-x54gaSsRRI+Nwz59TXpCsr6harB98qjXYzsRxGqvA5Ue3kQH+FxS7FYU81g/omn22ML2pZJkisy6Q+ElK8pBCA==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/utils/-/utils-7.8.0.tgz",
+      "integrity": "sha512-L0yFqOCflVqXxiZyXrDr80lnahQfSOfc9ELAAZ75sqicqp2i36kEZZGuUymHNFoYOqxRT05up760b4iGsl02nQ==",
       "dev": true,
       "dependencies": {
         "@eslint-community/eslint-utils": "^4.4.0",
         "@types/json-schema": "^7.0.15",
         "@types/semver": "^7.5.8",
-        "@typescript-eslint/scope-manager": "7.6.0",
-        "@typescript-eslint/types": "7.6.0",
-        "@typescript-eslint/typescript-estree": "7.6.0",
+        "@typescript-eslint/scope-manager": "7.8.0",
+        "@typescript-eslint/types": "7.8.0",
+        "@typescript-eslint/typescript-estree": "7.8.0",
         "semver": "^7.6.0"
       },
       "engines": {
@@ -907,12 +943,12 @@
       }
     },
     "node_modules/@typescript-eslint/visitor-keys": {
-      "version": "7.6.0",
-      "resolved": "https://registry.npmjs.org/@typescript-eslint/visitor-keys/-/visitor-keys-7.6.0.tgz",
-      "integrity": "sha512-4eLB7t+LlNUmXzfOu1VAIAdkjbu5xNSerURS9X/S5TUKWFRpXRQZbmtPqgKmYx8bj3J0irtQXSiWAOY82v+cgw==",
+      "version": "7.8.0",
+      "resolved": "https://registry.npmjs.org/@typescript-eslint/visitor-keys/-/visitor-keys-7.8.0.tgz",
+      "integrity": "sha512-q4/gibTNBQNA0lGyYQCmWRS5D15n8rXh4QjK3KV+MBPlTYHpfBUT3D3PaPR/HeNiI9W6R7FvlkcGhNyAoP+caA==",
       "dev": true,
       "dependencies": {
-        "@typescript-eslint/types": "7.6.0",
+        "@typescript-eslint/types": "7.8.0",
         "eslint-visitor-keys": "^3.4.3"
       },
       "engines": {
@@ -2650,9 +2686,9 @@
       }
     },
     "node_modules/orb-billing": {
-      "version": "2.2.0",
-      "resolved": "https://registry.npmjs.org/orb-billing/-/orb-billing-2.2.0.tgz",
-      "integrity": "sha512-GLBEgnT+tcWohTp4hkAGsjNEOLoIWZTUw36sNW12UaPH4ojbABcQADQGuK1aRmn9GSVQci1Kpt7pyj5/xvN67A==",
+      "version": "2.3.0",
+      "resolved": "https://registry.npmjs.org/orb-billing/-/orb-billing-2.3.0.tgz",
+      "integrity": "sha512-mh1UCTncIWu159fTGzkg5Dhg6ZLrwiKhz1F4ixIswp9UaCG0bnugd5gHqSa39olX70+Sr1umwgLatyhBsxQZRA==",
       "dependencies": {
         "@types/node": "^18.11.18",
         "@types/node-fetch": "^2.6.4",
@@ -2884,9 +2920,9 @@
       }
     },
     "node_modules/pino-abstract-transport": {
-      "version": "1.1.0",
-      "resolved": "https://registry.npmjs.org/pino-abstract-transport/-/pino-abstract-transport-1.1.0.tgz",
-      "integrity": "sha512-lsleG3/2a/JIWUtf9Q5gUNErBqwIu1tUKTT3dUzaf5DySw9ra1wcqKjJjLX1VTY64Wk1eEOYsVGSaGfCK85ekA==",
+      "version": "1.2.0",
+      "resolved": "https://registry.npmjs.org/pino-abstract-transport/-/pino-abstract-transport-1.2.0.tgz",
+      "integrity": "sha512-Guhh8EZfPCfH+PMXAb6rKOjGQEoy0xlAIn+irODG5kgfYV+BQ0rGYYWTIel3P5mmyXqkYkPmdIkywsn6QKUR1Q==",
       "dependencies": {
         "readable-stream": "^4.0.0",
         "split2": "^4.0.0"
@@ -3464,9 +3500,9 @@
       "dev": true
     },
     "node_modules/thread-stream": {
-      "version": "2.4.1",
-      "resolved": "https://registry.npmjs.org/thread-stream/-/thread-stream-2.4.1.tgz",
-      "integrity": "sha512-d/Ex2iWd1whipbT681JmTINKw0ZwOUBZm7+Gjs64DHuX34mmw8vJL2bFAaNacaW72zYiTJxSHi5abUuOi5nsfg==",
+      "version": "2.7.0",
+      "resolved": "https://registry.npmjs.org/thread-stream/-/thread-stream-2.7.0.tgz",
+      "integrity": "sha512-qQiRWsU/wvNolI6tbbCKd9iKaTnCXsTwVxhhKM6nctPdujTyztjlbUkUTUymidWcMnZ5pWR0ej4a0tjsW021vw==",
       "dependencies": {
         "real-require": "^0.2.0"
       }
@@ -3640,26 +3676,26 @@
       }
     },
     "node_modules/turbo": {
-      "version": "1.13.2",
-      "resolved": "https://registry.npmjs.org/turbo/-/turbo-1.13.2.tgz",
-      "integrity": "sha512-rX/d9f4MgRT3yK6cERPAkfavIxbpBZowDQpgvkYwGMGDQ0Nvw1nc0NVjruE76GrzXQqoxR1UpnmEP54vBARFHQ==",
+      "version": "1.13.3",
+      "resolved": "https://registry.npmjs.org/turbo/-/turbo-1.13.3.tgz",
+      "integrity": "sha512-n17HJv4F4CpsYTvKzUJhLbyewbXjq1oLCi90i5tW1TiWDz16ML1eDG7wi5dHaKxzh5efIM56SITnuVbMq5dk4g==",
       "dev": true,
       "bin": {
         "turbo": "bin/turbo"
       },
       "optionalDependencies": {
-        "turbo-darwin-64": "1.13.2",
-        "turbo-darwin-arm64": "1.13.2",
-        "turbo-linux-64": "1.13.2",
-        "turbo-linux-arm64": "1.13.2",
-        "turbo-windows-64": "1.13.2",
-        "turbo-windows-arm64": "1.13.2"
+        "turbo-darwin-64": "1.13.3",
+        "turbo-darwin-arm64": "1.13.3",
+        "turbo-linux-64": "1.13.3",
+        "turbo-linux-arm64": "1.13.3",
+        "turbo-windows-64": "1.13.3",
+        "turbo-windows-arm64": "1.13.3"
       }
     },
     "node_modules/turbo-darwin-64": {
-      "version": "1.13.2",
-      "resolved": "https://registry.npmjs.org/turbo-darwin-64/-/turbo-darwin-64-1.13.2.tgz",
-      "integrity": "sha512-CCSuD8CfmtncpohCuIgq7eAzUas0IwSbHfI8/Q3vKObTdXyN8vAo01gwqXjDGpzG9bTEVedD0GmLbD23dR0MLA==",
+      "version": "1.13.3",
+      "resolved": "https://registry.npmjs.org/turbo-darwin-64/-/turbo-darwin-64-1.13.3.tgz",
+      "integrity": "sha512-glup8Qx1qEFB5jerAnXbS8WrL92OKyMmg5Hnd4PleLljAeYmx+cmmnsmLT7tpaVZIN58EAAwu8wHC6kIIqhbWA==",
       "cpu": [
         "x64"
       ],
@@ -3670,9 +3706,9 @@
       ]
     },
     "node_modules/turbo-darwin-arm64": {
-      "version": "1.13.2",
-      "resolved": "https://registry.npmjs.org/turbo-darwin-arm64/-/turbo-darwin-arm64-1.13.2.tgz",
-      "integrity": "sha512-0HySm06/D2N91rJJ89FbiI/AodmY8B3WDSFTVEpu2+8spUw7hOJ8okWOT0e5iGlyayUP9gr31eOeL3VFZkpfCw==",
+      "version": "1.13.3",
+      "resolved": "https://registry.npmjs.org/turbo-darwin-arm64/-/turbo-darwin-arm64-1.13.3.tgz",
+      "integrity": "sha512-/np2xD+f/+9qY8BVtuOQXRq5f9LehCFxamiQnwdqWm5iZmdjygC5T3uVSYuagVFsZKMvX3ycySwh8dylGTl6lg==",
       "cpu": [
         "arm64"
       ],
@@ -3683,9 +3719,9 @@
       ]
     },
     "node_modules/turbo-linux-64": {
-      "version": "1.13.2",
-      "resolved": "https://registry.npmjs.org/turbo-linux-64/-/turbo-linux-64-1.13.2.tgz",
-      "integrity": "sha512-7HnibgbqZrjn4lcfIouzlPu8ZHSBtURG4c7Bedu7WJUDeZo+RE1crlrQm8wuwO54S0siYqUqo7GNHxu4IXbioQ==",
+      "version": "1.13.3",
+      "resolved": "https://registry.npmjs.org/turbo-linux-64/-/turbo-linux-64-1.13.3.tgz",
+      "integrity": "sha512-G+HGrau54iAnbXLfl+N/PynqpDwi/uDzb6iM9hXEDG+yJnSJxaHMShhOkXYJPk9offm9prH33Khx2scXrYVW1g==",
       "cpu": [
         "x64"
       ],
@@ -3696,9 +3732,9 @@
       ]
     },
     "node_modules/turbo-linux-arm64": {
-      "version": "1.13.2",
-      "resolved": "https://registry.npmjs.org/turbo-linux-arm64/-/turbo-linux-arm64-1.13.2.tgz",
-      "integrity": "sha512-sUq4dbpk6SNKg/Hkwn256Vj2AEYSQdG96repio894h5/LEfauIK2QYiC/xxAeW3WBMc6BngmvNyURIg7ltrePg==",
+      "version": "1.13.3",
+      "resolved": "https://registry.npmjs.org/turbo-linux-arm64/-/turbo-linux-arm64-1.13.3.tgz",
+      "integrity": "sha512-qWwEl5VR02NqRyl68/3pwp3c/olZuSp+vwlwrunuoNTm6JXGLG5pTeme4zoHNnk0qn4cCX7DFrOboArlYxv0wQ==",
       "cpu": [
         "arm64"
       ],
@@ -3709,9 +3745,9 @@
       ]
     },
     "node_modules/turbo-windows-64": {
-      "version": "1.13.2",
-      "resolved": "https://registry.npmjs.org/turbo-windows-64/-/turbo-windows-64-1.13.2.tgz",
-      "integrity": "sha512-DqzhcrciWq3dpzllJR2VVIyOhSlXYCo4mNEWl98DJ3FZ08PEzcI3ceudlH6F0t/nIcfSItK1bDP39cs7YoZHEA==",
+      "version": "1.13.3",
+      "resolved": "https://registry.npmjs.org/turbo-windows-64/-/turbo-windows-64-1.13.3.tgz",
+      "integrity": "sha512-Nudr4bRChfJzBPzEmpVV85VwUYRCGKecwkBFpbp2a4NtrJ3+UP1VZES653ckqCu2FRyRuS0n03v9euMbAvzH+Q==",
       "cpu": [
         "x64"
       ],
@@ -3722,9 +3758,9 @@
       ]
     },
     "node_modules/turbo-windows-arm64": {
-      "version": "1.13.2",
-      "resolved": "https://registry.npmjs.org/turbo-windows-arm64/-/turbo-windows-arm64-1.13.2.tgz",
-      "integrity": "sha512-WnPMrwfCXxK69CdDfS1/j2DlzcKxSmycgDAqV0XCYpK/812KB0KlvsVAt5PjEbZGXkY88pCJ1BLZHAjF5FcbqA==",
+      "version": "1.13.3",
+      "resolved": "https://registry.npmjs.org/turbo-windows-arm64/-/turbo-windows-arm64-1.13.3.tgz",
+      "integrity": "sha512-ouJCgsVLd3icjRLmRvHQDDZnmGzT64GBupM1Y+TjtYn2LVaEBoV6hicFy8x5DUpnqdLy+YpCzRMkWlwhmkX7sQ==",
       "cpu": [
         "arm64"
       ],
@@ -3910,7 +3946,7 @@
       "version": "0.0.0",
       "license": "MIT",
       "dependencies": {
-        "orb-billing": "^2.2.0",
+        "orb-billing": "^2.3.0",
         "pg": "^8.11.5",
         "yesql": "^7.0.0"
       },
diff --git a/package.json b/package.json
index 715d281..56b0754 100644
--- a/package.json
+++ b/package.json
@@ -20,12 +20,12 @@
     "apps/*"
   ],
   "devDependencies": {
-    "@typescript-eslint/eslint-plugin": "^7.6.0",
-    "@typescript-eslint/parser": "^7.6.0",
+    "@typescript-eslint/eslint-plugin": "7.8.0",
+    "@typescript-eslint/parser": "^7.8.0",
     "eslint": "^8.57.0",
     "eslint-config-prettier": "^9.1.0",
     "prettier": "^4.0.0-alpha.8",
-    "turbo": "^1.13.2",
+    "turbo": "^1.13.3",
     "typescript": "^5.4.5"
   },
   "repository": {
diff --git a/packages/orb-sync-lib/package.json b/packages/orb-sync-lib/package.json
index f60c6c0..f6a28f4 100644
--- a/packages/orb-sync-lib/package.json
+++ b/packages/orb-sync-lib/package.json
@@ -16,7 +16,7 @@
   "author": "Supabase",
   "license": "MIT",
   "dependencies": {
-    "orb-billing": "^2.2.0",
+    "orb-billing": "^2.3.0",
     "pg": "^8.11.5",
     "yesql": "^7.0.0"
   },
diff --git a/packages/orb-sync-lib/src/database/postgres.ts b/packages/orb-sync-lib/src/database/postgres.ts
index 67f3d58..bdd891b 100644
--- a/packages/orb-sync-lib/src/database/postgres.ts
+++ b/packages/orb-sync-lib/src/database/postgres.ts
@@ -19,6 +19,8 @@ export class PostgresClient {
       [Key: string]: any; // eslint-disable-line @typescript-eslint/no-explicit-any
     },
   >(entries: T[], table: string, tableSchema: JsonSchema): Promise<T[]> {
+    if (!entries.length) return [];
+
     const queries: Promise<pg.QueryResult<T>>[] = [];
 
     entries.forEach((entry) => {
diff --git a/packages/orb-sync-lib/src/orb-sync.ts b/packages/orb-sync-lib/src/orb-sync.ts
index 2954ee8..ba72540 100644
--- a/packages/orb-sync-lib/src/orb-sync.ts
+++ b/packages/orb-sync-lib/src/orb-sync.ts
@@ -1,11 +1,21 @@
 import Orb from 'orb-billing';
 import type { HeadersLike } from 'orb-billing/core';
-import type { CreditNoteWebhook, CustomerWebhook, InvoiceWebhook, OrbWebhook, SubscriptionWebhook } from './types';
+import type {
+  CreditNoteWebhook,
+  CreditNotesFetchParams,
+  CustomerWebhook,
+  CustomersFetchParams,
+  InvoiceWebhook,
+  InvoicesFetchParams,
+  OrbWebhook,
+  SubscriptionWebhook,
+  SubscriptionsFetchParams,
+} from './types';
 import { PostgresClient } from './database/postgres';
-import { syncCustomers } from './sync/customers';
-import { syncSubscriptions } from './sync/subscriptions';
-import { syncInvoices } from './sync/invoices';
-import { syncCreditNotes } from './sync/credit_notes';
+import { fetchAndSyncCustomers, syncCustomers } from './sync/customers';
+import { fetchAndSyncSubscriptions, syncSubscriptions } from './sync/subscriptions';
+import { fetchAndSyncInvoices, syncInvoices } from './sync/invoices';
+import { fetchAndSyncCreditNotes, syncCreditNotes } from './sync/credit_notes';
 
 export type OrbSyncConfig = {
   databaseUrl: string;
@@ -35,7 +45,27 @@ export class OrbSync {
     });
   }
 
-  async sync(payload: string, headers: HeadersLike | undefined) {
+  async sync(
+    entity: 'invoices' | 'customers' | 'credit_notes' | 'subscriptions',
+    params: InvoicesFetchParams | CustomersFetchParams | CreditNotesFetchParams | SubscriptionsFetchParams
+  ): Promise<number> {
+    switch (entity) {
+      case 'invoices': {
+        return fetchAndSyncInvoices(this.postgresClient, this.orb, params as InvoicesFetchParams);
+      }
+      case 'credit_notes': {
+        return fetchAndSyncCreditNotes(this.postgresClient, this.orb, params as CreditNotesFetchParams);
+      }
+      case 'customers': {
+        return fetchAndSyncCustomers(this.postgresClient, this.orb, params as CustomersFetchParams);
+      }
+      case 'subscriptions': {
+        return fetchAndSyncSubscriptions(this.postgresClient, this.orb, params as SubscriptionsFetchParams);
+      }
+    }
+  }
+
+  async processWebhook(payload: string, headers: HeadersLike | undefined) {
     if (this.config.verifyWebhookSignature ?? true) {
       this.orb.webhooks.verifySignature(payload, headers || {}, this.config.orbWebhookSecret);
     }
diff --git a/packages/orb-sync-lib/src/sync/credit_notes.ts b/packages/orb-sync-lib/src/sync/credit_notes.ts
index aea3d77..000d1bf 100644
--- a/packages/orb-sync-lib/src/sync/credit_notes.ts
+++ b/packages/orb-sync-lib/src/sync/credit_notes.ts
@@ -1,6 +1,8 @@
 import type { CreditNote } from 'orb-billing/resources/credit-notes';
 import type { PostgresClient } from '../database/postgres';
 import { creditNoteSchema } from '../schemas/credit_note';
+import type Orb from 'orb-billing';
+import { CreditNotesFetchParams } from '../types';
 
 const TABLE = 'credit_notes';
 
@@ -14,3 +16,19 @@ export async function syncCreditNotes(postgresClient: PostgresClient, creditNote
     creditNoteSchema
   );
 }
+
+export async function fetchAndSyncCreditNotes(
+  postgresClient: PostgresClient,
+  orbClient: Orb,
+  params: CreditNotesFetchParams
+): Promise<number> {
+  const creditNotes = [];
+
+  for await (const creditNote of orbClient.creditNotes.list({ limit: params.limit || 100 })) {
+    creditNotes.push(creditNote);
+  }
+
+  await syncCreditNotes(postgresClient, creditNotes);
+
+  return creditNotes.length;
+}
diff --git a/packages/orb-sync-lib/src/sync/customers.ts b/packages/orb-sync-lib/src/sync/customers.ts
index 952bf71..87a8f86 100644
--- a/packages/orb-sync-lib/src/sync/customers.ts
+++ b/packages/orb-sync-lib/src/sync/customers.ts
@@ -1,9 +1,33 @@
+import type Orb from 'orb-billing';
 import type { Customer } from 'orb-billing/resources/customers/customers';
 import type { PostgresClient } from '../database/postgres';
 import { customerSchema } from '../schemas/customer';
+import { CustomersFetchParams } from '../types';
 
 const TABLE = 'customers';
 
 export async function syncCustomers(postgresClient: PostgresClient, customers: Customer[]) {
   return postgresClient.upsertMany(customers, TABLE, customerSchema);
 }
+
+export async function fetchAndSyncCustomers(
+  postgresClient: PostgresClient,
+  orbClient: Orb,
+  params: CustomersFetchParams
+): Promise<number> {
+  const customers = [];
+
+  for await (const customer of orbClient.customers.list({
+    limit: params.limit || 100,
+    'created_at[gt]': params.createdAtGt,
+    'created_at[gte]': params.createdAtGte,
+    'created_at[lt]': params.createdAtLt,
+    'created_at[lte]': params.createdAtLte,
+  })) {
+    customers.push(customer);
+  }
+
+  await syncCustomers(postgresClient, customers);
+
+  return customers.length;
+}
diff --git a/packages/orb-sync-lib/src/sync/invoices.ts b/packages/orb-sync-lib/src/sync/invoices.ts
index c6f5e63..8c6ad76 100644
--- a/packages/orb-sync-lib/src/sync/invoices.ts
+++ b/packages/orb-sync-lib/src/sync/invoices.ts
@@ -1,6 +1,8 @@
+import type Orb from 'orb-billing';
 import type { Invoice } from 'orb-billing/resources/invoices';
 import type { PostgresClient } from '../database/postgres';
 import { invoiceSchema } from '../schemas/invoice';
+import { InvoicesFetchParams } from '../types';
 
 const TABLE = 'invoices';
 
@@ -15,3 +17,25 @@ export async function syncInvoices(postgresClient: PostgresClient, invoices: Inv
     invoiceSchema
   );
 }
+
+export async function fetchAndSyncInvoices(
+  postgresClient: PostgresClient,
+  orbClient: Orb,
+  params: InvoicesFetchParams
+): Promise<number> {
+  const invoices = [];
+
+  for await (const invoice of orbClient.invoices.list({
+    limit: params.limit || 100,
+    'invoice_date[gt]': params.createdAtGt,
+    'invoice_date[gte]': params.createdAtGte,
+    'invoice_date[lt]': params.createdAtLt,
+    'invoice_date[lte]': params.createdAtLte,
+  })) {
+    invoices.push(invoice);
+  }
+
+  await syncInvoices(postgresClient, invoices);
+
+  return invoices.length;
+}
diff --git a/packages/orb-sync-lib/src/sync/subscriptions.ts b/packages/orb-sync-lib/src/sync/subscriptions.ts
index 6c1bf4d..5481d5b 100644
--- a/packages/orb-sync-lib/src/sync/subscriptions.ts
+++ b/packages/orb-sync-lib/src/sync/subscriptions.ts
@@ -1,6 +1,8 @@
+import type Orb from 'orb-billing';
 import type { Subscription } from 'orb-billing/resources/subscriptions';
 import type { PostgresClient } from '../database/postgres';
 import { subscriptionSchema } from '../schemas/subscription';
+import { SubscriptionsFetchParams } from '../types';
 
 const TABLE = 'subscriptions';
 
@@ -15,3 +17,25 @@ export async function syncSubscriptions(postgresClient: PostgresClient, subscrip
     subscriptionSchema
   );
 }
+
+export async function fetchAndSyncSubscriptions(
+  postgresClient: PostgresClient,
+  orbClient: Orb,
+  params: SubscriptionsFetchParams
+): Promise<number> {
+  const subscriptions = [];
+
+  for await (const invoice of orbClient.subscriptions.list({
+    limit: params.limit || 100,
+    'created_at[gt]': params.createdAtGt,
+    'created_at[gte]': params.createdAtGte,
+    'created_at[lt]': params.createdAtLt,
+    'created_at[lte]': params.createdAtLte,
+  })) {
+    subscriptions.push(invoice);
+  }
+
+  await syncSubscriptions(postgresClient, subscriptions);
+
+  return subscriptions.length;
+}
diff --git a/packages/orb-sync-lib/src/types.ts b/packages/orb-sync-lib/src/types.ts
index 4cac33c..2cee009 100644
--- a/packages/orb-sync-lib/src/types.ts
+++ b/packages/orb-sync-lib/src/types.ts
@@ -56,3 +56,31 @@ export type SubscriptionWebhook = {
 export type CreditNoteWebhook = {
   credit_note: CreditNote;
 } & OrbWebhook;
+
+export type SubscriptionsFetchParams = {
+  limit?: number;
+  createdAtGt?: string | null;
+  createdAtGte?: string | null;
+  createdAtLt?: string | null;
+  createdAtLte?: string | null;
+};
+
+export type InvoicesFetchParams = {
+  limit?: number;
+  createdAtGt?: string | null;
+  createdAtGte?: string | null;
+  createdAtLt?: string | null;
+  createdAtLte?: string | null;
+};
+
+export type CustomersFetchParams = {
+  limit?: number;
+  createdAtGt?: string | null;
+  createdAtGte?: string | null;
+  createdAtLt?: string | null;
+  createdAtLte?: string | null;
+};
+
+export type CreditNotesFetchParams = {
+  limit?: number;
+};