add abort signal to content sources

.changeset/mean-trees-share.md

@@ -0,0 +1,5 @@
+---
+"@bluecadet/launchpad-content": minor
+---
+
+Send abortsignal to sources on content exit. Implemented for json, sanity, and strapi sources.

packages/content/src/launchpad-content.ts

@@ -1,5 +1,5 @@
 import path from "node:path";
-import { type Logger, LogManager, PluginDriver } from "@bluecadet/launchpad-utils";
+import { type Logger, LogManager, onExit, PluginDriver } from "@bluecadet/launchpad-utils";
 import chalk from "chalk";
 import { err, ok, okAsync, Result, ResultAsync } from "neverthrow";
 import {
@@ -23,6 +23,7 @@ class LaunchpadContent {
 	_rawSources: ConfigContentSource[];
 	_startDatetime = new Date();
 	_dataStore: DataStore;
+	_abortController = new AbortController();
 	_cwd: string;
 
 	constructor(config: ContentConfig, parentLogger: Logger, cwd = process.cwd()) {
@@ -37,6 +38,10 @@ class LaunchpadContent {
 		// create all sources
 		this._rawSources = this._config.sources;
 
+		onExit(() => {
+			this._abortController.abort();
+		});
+
 		const basePluginDriver = new PluginDriver(
 			{ logger: this._logger, cwd: this._cwd },
 			this._config.plugins,
@@ -307,6 +312,7 @@ class LaunchpadContent {
 		const initializedFetch = source.fetch({
 			logger: sourceLogger,
 			dataStore: this._dataStore,
+			abortSignal: this._abortController.signal,
 		});
 
 		const fetchAsArray = Array.isArray(initializedFetch) ? initializedFetch : [initializedFetch];

packages/content/src/sources/__tests__/airtable-source.test.ts

@@ -20,9 +20,12 @@ afterAll(() => {
 afterEach(() => server.resetHandlers());
 
 function createFetchContext() {
+	const abortController = new AbortController();
 	return {
 		logger: createMockLogger(),
 		dataStore: new DataStore("/"),
+		abortSignal: abortController.signal,
+		_abortController: abortController,
 	};
 }
 

packages/content/src/sources/__tests__/contentful-source.test.ts

@@ -20,9 +20,12 @@ afterAll(() => {
 afterEach(() => server.resetHandlers());
 
 function createFetchContext() {
+	const abortController = new AbortController();
 	return {
 		logger: createMockLogger(),
 		dataStore: new DataStore("/"),
+		abortSignal: abortController.signal,
+		_abortController: abortController,
 	};
 }
 

packages/content/src/sources/__tests__/json-source.test.ts

@@ -20,9 +20,12 @@ afterAll(() => {
 afterEach(() => server.resetHandlers());
 
 function createFetchContext() {
+	const abortController = new AbortController();
 	return {
 		logger: createMockLogger(),
 		dataStore: new DataStore("/"),
+		abortSignal: abortController.signal,
+		_abortController: abortController,
 	};
 }
 
@@ -132,4 +135,43 @@ describe("jsonSource", () => {
 		// @ts-expect-error - incomplete config
 		await expect(() => jsonSource({})).toThrow();
 	});
+
+	it("should cancel request on abortSignal", async () => {
+		const ctx = createFetchContext();
+
+		server.use(
+			http.get("https://api.example.com/slow", async () => {
+				await new Promise((resolve) => setTimeout(resolve, 2000));
+				return HttpResponse.json({ key: "value" });
+			}),
+		);
+
+		const source = await jsonSource({
+			id: "test-json-timeout",
+			files: {
+				"slow.json": "https://api.example.com/slow",
+			},
+			maxTimeout: 200,
+		});
+
+		const result = source.fetch(ctx);
+
+		const promise = result[0]!.data;
+
+		const abortReason = "Some abort reason";
+
+		// Abort the request after a short delay
+		setTimeout(() => {
+			ctx._abortController.abort(abortReason);
+		}, 100);
+
+		vi.runAllTimersAsync();
+
+		await expect(promise).rejects.toThrowError(abortReason);
+	});
+
+	it("should throw on incomplete config", async () => {
+		// @ts-expect-error - incomplete config
+		await expect(() => jsonSource({})).toThrow();
+	});
 });

packages/content/src/sources/__tests__/sanity-source.test.ts

@@ -20,9 +20,12 @@ afterAll(() => {
 afterEach(() => server.resetHandlers());
 
 function createFetchContext() {
+	const abortController = new AbortController();
 	return {
 		logger: createMockLogger(),
 		dataStore: new DataStore("/"),
+		abortSignal: abortController.signal,
+		_abortController: abortController,
 	};
 }
 
@@ -267,4 +270,48 @@ describe("sanitySource", () => {
 		expect(data1).toEqual({ _type: "test", title: "Test Document" });
 		expect(data2).toEqual({ _type: "test", title: "Test Document" });
 	});
+
+	it("should cancel request on abortSignal", async () => {
+		const ctx = createFetchContext();
+
+		server.use(
+			http.get("https://test-project.api.sanity.io/v2021-10-21/data/query/production", async () => {
+				await new Promise((resolve) => setTimeout(resolve, 2000));
+				return HttpResponse.json({
+					result: { _type: "test", title: "Test Document" },
+					ms: 2000,
+				});
+			}),
+		);
+
+		const source = await sanitySource({
+			id: "test-sanity",
+			projectId: "test-project",
+			apiToken: "test-token",
+			queries: [
+				{
+					id: "custom1",
+					query: '*[_type == "custom"][0]',
+				},
+			],
+			limit: 50,
+			mergePages: false,
+			useCdn: false,
+		});
+
+		const result = source.fetch(ctx);
+
+		const promise = result[0]!.data;
+
+		const abortReason = "Some abort reason";
+
+		// Abort the request after a short delay
+		setTimeout(() => {
+			ctx._abortController.abort(abortReason);
+		}, 50);
+
+		vi.runAllTimersAsync();
+
+		await expect(promise).rejects.toThrowError(abortReason);
+	});
 });

packages/content/src/sources/__tests__/strapi-source.test.ts

@@ -20,9 +20,12 @@ afterAll(() => {
 afterEach(() => server.resetHandlers());
 
 function createFetchContext() {
+	const abortController = new AbortController();
 	return {
 		logger: createMockLogger(),
 		dataStore: new DataStore("/"),
+		abortSignal: abortController.signal,
+		_abortController: abortController,
 	};
 }
 
@@ -348,4 +351,81 @@ describe.runIf(majorNodeVersion >= 20)("strapiSource", () => {
 
 		await expect(async () => (await result[0]!.data.next()).value).rejects.toThrow();
 	});
+
+	it("should cancel request on abortSignal", async () => {
+		const ctx = createFetchContext();
+
+		server.use(
+			http.post("http://localhost:1337/auth/local", () => {
+				return HttpResponse.json({ jwt: "test-token" });
+			}),
+			http.get("http://localhost:1337/api/custom-content", async ({ request }) => {
+				await new Promise((resolve) => setTimeout(resolve, 2000));
+				const url = new URL(request.url);
+				expect(url.searchParams.get("filters[type][$eq]")).toBe("test");
+				expect(url.searchParams.get("sort[0]")).toBe("createdAt:desc");
+
+				if (url.searchParams.get("pagination[page]") === "2") {
+					return HttpResponse.json({
+						data: [],
+						meta: {
+							pagination: { page: 2, pageSize: 100, pageCount: 1, total: 0 },
+						},
+					});
+				}
+
+				return HttpResponse.json({
+					data: [
+						{
+							id: 1,
+							attributes: {
+								title: "Custom Content",
+								type: "test",
+							},
+						},
+					],
+					meta: {
+						pagination: {
+							page: 1,
+							pageSize: 100,
+							pageCount: 1,
+							total: 1,
+						},
+					},
+				});
+			}),
+		);
+
+		const source = await strapiSource({
+			id: "test-strapi",
+			version: "4",
+			baseUrl: "http://localhost:1337",
+			identifier: "test@example.com",
+			password: "password",
+			queries: [
+				{
+					contentType: "custom-content",
+					params: {
+						"filters[type][$eq]": "test",
+						"sort[0]": "createdAt:desc",
+					},
+				},
+			],
+		});
+
+		const result = source.fetch(ctx);
+
+		const promise = result[0]!.data.next();
+
+		const abortReason = "Some abort reason";
+
+		// Abort the request after a short delay
+		setTimeout(() => {
+			ctx._abortController.abort(abortReason);
+		}, 100);
+
+		vi.runAllTimersAsync();
+
+		await expect(promise).rejects.toThrowError(abortReason);
+	});
 });