feat: add incremental catalog ingestion (cursor-resumable, namespace model)

.changeset/fetch-all-pages-cursor-forwarding.md

@@ -0,0 +1,9 @@
+---
+'@openchoreo/backstage-plugin-catalog-backend-module': patch
+---
+
+Fix three `fetchAllPages` call sites in the scheduled entity provider that
+ignored the pagination cursor (workflowplanes, observabilityplanes and
+deploymentpipelines). Each closure now forwards the cursor and requests
+`limit: 100`, so namespaces with more than one page of these resources are
+fully ingested instead of silently stopping after the first page.

.changeset/fetch-all-pages-hardening.md

@@ -0,0 +1,18 @@
+---
+'@openchoreo/openchoreo-client-node': minor
+---
+
+Harden `fetchAllPages` with an optional options bag: `maxPages` caps how
+many pages are fetched (throwing instead of silently truncating, and kept
+opt-in with no default so existing callers see no new failure modes),
+`timeoutMs` gives the whole run a wall-clock budget (chosen as a finite
+60s default so unbounded pagination cannot hang a backend, with `0` as
+the escape hatch that disables it), and `signal` lets callers abort the
+run at entry and between pages. The helper now also detects stuck
+cursors (a page returning the same non-empty cursor it was fetched with)
+and malformed page responses (a nullish page or a missing `items`
+array), throwing descriptive errors that name the page index, cursor,
+and collected item count. `PaginatedResponse` and the new
+`FetchAllPagesOptions` type are now exported. Behavior is unchanged for
+callers that pass only `fetchPage`, apart from the new default timeout
+kicking in.

.changeset/incremental-ingestion-package.md

@@ -0,0 +1,14 @@
+---
+'@openchoreo/backstage-plugin-catalog-backend-module-openchoreo-incremental': minor
+---
+
+Add `@openchoreo/backstage-plugin-catalog-backend-module-openchoreo-incremental`
+— a new package providing burst-based, cursor-resumable incremental catalog
+ingestion on the namespace model (namespaces → projects → components). The
+module is config-gated off by default via
+`openchoreo.features.incrementalIngestion.enabled`; when the flag is true the
+scheduled full-sync `OpenChoreoEntityProvider` in the sibling catalog module
+stands down so entities are not double-ingested. Ingestion state persists
+through knex migrations on the plugin database (SQLite in dev, Postgres in
+prod), applied lazily on first enable. The package depends on the hardened
+`fetchAllPages`-era `@openchoreo/openchoreo-client-node` client (PR-A).

app-config.local.yaml.example

@@ -77,6 +77,17 @@ openchoreo:
       enabled: true    # Set to false to hide Secrets settings tab
     assistant:
       enabled: true   # Opt-in. Set to true after deploying perch-agent and OPENCHOREO_PERCH_AGENT_URL
+    incrementalIngestion:
+      enabled: false  # Opt-in; replaces the scheduled full-sync provider when true
+
+  # Incremental ingestion tuning (optional). Only read when
+  # features.incrementalIngestion.enabled is true; uncomment to override
+  # the built-in defaults shown here.
+  # incremental:
+  #   burstLength: 10 # seconds per ingestion burst (default: 10)
+  #   burstInterval: 30 # seconds between bursts (default: 30)
+  #   chunkSize: 100 # entities per page; the API caps page size at 100 (default: 100)
+  #   restLength: 30 # minutes of rest after a completed ingestion (default: 30)
 
 # Thunder IDP configuration (k3d cluster)
 thunder:

app-config.production.yaml

@@ -219,6 +219,14 @@ openchoreo:
   events:
     enabled: ${OPENCHOREO_EVENTS_ENABLED}
 
+  # Incremental ingestion tuning. Only read when
+  # openchoreo.features.incrementalIngestion.enabled is true.
+  incremental:
+    burstLength: 16 # seconds per ingestion burst
+    burstInterval: 8 # seconds between bursts
+    chunkSize: 100 # entities per page; the API caps page size at 100
+    restLength: 60 # minutes of rest after a completed ingestion
+
   # Observability tuning.
   # Hard cap (seconds) on a single wirelogs SSE stream before the backend ends
   # it; the UI shows soft warnings at ~1/3 and ~2/3 of this value and a toast
