import { createInflate } from 'zlib' ; import connectBusboy from 'connect-busboy' ; import express from 'express' ; import cors from 'cors' ; import type { SegmentFile , SessionReplayCall } from '../../types/serverEvents' ; import type { EventRegistry } from '../eventsRegistry' ; export function createIntakeServerApp ( serverEvents , bridgeEvents ) { const app = express ( ) ; app . use ( cors ( ) ) ; app . use ( express . text ( ) ) ; app . use ( connectBusboy ( { immediate : true } ) ) ; app . post ( '/v1/input/:endpoint' , async ( req , res ) => { const endpoint = req . params . endpoint ; const isBridge = req . query . bridge ; const events = isBridge ? bridgeEvents : serverEvents ; if ( endpoint === 'rum' || endpoint === 'logs' || endpoint === 'internalMonitoring' ) { ; ( req . body as string ) . split ( '\n' ) . map ( ( rawEvent ) => events . push ( endpoint , JSON . parse ( rawEvent ) ) ) ; } if ( endpoint === 'sessionReplay' && req . busboy ) { events . push ( 'sessionReplay' , await readSessionReplay ( req ) ) ; } res . end ( ) ; } ) ; return app ; } async function readSessionReplay ( req ) { return new Promise ( ( resolve , reject ) => { const meta = { } ; let segmentPromise ; req . busboy . on ( 'file' , ( fieldname , file , filename , encoding , mimetype ) => { if ( fieldname === 'segment' ) { segmentPromise = readStream ( file . pipe ( createInflate ( ) ) ) . then ( ( data ) => ( { encoding , filename , mimetype , data : JSON . parse ( data . toString ( ) ) , } ) ) ; } } ) ; req . busboy . on ( 'field' , ( key , value ) => { meta [ key ] = value ; } ) ; req . busboy . on ( 'finish' , ( ) => { segmentPromise . then ( ( segment ) => resolve ( { meta , segment } ) ) . catch ( ( e ) => reject ( e ) ) ; } ) ; } ) ; } async function readStream ( stream ) { return new Promise ( ( resolve , reject ) => { const buffers = [ ] ; stream . on ( 'data' , ( data ) => { buffers . push ( data ) ; } ) ; stream . on ( 'error' , ( error ) => { reject ( error ) ; } ) ; stream . on ( 'end' , ( ) => { resolve ( Buffer . concat ( buffers ) ) ; } ) ; } ) ; }