@@ -132,105 +132,143 @@ OUTPUT: respond with ONLY a JSON object (no prose, no markdown fence):
132132At most ${ max } ;
133133}
134134
135+ // Bounded parallelism. The search-results queue made concurrent crawls the
136+ // norm (N searches in flight across N orgs), and the old awaited-in-loop
137+ // consumer serialized them — a batch of five put the tail past the caller's
138+ // 600s ceiling (observed live 2026-07-24: both NYT crawls timed out behind
139+ // three others). Cap of 3 keeps parallel model turns modest for rate limits;
140+ // a released slot hands off directly to the next waiter.
141+ const MAX_CONCURRENT_CRAWLS = Number ( process . env . MAX_CONCURRENT_CRAWLS ?? 3 ) ;
142+ let activeCrawls = 0 ;
143+ const crawlWaiters : ( ( ) => void ) [ ] = [ ] ;
144+
145+ function acquireCrawlSlot ( ) : Promise < void > {
146+ if ( activeCrawls < MAX_CONCURRENT_CRAWLS ) {
147+ activeCrawls += 1 ;
148+ return Promise . resolve ( ) ;
149+ }
150+ return new Promise ( ( resolve ) => crawlWaiters . push ( resolve ) ) ;
151+ }
152+
153+ function releaseCrawlSlot ( ) : void {
154+ const next = crawlWaiters . shift ( ) ;
155+ if ( next ) next ( ) ; // the slot passes directly; activeCrawls stays counted
156+ else activeCrawls -= 1 ;
157+ }
158+
159+ type CrawlMsg = { json < T > ( ) : T ; reply ?: string ; respond ( data : string ) : void } ;
160+
135161export function registerCrawlHandler ( nc : NatsConnection ) : void {
136162 ( async ( ) => {
137163 const sub = nc . subscribe ( 'organization.crawl.requested' ) ;
138164 for await ( const msg of sub ) {
139- const args = msg . json ( ) as CrawlInput ;
140- const started = Date . now ( ) ;
141- console . log ( JSON . stringify ( { level : 'info' , msg : 'crawl started' , ...args } ) ) ;
142- try {
143- if ( ! args . org_slug ?. trim ( ) ) throw new Error ( 'organization.crawl: org_slug is required' ) ;
144- if ( ! [ 'links' , 'streams' , 'team' ] . includes ( args . target ) ) {
145- throw new Error ( `organization.crawl: unknown target ${ String ( args . target ) } ) ;
146- }
147- const max = Math . min ( Math . max ( args . max_results ?? 12 , 1 ) , 25 ) ;
148-
149- const detail = await natsJson < { ok : boolean ; org ?: OrgDetail ; error ?: string } > (
150- nc ,
151- 'organization.detail.requested' ,
152- { org_slug : args . org_slug , client : args . client } ,
153- ) ;
154- if ( ! detail . ok || ! detail . org ) throw new Error ( detail . error || 'organization.detail failed' ) ;
155- const org = detail . org ;
156-
157- const briefReply = await natsJson < { ok : boolean ; brief ?: string | null } > (
158- nc ,
159- 'client.brief.get.requested' ,
160- { client : args . client } ,
161- ) ;
162- const brief = briefReply . ok ? ( briefReply . brief ?? null ) : null ;
163-
164- const existing =
165- args . target === 'streams'
166- ? ( org . media_streams ?? [ ] ) . map ( ( e ) => e ?. url ?? '' ) . filter ( Boolean )
167- : ( org . org_links ?? [ ] ) . map ( ( e ) => e ?. url ?? '' ) . filter ( Boolean ) ;
168-
169- const request = buildRequest ( promptFor ( args . target , org , existing , brief , max ) , {
170- model : CRAWL_MODEL ,
171- maxTokens : CRAWL_MAX_TOKENS ,
172- tools : [ 'web_search' ] ,
173- } ) ;
174- const text = await runPrompt ( request ) ;
175- const parsed = extractJson ( text ) ;
176-
177- if ( args . target === 'team' ) {
178- const obj = ( parsed ?? { } ) as {
179- source_urls ?: unknown [ ] ;
180- people ?: Partial < CrawlPerson > [ ] ;
181- filtered_note ?: string ;
182- } ;
183- const people : CrawlPerson [ ] = ( obj . people ?? [ ] )
184- . filter ( ( p ) => typeof p ?. name === 'string' && p . name . trim ( ) )
185- . slice ( 0 , max )
186- . map ( ( p ) => ( {
187- name : ( p . name as string ) . trim ( ) ,
188- role : p . role ?. toString ( ) . trim ( ) || null ,
189- headline : p . headline ?. toString ( ) . trim ( ) || null ,
190- linkedin_url : p . linkedin_url ?. toString ( ) . trim ( ) || null ,
191- bio_url : p . bio_url ?. toString ( ) . trim ( ) || null ,
192- } ) ) ;
193- const reply = {
194- ok : true ,
195- people,
196- filtered_note : obj . filtered_note ?. toString ( ) ?? '' ,
197- source_urls : ( obj . source_urls ?? [ ] ) . map ( String ) . filter ( Boolean ) ,
198- } ;
199- if ( msg . reply ) msg . respond ( JSON . stringify ( reply ) ) ;
200- } else {
201- const have = new Set ( existing . map ( ( u ) => u . trim ( ) ) ) ;
202- const seen = new Set < string > ( ) ;
203- const results = ( ( Array . isArray ( parsed ) ? parsed : [ ] ) as Partial < CrawlLinkCandidate > [ ] )
204- . filter ( ( r ) => typeof r ?. url === 'string' && r . url . trim ( ) )
205- . map ( ( r ) => ( {
206- url : ( r . url as string ) . trim ( ) ,
207- kind : r . kind ?. toString ( ) . trim ( ) || undefined ,
208- name : r . name ?. toString ( ) . trim ( ) || undefined ,
209- title : r . title ?. toString ( ) . trim ( ) || ( r . url as string ) . trim ( ) ,
210- content : r . content ?. toString ( ) . trim ( ) || '' ,
211- } ) )
212- . filter ( ( r ) => {
213- if ( have . has ( r . url ) || seen . has ( r . url ) ) return false ;
214- seen . add ( r . url ) ;
215- return true ;
216- } )
217- . slice ( 0 , max ) ;
218- if ( msg . reply ) {
219- msg . respond ( JSON . stringify ( { ok : true , provider : 'didi-crawl' , results } ) ) ;
220- }
165+ // Spawn, don't await — the loop keeps consuming while crawls run.
166+ void ( async ( ) => {
167+ await acquireCrawlSlot ( ) ;
168+ try {
169+ await handleCrawl ( nc , msg as unknown as CrawlMsg ) ;
170+ } finally {
171+ releaseCrawlSlot ( ) ;
221172 }
222- console . log ( JSON . stringify ( {
223- level : 'info' ,
224- msg : 'crawl completed' ,
225- org_slug : args . org_slug ,
226- target : args . target ,
227- ms : Date . now ( ) - started ,
173+ } ) ( ) ;
174+ }
175+ } ) ( ) ;
176+ }
177+
178+ async function handleCrawl ( nc : NatsConnection , msg : CrawlMsg ) : Promise < void > {
179+ const args = msg . json ( ) as CrawlInput ;
180+ const started = Date . now ( ) ;
181+ console . log ( JSON . stringify ( { level : 'info' , msg : 'crawl started' , ...args } ) ) ;
182+ try {
183+ if ( ! args . org_slug ?. trim ( ) ) throw new Error ( 'organization.crawl: org_slug is required' ) ;
184+ if ( ! [ 'links' , 'streams' , 'team' ] . includes ( args . target ) ) {
185+ throw new Error ( `organization.crawl: unknown target ${ String ( args . target ) } ) ;
186+ }
187+ const max = Math . min ( Math . max ( args . max_results ?? 12 , 1 ) , 25 ) ;
188+
189+ const detail = await natsJson < { ok : boolean ; org ?: OrgDetail ; error ?: string } > (
190+ nc ,
191+ 'organization.detail.requested' ,
192+ { org_slug : args . org_slug , client : args . client } ,
193+ ) ;
194+ if ( ! detail . ok || ! detail . org ) throw new Error ( detail . error || 'organization.detail failed' ) ;
195+ const org = detail . org ;
196+
197+ const briefReply = await natsJson < { ok : boolean ; brief ?: string | null } > (
198+ nc ,
199+ 'client.brief.get.requested' ,
200+ { client : args . client } ,
201+ ) ;
202+ const brief = briefReply . ok ? ( briefReply . brief ?? null ) : null ;
203+
204+ const existing =
205+ args . target === 'streams'
206+ ? ( org . media_streams ?? [ ] ) . map ( ( e ) => e ?. url ?? '' ) . filter ( Boolean )
207+ : ( org . org_links ?? [ ] ) . map ( ( e ) => e ?. url ?? '' ) . filter ( Boolean ) ;
208+
209+ const request = buildRequest ( promptFor ( args . target , org , existing , brief , max ) , {
210+ model : CRAWL_MODEL ,
211+ maxTokens : CRAWL_MAX_TOKENS ,
212+ tools : [ 'web_search' ] ,
213+ } ) ;
214+ const text = await runPrompt ( request ) ;
215+ const parsed = extractJson ( text ) ;
216+
217+ if ( args . target === 'team' ) {
218+ const obj = ( parsed ?? { } ) as {
219+ source_urls ?: unknown [ ] ;
220+ people ?: Partial < CrawlPerson > [ ] ;
221+ filtered_note ?: string ;
222+ } ;
223+ const people : CrawlPerson [ ] = ( obj . people ?? [ ] )
224+ . filter ( ( p ) => typeof p ?. name === 'string' && p . name . trim ( ) )
225+ . slice ( 0 , max )
226+ . map ( ( p ) => ( {
227+ name : ( p . name as string ) . trim ( ) ,
228+ role : p . role ?. toString ( ) . trim ( ) || null ,
229+ headline : p . headline ?. toString ( ) . trim ( ) || null ,
230+ linkedin_url : p . linkedin_url ?. toString ( ) . trim ( ) || null ,
231+ bio_url : p . bio_url ?. toString ( ) . trim ( ) || null ,
228232 } ) ) ;
229- } catch ( err : unknown ) {
230- const error = describeError ( err ) ;
231- console . error ( JSON . stringify ( { level : 'error' , msg : 'crawl failed' , error } ) ) ;
232- if ( msg . reply ) msg . respond ( JSON . stringify ( { ok : false , error } ) ) ;
233+ const reply = {
234+ ok : true ,
235+ people,
236+ filtered_note : obj . filtered_note ?. toString ( ) ?? '' ,
237+ source_urls : ( obj . source_urls ?? [ ] ) . map ( String ) . filter ( Boolean ) ,
238+ } ;
239+ if ( msg . reply ) msg . respond ( JSON . stringify ( reply ) ) ;
240+ } else {
241+ const have = new Set ( existing . map ( ( u ) => u . trim ( ) ) ) ;
242+ const seen = new Set < string > ( ) ;
243+ const results = ( ( Array . isArray ( parsed ) ? parsed : [ ] ) as Partial < CrawlLinkCandidate > [ ] )
244+ . filter ( ( r ) => typeof r ?. url === 'string' && r . url . trim ( ) )
245+ . map ( ( r ) => ( {
246+ url : ( r . url as string ) . trim ( ) ,
247+ kind : r . kind ?. toString ( ) . trim ( ) || undefined ,
248+ name : r . name ?. toString ( ) . trim ( ) || undefined ,
249+ title : r . title ?. toString ( ) . trim ( ) || ( r . url as string ) . trim ( ) ,
250+ content : r . content ?. toString ( ) . trim ( ) || '' ,
251+ } ) )
252+ . filter ( ( r ) => {
253+ if ( have . has ( r . url ) || seen . has ( r . url ) ) return false ;
254+ seen . add ( r . url ) ;
255+ return true ;
256+ } )
257+ . slice ( 0 , max ) ;
258+ if ( msg . reply ) {
259+ msg . respond ( JSON . stringify ( { ok : true , provider : 'didi-crawl' , results } ) ) ;
233260 }
234261 }
235- } ) ( ) ;
262+ console . log ( JSON . stringify ( {
263+ level : 'info' ,
264+ msg : 'crawl completed' ,
265+ org_slug : args . org_slug ,
266+ target : args . target ,
267+ ms : Date . now ( ) - started ,
268+ } ) ) ;
269+ } catch ( err : unknown ) {
270+ const error = describeError ( err ) ;
271+ console . error ( JSON . stringify ( { level : 'error' , msg : 'crawl failed' , error } ) ) ;
272+ if ( msg . reply ) msg . respond ( JSON . stringify ( { ok : false , error } ) ) ;
273+ }
236274}
0 commit comments