fix(catalog): forward pagination cursor for planes and pipelines

.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.

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);
+  });
+});