@@ -254,6 +262,11 @@ openchoreo:
     # LLM API key Secret to be deployed.
     assistant:
       enabled: ${OPENCHOREO_FEATURES_ASSISTANT_ENABLED}
+    # Incremental catalog ingestion - burst-based, cursor-resumable ingestion.
+    # When enabled, replaces the scheduled full-sync entity provider.
+    # Environment variable: OPENCHOREO_FEATURES_INCREMENTAL_INGESTION_ENABLED
+    incrementalIngestion:
+      enabled: ${OPENCHOREO_FEATURES_INCREMENTAL_INGESTION_ENABLED}
     # Secret Management configuration
     # When disabled, hides the Secrets settings tab and disables secret APIs
     secretManagement:

app-config.yaml

@@ -195,6 +195,14 @@ openchoreo:
     frequency: 300 # seconds between runs (default: 300, the periodic full sync acts as a safety net behind the event-forwarder's real-time deltas)
     timeout: 120 # seconds for timeout (default: 120)
 
+  # Incremental ingestion tuning. Only read when
+  # openchoreo.features.incrementalIngestion.enabled is true.
+  incremental:
+    burstLength: 16 # seconds per ingestion burst
+    burstInterval: 8 # seconds between bursts
+    chunkSize: 100 # entities per page; the API caps page size at 100
+    restLength: 60 # minutes of rest after a completed ingestion
+
   # Observability tuning (optional).
   # Hard cap (seconds) on a single wirelogs SSE stream before the backend ends
   # it; the UI shows soft warnings at ~1/3 and ~2/3 of this value and a toast
@@ -235,6 +243,11 @@ openchoreo:
     # FailedBuildSnackbar, and the build / logs / component-create launchers.
     assistant:
       enabled: ${OPENCHOREO_FEATURES_ASSISTANT_ENABLED}
+    # Incremental catalog ingestion - burst-based, cursor-resumable ingestion.
+    # When enabled, replaces the scheduled full-sync entity provider.
+    # Environment variable: OPENCHOREO_FEATURES_INCREMENTAL_INGESTION_ENABLED
+    incrementalIngestion:
+      enabled: ${OPENCHOREO_FEATURES_INCREMENTAL_INGESTION_ENABLED}
 
   # Component Type Mappings (optional - defaults provided)
   # Maps OpenChoreo component types to Backstage page variants

packages/openchoreo-client-node/src/index.ts

@@ -52,7 +52,11 @@ export {
 } from './resource-utils';
 
 // Export pagination utilities (new API)
-export { fetchAllPages } from './pagination-utils';
+export {
+  fetchAllPages,
+  type FetchAllPagesOptions,
+  type PaginatedResponse,
+} from './pagination-utils';
 
 // Export generated types as namespaces
 export * as OpenChoreoAPI from './generated/openchoreo';

packages/openchoreo-client-node/src/pagination-utils.test.ts

@@ -1,3 +1,4 @@
+import { getEventListeners } from 'events';
 import { fetchAllPages } from './pagination-utils';
 
 describe('fetchAllPages', () => {
@@ -99,3 +100,255 @@ describe('fetchAllPages', () => {
     expect(fetchPage).toHaveBeenCalledTimes(2);
   });
 });
