Method queueProcessor

src/webhook-sender.ts:56–166

(job, done)

     */
    constructor(protected server: Server) {
        let queueProcessor = (job, done) => {
            let rawData: JobData = job.data;

            const { appKey, payload, originalPusherSignature } = rawData;

            server.appManager.findByKey(appKey).then(app => {
                // Ensure the payload hasn't been tampered with between the job being dispatched
                // and here, as we may need to recalculate the signature post filtration.
                if (originalPusherSignature !== createWebhookHmac(JSON.stringify(payload), app.secret)) {
                    return;
                }

                async.each(app.webhooks, (webhook: WebhookInterface, resolveWebhook) => {
                    const originalEventsLength = payload.events.length;
                    let filteredPayloadEvents = payload.events;

                    filteredPayloadEvents = filteredPayloadEvents.filter(event => {
                        if (!webhook.event_types.includes(event.name)) {
                            return false;
                        }

                        if (webhook.filter) {
                            if (webhook.filter.channel_name_starts_with && !event.channel.startsWith(webhook.filter.channel_name_starts_with)) {
                                return false;
                            }

                            if (webhook.filter.channel_name_ends_with && !event.channel.endsWith(webhook.filter.channel_name_ends_with)) {
                                return false;
                            }
                        }

                        return true;
                    });

                    // If there's no webhooks to send after filtration, we should resolve early.
                    if (filteredPayloadEvents.length === 0) {
                        return resolveWebhook();
                    }

                    // If any events have been filtered out, regenerate the signature
                    let pusherSignature = (originalEventsLength !== filteredPayloadEvents.length)
                        ? createWebhookHmac(JSON.stringify(payload), app.secret)
                        : originalPusherSignature;

                    if (this.server.options.debug) {
                        Log.webhookSenderTitle('🚀 Processing webhook from queue.');
                        Log.webhookSender({ appKey, payload, pusherSignature });
                    }

                    const headers = {
                        'Accept': 'application/json',
                        'Content-Type': 'application/json',
                        'User-Agent': `SoketiWebhooksAxiosClient/1.0 (Process: ${this.server.options.instance.process_id})`,
                        // We specifically merge in the custom headers here so the headers below cannot be overwritten
                        ...webhook.headers ?? {},
                        'X-Pusher-Key': appKey,
                        'X-Pusher-Signature': pusherSignature,
                    };