diff --git a/src/crawlers.ts b/src/crawlers.ts index 9515bd3..dd77495 100644 --- a/src/crawlers.ts +++ b/src/crawlers.ts @@ -20,7 +20,7 @@ import { ContentCrawlerTypes, GOOGLE_STANDARD_RESULTS_PER_PAGE } from './const.j import { deduplicateResults, scrapeOrganicResults } from './google-search/google-extractors-urls.js'; import { getMiniActor } from './mini-actors.js'; import { failedRequestHandler, requestHandlerCheerio, requestHandlerPlaywright } from './request-handler.js'; -import { addEmptyResultToResponse, sendResponseError } from './responses.js'; +import { addEmptyResultToResponse, sendResponseError, sendResponseIfFinished } from './responses.js'; import type { ContentCrawlerOptions, ContentCrawlerUserData, SearchCrawlerUserData } from './types.js'; import { addTimeMeasureEvent, createRequest, createSearchRequest, isActorStandby, randomId } from './utils.js'; @@ -107,7 +107,7 @@ export async function createAndStartSearchCrawler( const organicResults = scrapeOrganicResults($); // Destructure userData for easier access (pagination fields are initialized in createSearchRequest) - const { collectedResults, currentPage, totalPages, maxResults, userAuthorization } = request.userData; + const { collectedResults, currentPage, totalPages, maxResults, actorRequestId } = request.userData; // Merge with previously collected results and deduplicate const allResults = [...collectedResults, ...organicResults]; @@ -155,7 +155,7 @@ export async function createAndStartSearchCrawler( responseId, request.userData.contentScraperSettings!, request.userData.timeMeasures!, - userAuthorization, + actorRequestId, ); await addContentCrawlRequest(r, responseId, request.userData.contentCrawlerKey!); } @@ -232,10 +232,12 @@ async function createPlaywrightContentCrawler( requestHandler: (async (context) => { const typedContext = context as unknown as PlaywrightCrawlingContext; await requestHandlerPlaywright(typedContext, blocker); - await maybeCharge(ContentCrawlerTypes.PLAYWRIGHT, typedContext.request.userData.userAuthorization); + await maybeCharge(ContentCrawlerTypes.PLAYWRIGHT, typedContext.request.userData.actorRequestId); + sendResponseIfFinished(typedContext.request.userData.responseId!); }), failedRequestHandler: async ({ request }, err) => { await failedRequestHandler(request, err, ContentCrawlerTypes.PLAYWRIGHT); + sendResponseIfFinished(request.userData.responseId!); }, }); } @@ -253,10 +255,12 @@ async function createCheerioContentCrawler( requestHandler: (async (context) => { const typedContext = context as unknown as CheerioCrawlingContext; await requestHandlerCheerio(typedContext); - await maybeCharge(ContentCrawlerTypes.CHEERIO, typedContext.request.userData.userAuthorization); + await maybeCharge(ContentCrawlerTypes.CHEERIO, typedContext.request.userData.actorRequestId); + sendResponseIfFinished(typedContext.request.userData.responseId!); }), failedRequestHandler: async ({ request }, err) => { await failedRequestHandler(request, err, ContentCrawlerTypes.CHEERIO); + sendResponseIfFinished(request.userData.responseId!); }, }); } @@ -276,9 +280,9 @@ async function chargeNormal(eventName: string): Promise { /** * Multi-tenant standby charging: POSTs directly to the platform charge REST endpoint, - * passing the calling end-user's authorization so that user (not the Actor owner) is billed. + * passing the calling request's ID so that the correct caller (not the Actor owner) is billed. */ -async function chargeStandby(eventName: string, userAuthorization: string): Promise { +async function chargeStandby(eventName: string, actorRequestId: string): Promise { const { apiBaseUrl, actorRunId, token } = Actor.getEnv(); if (!apiBaseUrl || !actorRunId || !token) { log.warning(`Skipping standby charge for ${eventName} event: missing apiBaseUrl/actorRunId/token from Actor.getEnv().`); @@ -292,7 +296,7 @@ async function chargeStandby(eventName: string, userAuthorization: string): Prom Authorization: `Bearer ${token}`, 'Idempotency-Key': randomId(), }, - body: JSON.stringify({ eventName, count: 1, userAuthorization }), + body: JSON.stringify({ eventName, count: 1, requestId: actorRequestId }), }); if (!response.ok) { const resText = await response.text(); @@ -304,21 +308,29 @@ async function chargeStandby(eventName: string, userAuthorization: string): Prom * Dispatches to the correct charging path (normal single-run vs. multi-tenant standby) * based on isActorStandby(). */ -async function maybeCharge(crawlerType: ContentCrawlerTypes, userAuthorization?: string) { +const CHARGE_TIMEOUT_MILLIS = 5_000; + +async function maybeCharge(crawlerType: ContentCrawlerTypes, actorRequestId?: string) { if (getMiniActor().name !== 'url-to-markdown') { return; } const eventName = getEventName(crawlerType); try { - if (isActorStandby()) { - if (!userAuthorization) { - log.warning(`Skipping standby charge for ${eventName} event: missing userAuthorization (x-apify-user-authorization header was not provided).`); - return; - } - await chargeStandby(eventName, userAuthorization); - } else { - await chargeNormal(eventName); - } + const chargePromise = isActorStandby() + ? (async () => { + if (!actorRequestId) { + log.warning(`Skipping standby charge for ${eventName} event: missing actorRequestId (x-actor-request-id header was not provided).`); + return; + } + await chargeStandby(eventName, actorRequestId); + })() + : chargeNormal(eventName); + + const timeoutPromise = new Promise((_, reject) => { + setTimeout(() => reject(new Error(`Charging timed out after ${CHARGE_TIMEOUT_MILLIS} ms`)), CHARGE_TIMEOUT_MILLIS); + }); + + await Promise.race([chargePromise, timeoutPromise]); } catch (err) { log.error(`Failed to charge for ${eventName} event: ${err instanceof Error ? err.message : String(err)}`); } diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 86ab28a..8404317 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -11,7 +11,7 @@ import type { Input } from '../types.js'; export class McpServer { private server: Server; private toolName: string; - private userAuthorization?: string; + private actorRequestId?: string; constructor() { const miniActor = getMiniActor(); @@ -58,15 +58,15 @@ export class McpServer { this.server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params; if (name === this.toolName) { - const content = await handleModelContextProtocol(args as unknown as Input, this.userAuthorization); + const content = await handleModelContextProtocol(args as unknown as Input, this.actorRequestId); return { content: content.map((message) => ({ type: 'text', text: JSON.stringify(message) })) }; } throw new Error(`Unknown tool: ${name}`); }); } - async connect(transport: Transport, userAuthorization?: string): Promise { - this.userAuthorization = userAuthorization; + async connect(transport: Transport, actorRequestId?: string): Promise { + this.actorRequestId = actorRequestId; await this.server.connect(transport); } } diff --git a/src/request-handler.ts b/src/request-handler.ts index cf42198..1a36b60 100644 --- a/src/request-handler.ts +++ b/src/request-handler.ts @@ -7,7 +7,7 @@ import { type CheerioCrawlingContext, htmlToText, log, type PlaywrightCrawlingCo import { ContentCrawlerStatus, ContentCrawlerTypes } from './const.js'; import { blockMediaRequests, SKIPPED_MEDIA_FILE_MESSAGE } from './media.js'; -import { addResultToResponse, responseData, sendResponseIfFinished } from './responses.js'; +import { addResultToResponse, responseData } from './responses.js'; import type { ContentCrawlerUserData, Output } from './types.js'; import { addTimeMeasureEvent, isActorStandby, transformTimeMeasuresToRelative } from './utils.js'; import { @@ -153,7 +153,6 @@ async function pushSkippedResult( await context.pushData(resultSkipped); if (responseId) { addResultToResponse(responseId, request.uniqueKey, resultSkipped); - sendResponseIfFinished(responseId); } } @@ -230,7 +229,6 @@ async function handleContent( // Get responseId from the request.userData, which corresponds to the original search request if (responseId) { addResultToResponse(responseId, request.uniqueKey, result); - sendResponseIfFinished(responseId); } } @@ -348,6 +346,5 @@ export async function failedRequestHandler(request: Request, err: Error, crawler log.info(`Adding result to the Apify dataset, url: ${request.url}`); await Actor.pushData(resultErr); addResultToResponse(responseId, request.uniqueKey, resultErr); - sendResponseIfFinished(responseId); } } diff --git a/src/search.ts b/src/search.ts index 620c460..3236626 100644 --- a/src/search.ts +++ b/src/search.ts @@ -13,7 +13,7 @@ import { addTimeMeasureEvent, createRequest, createSearchRequest, - extractUserAuthorization, + extractActorRequestId, interpretAsUrl, parseParameters, randomId, @@ -29,7 +29,7 @@ function prepareRequest( searchCrawlerOptions: CheerioCrawlerOptions, contentCrawlerKey: string, contentScraperSettings: ContentScraperSettings, - userAuthorization?: string, + actorRequestId?: string, ) { if (!getMiniActor().runsSearch) { const { url } = (input as Input & UrlToMarkdownInput); @@ -48,7 +48,7 @@ function prepareRequest( responseId, contentScraperSettings, null, - userAuthorization, + actorRequestId, ); addTimeMeasureEvent(req.userData!, 'request-received', Date.now()); return { req, isUrl: true, responseId }; @@ -69,7 +69,7 @@ function prepareRequest( responseId, contentScraperSettings, null, - userAuthorization, + actorRequestId, ) : createSearchRequest( { @@ -78,7 +78,7 @@ function prepareRequest( maxResults, contentCrawlerKey, contentScraperSettings, - userAuthorization, + actorRequestId, }, searchCrawlerOptions.proxyConfiguration, ); @@ -91,7 +91,7 @@ function prepareRequest( * Internal function that handles the common logic for search. * Returns a promise that resolves to the final results array of Output objects. */ -async function runSearchProcess(params: Partial, userAuthorization?: string): Promise { +async function runSearchProcess(params: Partial, actorRequestId?: string): Promise { // Process the query parameters the same way as normal inputs const { input, @@ -111,7 +111,7 @@ async function runSearchProcess(params: Partial, userAuthorization?: stri searchCrawlerOptions, contentCrawlerKey, contentScraperSettings, - userAuthorization, + actorRequestId, ); // Create a promise that resolves when all requests are processed @@ -143,9 +143,9 @@ export async function handleSearchRequest(request: IncomingMessage, response: Se const params = parseParameters(request.url?.slice(getMiniActor().route.length) ?? ''); log.info(`Received query parameters: ${JSON.stringify(params)}`); - const userAuthorization = extractUserAuthorization(request.headers); + const actorRequestId = extractActorRequestId(request.headers); - const results = await runSearchProcess(params, userAuthorization); + const results = await runSearchProcess(params, actorRequestId); response.writeHead(200, { 'Content-Type': 'application/json' }); response.end(JSON.stringify(results)); @@ -162,10 +162,10 @@ export async function handleSearchRequest(request: IncomingMessage, response: Se * Handles the model context protocol scenario (non-HTTP scenario). * Uses the same runSearchProcess function but just returns the results as a promise. */ -export async function handleModelContextProtocol(params: Partial, userAuthorization?: string): Promise { +export async function handleModelContextProtocol(params: Partial, actorRequestId?: string): Promise { try { log.info(`Received parameters: ${JSON.stringify(params)}`); - return await runSearchProcess(params, userAuthorization); + return await runSearchProcess(params, actorRequestId); } catch (e) { const error = e as Error; log.error(`UserInputError occurred: ${error.message}`); diff --git a/src/server.ts b/src/server.ts index d866740..e014a12 100644 --- a/src/server.ts +++ b/src/server.ts @@ -6,7 +6,7 @@ import { Routes } from './const.js'; import { McpServer } from './mcp/server.js'; import { getMiniActor } from './mini-actors.js'; import { handleSearchRequest } from './search.js'; -import { extractUserAuthorization } from './utils.js'; +import { extractActorRequestId } from './utils.js'; export function createServer(): express.Express { const app = express(); @@ -36,8 +36,8 @@ export function createServer(): express.Express { app.get(Routes.SSE, async (req: Request, res: Response) => { log.info(`Received GET message at: ${req.url}`); transport = new SSEServerTransport(Routes.MESSAGE, res); - const userAuthorization = extractUserAuthorization(req.headers); - await mcpServer.connect(transport, userAuthorization); + const actorRequestId = extractActorRequestId(req.headers); + await mcpServer.connect(transport, actorRequestId); }); app.post(Routes.MESSAGE, async (req: Request, res: Response) => { diff --git a/src/types.ts b/src/types.ts index d50e63f..dd6b408 100644 --- a/src/types.ts +++ b/src/types.ts @@ -111,7 +111,7 @@ export type SearchCrawlerUserData = { currentPage: number; /** Max pages: ceil(maxResults/10) + 1 to handle pages with <10 results */ totalPages: number; - userAuthorization?: string; + actorRequestId?: string; }; /** @@ -127,7 +127,7 @@ export type ContentCrawlerUserData = { searchResult?: OrganicResult; contentCrawlerKey?: string; contentScraperSettings: ContentScraperSettings; - userAuthorization?: string; + actorRequestId?: string; }; export type Output = { diff --git a/src/utils.ts b/src/utils.ts index 77f7107..63eb813 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -64,11 +64,11 @@ export async function abortRun(statusMessage: string): Promise { } /** - * Extracts the calling end-user's authorization from the x-apify-user-authorization header. + * Extracts the actor request ID from the x-actor-request-id header. * Node lower-cases header names, and the header could theoretically arrive as a string array. */ -export function extractUserAuthorization(headers: IncomingHttpHeaders): string | undefined { - const header = headers['x-apify-user-authorization']; +export function extractActorRequestId(headers: IncomingHttpHeaders): string | undefined { + const header = headers['x-actor-request-id']; return Array.isArray(header) ? header[0] : header; } @@ -172,7 +172,7 @@ export function createSearchRequest( collectedResults, currentPage, totalPages, - userAuthorization: userData.userAuthorization, + actorRequestId: userData.actorRequestId, }, }; } @@ -186,7 +186,7 @@ export function createRequest( responseId: string, contentScraperSettings: ContentScraperSettings, timeMeasures: TimeMeasure[] | null = null, - userAuthorization?: string, + actorRequestId?: string, ): RequestOptions { return { url: result.url!, @@ -199,7 +199,7 @@ export function createRequest( searchResult: result.url && result.title ? result : undefined, timeMeasures: timeMeasures ? [...timeMeasures] : [], contentScraperSettings, - userAuthorization, + actorRequestId, }, }; }