+
+async function captureError(run: () => Promise<unknown>): Promise<Error> {
+  try {
+    await run();
+  } catch (error) {
+    return error as Error;
+  }
+  throw new Error('Expected the operation to reject, but it resolved');
+}
+
+describe('fetchAllPages hardening', () => {
+  it('throws when fetchPage resolves undefined, naming the page index', async () => {
+    const fetchPage = jest.fn().mockResolvedValue(undefined);
+
+    const error = await captureError(() => fetchAllPages(fetchPage));
+
+    expect(error).toBeInstanceOf(Error);
+    expect(error.message).toContain('page 0');
+    expect(error.message).toContain('undefined');
+  });
+
+  it('throws when a page has no items array', async () => {
+    const fetchPage = jest.fn().mockResolvedValue({});
+
+    const error = await captureError(() => fetchAllPages(fetchPage));
+
+    expect(error).toBeInstanceOf(Error);
+    expect(error.message).toContain('page 0');
+    expect(error.message).toContain('items');
+  });
+
+  it('throws when the next cursor repeats the cursor used to fetch the page', async () => {
+    const fetchPage = jest
+      .fn()
+      .mockResolvedValueOnce({
+        items: [{ id: 1 }],
+        pagination: { nextCursor: 'cursor-1' },
+      })
+      .mockResolvedValueOnce({
+        items: [{ id: 2 }],
+        pagination: { nextCursor: 'cursor-1' },
+      });
+
+    const error = await captureError(() => fetchAllPages(fetchPage));
+
+    expect(error).toBeInstanceOf(Error);
+    expect(error.message).toContain('stuck');
+    // The error must name the cursor value that is not advancing.
+    expect(error.message).toContain('"cursor-1"');
+    expect(fetchPage).toHaveBeenCalledTimes(2);
+  });
+
+  it('stops when nextCursor is null', async () => {
+    const fetchPage = jest
+      .fn()
+      .mockResolvedValueOnce({
+        items: [{ id: 1 }],
+        pagination: { nextCursor: 'cursor-1' },
+      })
+      .mockResolvedValueOnce({
+        items: [{ id: 2 }],
+        pagination: { nextCursor: null },
+      });
+
+    const result = await fetchAllPages(fetchPage);
+
+    expect(result).toEqual([{ id: 1 }, { id: 2 }]);
+    expect(fetchPage).toHaveBeenCalledTimes(2);
+  });
+
+  it('stops when nextCursor is an empty string', async () => {
+    const fetchPage = jest
+      .fn()
+      .mockResolvedValueOnce({
+        items: [{ id: 1 }],
+        pagination: { nextCursor: 'cursor-1' },
+      })
+      .mockResolvedValueOnce({
+        items: [{ id: 2 }],
+        pagination: { nextCursor: '' },
+      });
+
+    const result = await fetchAllPages(fetchPage);
+
+    expect(result).toEqual([{ id: 1 }, { id: 2 }]);
+    expect(fetchPage).toHaveBeenCalledTimes(2);
+  });
+
+  it('throws when fetching more pages than maxPages allows', async () => {
+    const fetchPage = jest
+      .fn()
+      .mockResolvedValueOnce({
+        items: [{ id: 1 }, { id: 2 }],
+        pagination: { nextCursor: 'cursor-1' },
+      })
+      .mockResolvedValueOnce({
+        items: [{ id: 3 }, { id: 4 }],
+        pagination: { nextCursor: 'cursor-2' },
+      })
+      .mockResolvedValueOnce({
+        items: [{ id: 5 }, { id: 6 }],
+        pagination: {},
+      });
+
+    const error = await captureError(() =>
+      fetchAllPages(fetchPage, { maxPages: 2 }),
+    );
+
+    expect(error).toBeInstanceOf(Error);
+    expect(error.message).toContain('maxPages of 2');
+    expect(error.message).toContain('6 items');
+    expect(fetchPage).toHaveBeenCalledTimes(3);
+  });
+
+  it('does not throw when the page count exactly reaches maxPages', async () => {
+    const fetchPage = jest
+      .fn()
+      .mockResolvedValueOnce({
+        items: [{ id: 1 }],
+        pagination: { nextCursor: 'cursor-1' },
+      })
+      .mockResolvedValueOnce({
+        items: [{ id: 2 }],
+        pagination: {},
+      });
+
+    const result = await fetchAllPages(fetchPage, { maxPages: 2 });
+
+    expect(result).toEqual([{ id: 1 }, { id: 2 }]);
+    expect(fetchPage).toHaveBeenCalledTimes(2);
+  });
+});
+
+describe('fetchAllPages timeout', () => {
+  beforeEach(() => {
+    jest.useFakeTimers();
+  });
+
+  afterEach(() => {
+    if (jest.getTimerCount() !== 0) {
+      throw new Error(`leaked ${jest.getTimerCount()} fake timer(s)`);
+    }
+    jest.useRealTimers();
+  });
+
+  it('rejects when the default overall timeout expires', async () => {
+    const fetchPage = jest.fn().mockImplementation(
+      () =>
+        new Promise(resolve => {
+          setTimeout(
+            () => resolve({ items: [{ id: 1 }], pagination: {} }),
+            120_000,
+          );
+        }),
+    );
+
+    const settled = fetchAllPages(fetchPage).catch(error => error);
+
+    await jest.advanceTimersByTimeAsync(60_000);
+    const error = (await settled) as Error;
+
+    expect(error).toBeInstanceOf(Error);
+    expect(error.message).toBe('Pagination timed out after 60000 ms');
+
+    // Let the abandoned fetch settle so no fake timers are left pending.
+    await jest.advanceTimersByTimeAsync(60_000);
+  });
+
+  it('does not time out when timeoutMs is 0', async () => {
+    const fetchPage = jest.fn().mockImplementation(
+      () =>
+        new Promise(resolve => {
+          setTimeout(
+            () => resolve({ items: [{ id: 1 }], pagination: {} }),
+            120_000,
+          );
+        }),
+    );
+
+    const pending = fetchAllPages(fetchPage, { timeoutMs: 0 });
+
+    // Far beyond the 60s default budget; the run must still complete.
+    await jest.advanceTimersByTimeAsync(120_000);
+
+    await expect(pending).resolves.toEqual([{ id: 1 }]);
+  });
+});
+
+describe('fetchAllPages cancellation', () => {
+  beforeEach(() => {
+    jest.useFakeTimers();
+  });
+
+  afterEach(() => {
+    if (jest.getTimerCount() !== 0) {
+      throw new Error(`leaked ${jest.getTimerCount()} fake timer(s)`);
+    }
+    jest.useRealTimers();
+  });
+
+  it('rejects without fetching when the signal is already aborted', async () => {
+    const controller = new AbortController();
+    controller.abort();
+    const fetchPage = jest.fn();
+
+    const error = await captureError(() =>
+      fetchAllPages(fetchPage, { signal: controller.signal }),
+    );
+
+    expect(error.name).toBe('AbortError');
+    expect(error.message).toContain('Pagination aborted');
+    expect(fetchPage).not.toHaveBeenCalled();
+  });
+
+  it('rejects when the signal is aborted after the first page', async () => {
+    const controller = new AbortController();
+    const fetchPage = jest
+      .fn()
+      .mockResolvedValueOnce({
+        items: [{ id: 1 }],
+        pagination: { nextCursor: 'cursor-1' },
+      })
+      .mockImplementationOnce(() => {
+        controller.abort();
+        return new Promise(() => {});
+      });
+
+    const error = await captureError(() =>
+      fetchAllPages(fetchPage, { signal: controller.signal }),
+    );
+
+    expect(error.name).toBe('AbortError');
+    expect(error.message).toContain('Pagination aborted');
+    expect(fetchPage).toHaveBeenCalledTimes(2);
+  });
+
+  it('removes its timer and abort listener once it completes', async () => {
+    const controller = new AbortController();
+    const fetchPage = jest.fn().mockResolvedValue({
+      items: [{ id: 1 }],
+      pagination: {},
+    });
+
+    const result = await fetchAllPages(fetchPage, {
+      signal: controller.signal,
+    });
+
+    expect(result).toEqual([{ id: 1 }]);
+    expect(jest.getTimerCount()).toBe(0);
+    expect(getEventListeners(controller.signal, 'abort')).toHaveLength(0);
+  });
+});