server.js

'use strict';

import http from 'http';
import https from 'https';
import http2 from 'http2';
import fs from 'fs';
import crypto from 'crypto';
import webpush from 'web-push';
import fetch from 'node-fetch';

function toBase64Url(input) {
	const buffer = Buffer.isBuffer(input) ? input : Buffer.from(input);
	return buffer
		.toString('base64')
		.replace(/\+/g, '-')
		.replace(/\//g, '_')
		.replace(/=+$/g, '');
}

class ApplePushRequestProcessor {

	init(options, webhookAgents) {

		this.teamId = options.teamId;
		this.keyId = options.keyId;
		this.defaultTopic = options.topic;
		this.defaultEnvironment = options.environment || 'production';
		this.maxConcurrency = options.maxConcurrency || 100;
		this.requestTimeoutMs = options.requestTimeoutMs || 10000;
		this.retryMax = options.retryMax || 5;
		this.retryBaseMs = options.retryBaseMs || 300;

		this.privateKey = fs.readFileSync(options.authKeyPath, 'utf8');

		this.httpAgent = webhookAgents.httpAgent;
		this.httpsAgent = webhookAgents.httpsAgent;

		this.queue = [];
		this.activeCount = 0;

		this.sessions = {};
		this.jwtCache = null;
		this.jwtValidUntil = 0;
	}

	hasRequiredConfig() {
		return !!(this.teamId && this.keyId && this.defaultTopic && this.privateKey);
	}

	queueMessages(messages, requestId) {

		const result = {
			accepted: 0,
			rejected: 0,
			errors: []
		};

		for (let i = 0; i < messages.length; i++) {

			const message = messages[i];
			const error = this.validateMessage(message);

			if (error) {

				result.rejected++;
				result.errors.push({
					index: i,
					code: error.code,
					message: error.message
				});

				continue;
			}

			this.queue.push({
				requestId: requestId,
				messageIndex: i,
				message: message
			});

			result.accepted++;
		}

		this.processQueue();

		return result;
	}

	validateMessage(message) {

		if ((message == null) || (typeof message !== 'object')) {
			return { code: 'INVALID_MESSAGE', message: 'message must be an object' };
		}

		if ((message.clientId == null) || (String(message.clientId).trim() == '')) {
			return { code: 'INVALID_CLIENT_ID', message: 'clientId missing or invalid' };
		}

		if (!('payload' in message)) {
			return { code: 'INVALID_PAYLOAD', message: 'payload missing' };
		}

		if (typeof message.payload === 'string') {
			try {
				JSON.parse(message.payload);
			} catch (e) {
				return { code: 'INVALID_PAYLOAD', message: 'payload is not valid JSON' };
			}
		}

		if ((message.priority != null) && (message.priority !== 5) && (message.priority !== 10)) {
			return { code: 'INVALID_PRIORITY', message: 'priority must be 5 or 10' };
		}

		if ((message.environment != null) && !['production', 'sandbox'].includes(message.environment)) {
			return { code: 'INVALID_ENVIRONMENT', message: 'environment must be production or sandbox' };
		}

		return null;
	}

	processQueue() {

		while ((this.activeCount < this.maxConcurrency) && (this.queue.length > 0)) {

			const entry = this.queue.shift();
			this.activeCount++;

			this.processEntry(entry)
				.catch((error) => {
					console.log('Failed to process Apple push entry', error);
				})
				.finally(() => {
					this.activeCount--;
					this.processQueue();
				});
		}
	}

	async processEntry(entry) {

		const requestId = entry.requestId;
		const messageIndex = entry.messageIndex;
		const message = entry.message;

		const result = await this.sendMessageWithRetry(message, requestId, messageIndex);

		if (message.webhook) {
			this.sendPushResult(message.webhook, result);
		}
	}

	async sendMessageWithRetry(message, requestId, messageIndex) {

		let attempt = 0;
		let lastResult = null;

		while (attempt < this.retryMax) {

			attempt++;

			try {
				const apnsResult = await this.sendToApns(message);

				const success = apnsResult.httpStatus >= 200 && apnsResult.httpStatus < 300;
				const reason = apnsResult.reason || null;

				const retryable = !success && this.isRetryableStatus(apnsResult.httpStatus, reason);

				lastResult = this.createWebhookResult(requestId, messageIndex, message, {
					status: success ? 'sent' : 'failed',
					success: success,
					httpStatus: apnsResult.httpStatus,
					apnsId: apnsResult.apnsId,
					reason: reason,
					retryable: retryable,
					attempts: attempt,
					final: !retryable || (attempt >= this.retryMax)
				});

				if (success || !retryable) {
					return lastResult;
				}
			} catch (error) {

				const retryable = this.isRetryableError(error);

				lastResult = this.createWebhookResult(requestId, messageIndex, message, {
					status: 'failed',
					success: false,
					httpStatus: 0,
					apnsId: null,
					reason: error.message,
					retryable: retryable,
					attempts: attempt,
					final: !retryable || (attempt >= this.retryMax)
				});

				if (!retryable) {
					return lastResult;
				}
			}

			const backoffMs = Math.round(this.retryBaseMs * Math.pow(2, attempt - 1) * (0.5 + Math.random()));
			await new Promise((resolve) => setTimeout(resolve, backoffMs));
		}

		return lastResult;
	}

	isRetryableStatus(httpStatus, reason) {

		if ([429, 500, 503].includes(httpStatus)) {
			return true;
		}

		const retryableReasons = {
			TooManyRequests: true,
			InternalServerError: true,
			ServiceUnavailable: true,
			Shutdown: true
		};

		return !!retryableReasons[reason];
	}

	isRetryableError(error) {

		const retryableCodes = {
			ECONNRESET: true,
			ETIMEDOUT: true,
			EAI_AGAIN: true,
			ENOTFOUND: true,
			EPIPE: true
		};

		if (error && error.code && retryableCodes[error.code]) {
			return true;
		}

		if (error && typeof error.message === 'string') {
			return error.message.includes('timeout') || error.message.includes('APNS_TIMEOUT');
		}

		return false;
	}

	getApnsHost(environment) {
		if (environment === 'sandbox') {
			return 'api.sandbox.push.apple.com';
		}

		return 'api.push.apple.com';
	}

	getSession(environment) {

		const host = this.getApnsHost(environment);

		if (!this.sessions[host] || this.sessions[host].closed || this.sessions[host].destroyed) {

			const session = http2.connect('https://' + host + ':443');

			session.on('error', (error) => {
				console.log('APNs session error', host, error);
			});

			session.on('close', () => {
				if (this.sessions[host] === session) {
					delete this.sessions[host];
				}
			});

			this.sessions[host] = session;
		}

		return this.sessions[host];
	}

	getProviderToken() {

		const now = Math.floor(Date.now() / 1000);

		if (this.jwtCache && (this.jwtValidUntil > now)) {
			return this.jwtCache;
		}

		const header = toBase64Url(JSON.stringify({
			alg: 'ES256',
			kid: this.keyId
		}));

		const payload = toBase64Url(JSON.stringify({
			iss: this.teamId,
			iat: now
		}));

		const unsignedToken = header + '.' + payload;

		const signer = crypto.createSign('sha256');
		signer.update(unsignedToken);
		signer.end();

		const signature = signer.sign(this.privateKey);
		const token = unsignedToken + '.' + toBase64Url(signature);

		this.jwtCache = token;
		this.jwtValidUntil = now + (50 * 60);

		return token;
	}

	async sendToApns(message) {

		const environment = message.environment || this.defaultEnvironment;
		const topic = message.topic || this.defaultTopic;

		const token = this.getProviderToken();
		const session = this.getSession(environment);

		const payloadObject = (typeof message.payload === 'string') ? JSON.parse(message.payload) : message.payload;
		const payloadJson = JSON.stringify(payloadObject);

		const path = '/3/device/' + String(message.clientId).trim();
		console.log('Sending Apple notification to APNs: clientId=' + String(message.clientId).trim() + ', topic=' + topic + ', environment=' + environment);

		const headers = {
			':method': 'POST',
			':path': path,
			'authorization': 'bearer ' + token,
			'apns-topic': topic,
			'apns-push-type': message.pushType || 'alert',
			'apns-priority': String(message.priority || 10),
			'content-type': 'application/json'
		};

		if (message.collapseId) {
			headers['apns-collapse-id'] = String(message.collapseId);
		}

		if (message.expiration != null) {
			headers['apns-expiration'] = String(message.expiration);
		}

		return new Promise((resolve, reject) => {

			const req = session.request(headers);
			let responseHeaders = {};
			let body = '';
			let settled = false;

			const settleReject = (error) => {
				if (!settled) {
					settled = true;
					reject(error);
				}
			};

			const settleResolve = (value) => {
				if (!settled) {
					settled = true;
					resolve(value);
				}
			};

			req.setTimeout(this.requestTimeoutMs, () => {
				req.close(http2.constants.NGHTTP2_CANCEL);
				settleReject(new Error('APNS_TIMEOUT'));
			});

			req.on('response', (headersObject) => {
				responseHeaders = headersObject;
			});

			req.on('data', (chunk) => {
				body += chunk.toString();
			});

			req.on('error', (error) => {
				settleReject(error);
			});

			req.on('end', () => {

				let parsedBody = null;

				if (body != '') {
					try {
						parsedBody = JSON.parse(body);
					} catch (e) {
						parsedBody = null;
					}
				}

				const httpStatus = Number(responseHeaders[':status'] || 0);
				const apnsId = responseHeaders['apns-id'] || null;
				const reason = parsedBody && parsedBody.reason ? parsedBody.reason : null;

				settleResolve({
					httpStatus: httpStatus,
					apnsId: apnsId,
					reason: reason,
					body: parsedBody
				});
			});

			req.end(payloadJson);
		});
	}

	createWebhookResult(requestId, messageIndex, message, statusInfo) {
		return {
			requestId: requestId,
			messageIndex: messageIndex,
			platform: 'ios',
			status: statusInfo.status,
			success: statusInfo.success,
			httpStatus: statusInfo.httpStatus,
			apnsId: statusInfo.apnsId,
			reason: statusInfo.reason,
			retryable: statusInfo.retryable,
			attempts: statusInfo.attempts,
			final: statusInfo.final,
			clientId: message.clientId,
			topic: message.topic || this.defaultTopic,
			timestamp: new Date().toISOString(),
			meta: message.meta || {}
		};
	}

	sendPushResult(webhookUrl, result) {

		fetch(webhookUrl, {
			method: 'POST',
			body: JSON.stringify(result),
			headers: { 'Content-Type': 'application/json' },
			agent: function (parsedURL) {
				if (parsedURL.protocol == 'http:') {
					return this.httpAgent;
				}

				return this.httpsAgent;
			}.bind(this)
		}).catch((error) => {
			console.log('Apple webhook dispatch failed', webhookUrl, error);
		});
	}

}

class FirebasePushRequestProcessor {

	init(options, webhookAgents) {

		this.projectId = options.projectId;
		this.maxConcurrency = options.maxConcurrency || 100;
		this.requestTimeoutMs = options.requestTimeoutMs || 10000;
		this.retryMax = options.retryMax || 5;
		this.retryBaseMs = options.retryBaseMs || 300;

		this.httpAgent = webhookAgents.httpAgent;
		this.httpsAgent = webhookAgents.httpsAgent;

		this.queue = [];
		this.activeCount = 0;

		this.accessToken = null;
		this.accessTokenValidUntil = 0;

		this.credentials = this.loadCredentials(options);
		this.tokenUri = (this.credentials && this.credentials.token_uri) || 'https://oauth2.googleapis.com/token';
	}

	hasRequiredConfig() {
		return !!(
			this.projectId &&
			this.credentials &&
			this.credentials.client_email &&
			this.credentials.private_key
		);
	}

	loadCredentials(options) {

		if (options.credentialsJson) {
			return JSON.parse(options.credentialsJson);
		}

		if (options.credentialsPath) {
			return JSON.parse(fs.readFileSync(options.credentialsPath, 'utf8'));
		}

		return null;
	}

	queueMessages(messages, requestId) {

		const result = {
			accepted: 0,
			rejected: 0,
			errors: []
		};

		for (let i = 0; i < messages.length; i++) {

			const message = messages[i];
			const error = this.validateMessage(message);

			if (error) {

				result.rejected++;
				result.errors.push({
					index: i,
					code: error.code,
					message: error.message
				});

				continue;
			}

			this.queue.push({
				requestId: requestId,
				messageIndex: i,
				message: message
			});

			result.accepted++;
		}

		this.processQueue();

		return result;
	}

	validateMessage(message) {

		if ((message == null) || (typeof message !== 'object')) {
			return { code: 'INVALID_MESSAGE', message: 'message must be an object' };
		}

		const hasTarget =
			((message.token != null) && (String(message.token).trim() != '')) ||
			((message.clientId != null) && (String(message.clientId).trim() != '')) ||
			(message.fcmMessage && (typeof message.fcmMessage === 'object') && (String(message.fcmMessage.token || '').trim() != ''));

		if (!hasTarget) {
			return { code: 'INVALID_CLIENT_ID', message: 'token/clientId missing or invalid' };
		}

		const hasPayload = ('payload' in message) || ('fcmMessage' in message);

		if (!hasPayload) {
			return { code: 'INVALID_PAYLOAD', message: 'payload or fcmMessage missing' };
		}

		if (typeof message.payload === 'string') {
			try {
				JSON.parse(message.payload);
			} catch (e) {
				return { code: 'INVALID_PAYLOAD', message: 'payload is not valid JSON' };
			}
		}

		return null;
	}

	processQueue() {

		while ((this.activeCount < this.maxConcurrency) && (this.queue.length > 0)) {

			const entry = this.queue.shift();
			this.activeCount++;

			this.processEntry(entry)
				.catch((error) => {
					console.log('Failed to process Firebase push entry', error);
				})
				.finally(() => {
					this.activeCount--;
					this.processQueue();
				});
		}
	}

	async processEntry(entry) {

		const requestId = entry.requestId;
		const messageIndex = entry.messageIndex;
		const message = entry.message;

		const result = await this.sendMessageWithRetry(message, requestId, messageIndex);

		if (message.webhook) {
			this.sendPushResult(message.webhook, result);
		}
	}

	async sendMessageWithRetry(message, requestId, messageIndex) {

		let attempt = 0;
		let lastResult = null;

		while (attempt < this.retryMax) {

			attempt++;

			try {
				const fcmResult = await this.sendToFcm(message);

				const success = fcmResult.httpStatus >= 200 && fcmResult.httpStatus < 300;
				const reason = this.extractReason(fcmResult);

				const retryable = !success && this.isRetryableStatus(fcmResult.httpStatus, reason);

				lastResult = this.createWebhookResult(requestId, messageIndex, message, {
					status: success ? 'sent' : 'failed',
					success: success,
					httpStatus: fcmResult.httpStatus,
					messageId: fcmResult.messageId,
					reason: reason,
					retryable: retryable,
					attempts: attempt,
					final: !retryable || (attempt >= this.retryMax)
				});

				if (success || !retryable) {
					return lastResult;
				}
			} catch (error) {

				const retryable = this.isRetryableError(error);

				lastResult = this.createWebhookResult(requestId, messageIndex, message, {
					status: 'failed',
					success: false,
					httpStatus: 0,
					messageId: null,
					reason: error.message,
					retryable: retryable,
					attempts: attempt,
					final: !retryable || (attempt >= this.retryMax)
				});

				if (!retryable) {
					return lastResult;
				}
			}

			const backoffMs = Math.round(this.retryBaseMs * Math.pow(2, attempt - 1) * (0.5 + Math.random()));
			await new Promise((resolve) => setTimeout(resolve, backoffMs));
		}

		return lastResult;
	}

	isRetryableStatus(httpStatus, reason) {

		if ([408, 429, 500, 502, 503, 504].includes(httpStatus)) {
			return true;
		}

		const retryableReasons = {
			UNAVAILABLE: true,
			INTERNAL: true,
			RESOURCE_EXHAUSTED: true,
			DEADLINE_EXCEEDED: true
		};

		return !!retryableReasons[reason];
	}

	isRetryableError(error) {

		const retryableCodes = {
			ECONNRESET: true,
			ETIMEDOUT: true,
			EAI_AGAIN: true,
			ENOTFOUND: true,
			EPIPE: true,
			UND_ERR_CONNECT_TIMEOUT: true,
			UND_ERR_HEADERS_TIMEOUT: true,
			UND_ERR_BODY_TIMEOUT: true
		};

		if (error && error.code && retryableCodes[error.code]) {
			return true;
		}

		if (error && typeof error.message === 'string') {
			return error.message.includes('timeout') || error.message.includes('FCM_TIMEOUT');
		}

		return false;
	}

	async getAccessToken() {

		const now = Math.floor(Date.now() / 1000);

		if (this.accessToken && (this.accessTokenValidUntil > now)) {
			return this.accessToken;
		}

		const header = toBase64Url(JSON.stringify({
			alg: 'RS256',
			typ: 'JWT'
		}));

		const payload = toBase64Url(JSON.stringify({
			iss: this.credentials.client_email,
			sub: this.credentials.client_email,
			aud: this.tokenUri,
			scope: 'https://www.googleapis.com/auth/firebase.messaging',
			iat: now,
			exp: now + 3600
		}));

		const unsignedToken = header + '.' + payload;

		const signer = crypto.createSign('RSA-SHA256');
		signer.update(unsignedToken);
		signer.end();

		const signature = signer.sign(this.credentials.private_key);
		const assertion = unsignedToken + '.' + toBase64Url(signature);

		const response = await fetch(this.tokenUri, {
			method: 'POST',
			headers: {
				'content-type': 'application/x-www-form-urlencoded'
			},
			body: new URLSearchParams({
				grant_type: 'urn:ietf:params:oauth:grant-type:jwt-bearer',
				assertion: assertion
			}).toString()
		});

		const responseBody = await response.json();

		if (!response.ok || !responseBody.access_token) {
			throw new Error('FCM_AUTH_FAILED: ' + JSON.stringify(responseBody));
		}

		this.accessToken = responseBody.access_token;
		this.accessTokenValidUntil = now + Math.max(60, Number(responseBody.expires_in || 3600) - 60);

		return this.accessToken;
	}

	stringifyDataValues(dataObject) {

		const result = {};

		for (const key of Object.keys(dataObject || {})) {
			const value = dataObject[key];

			if (typeof value === 'string') {
				result[key] = value;
			} else if (value == null) {
				result[key] = '';
			} else {
				result[key] = JSON.stringify(value);
			}
		}

		return result;
	}

	buildFcmMessage(message) {

		const targetToken = String(message.token || message.clientId || ((message.fcmMessage && message.fcmMessage.token) || '')).trim();

		const fcmMessage = Object.assign({}, (message.fcmMessage && typeof message.fcmMessage === 'object') ? message.fcmMessage : {});
		fcmMessage.token = fcmMessage.token || targetToken;

		if (!('payload' in message)) {
			return fcmMessage;
		}

		const payloadObject = (typeof message.payload === 'string') ? JSON.parse(message.payload) : message.payload;

		if ((payloadObject != null) && (typeof payloadObject === 'object')) {

			if (payloadObject.message && (typeof payloadObject.message === 'object')) {
				Object.assign(fcmMessage, payloadObject.message);
				fcmMessage.token = fcmMessage.token || targetToken;
				return fcmMessage;
			}

			if ((payloadObject.notification != null) && (fcmMessage.notification == null)) {
				fcmMessage.notification = payloadObject.notification;
			}

			if ((payloadObject.data != null) && (fcmMessage.data == null)) {
				fcmMessage.data = this.stringifyDataValues(payloadObject.data);
			}

			if ((payloadObject.android != null) && (fcmMessage.android == null)) {
				fcmMessage.android = payloadObject.android;
			}

			if ((payloadObject.apns != null) && (fcmMessage.apns == null)) {
				fcmMessage.apns = payloadObject.apns;
			}

			if ((payloadObject.webpush != null) && (fcmMessage.webpush == null)) {
				fcmMessage.webpush = payloadObject.webpush;
			}

			if ((payloadObject.fcmOptions != null) && (fcmMessage.fcmOptions == null)) {
				fcmMessage.fcmOptions = payloadObject.fcmOptions;
			}

			if (!fcmMessage.notification && payloadObject.title) {
				fcmMessage.notification = {
					title: String(payloadObject.title),
					body: String(payloadObject.body || '')
				};
			}

			if (!fcmMessage.data && !fcmMessage.notification) {
				fcmMessage.data = {
					payload: JSON.stringify(payloadObject)
				};
			}
		}

		return fcmMessage;
	}

	async sendToFcm(message) {

		const token = await this.getAccessToken();
		const fcmMessage = this.buildFcmMessage(message);

		console.log('Sending Firebase notification to FCM: token=' + String(fcmMessage.token || '').trim() + ', projectId=' + this.projectId);

		const controller = new AbortController();
		const timeout = setTimeout(() => controller.abort(new Error('FCM_TIMEOUT')), this.requestTimeoutMs);

		let response;

		try {
			response = await fetch('https://fcm.googleapis.com/v1/projects/' + encodeURIComponent(this.projectId) + '/messages:send', {
				method: 'POST',
				headers: {
					'authorization': 'Bearer ' + token,
					'content-type': 'application/json'
				},
				body: JSON.stringify({
					message: fcmMessage,
					validate_only: !!message.validateOnly
				}),
				signal: controller.signal
			});
		} catch (error) {
			if (error && error.name === 'AbortError') {
				throw new Error('FCM_TIMEOUT');
			}

			throw error;
		} finally {
			clearTimeout(timeout);
		}

		let parsedBody = null;
		let bodyText = '';

		try {
			bodyText = await response.text();
			parsedBody = bodyText ? JSON.parse(bodyText) : null;
		} catch (e) {
			parsedBody = null;
		}

		return {
			httpStatus: response.status,
			messageId: parsedBody && parsedBody.name ? parsedBody.name : null,
			error: parsedBody && parsedBody.error ? parsedBody.error : null,
			body: parsedBody || bodyText
		};
	}

	extractReason(fcmResult) {

		if (!fcmResult || !fcmResult.error) {
			return null;
		}

		if (fcmResult.error.status) {
			return fcmResult.error.status;
		}

		if (fcmResult.error.message) {
			return fcmResult.error.message;
		}

		return 'FCM_ERROR';
	}

	createWebhookResult(requestId, messageIndex, message, statusInfo) {
		return {
			requestId: requestId,
			messageIndex: messageIndex,
			platform: 'firebase',
			status: statusInfo.status,
			success: statusInfo.success,
			httpStatus: statusInfo.httpStatus,
			messageId: statusInfo.messageId,
			reason: statusInfo.reason,
			retryable: statusInfo.retryable,
			attempts: statusInfo.attempts,
			final: statusInfo.final,
			clientId: message.clientId || message.token || null,
			projectId: this.projectId,
			timestamp: new Date().toISOString(),
			meta: message.meta || {}
		};
	}

	sendPushResult(webhookUrl, result) {

		fetch(webhookUrl, {
			method: 'POST',
			body: JSON.stringify(result),
			headers: { 'Content-Type': 'application/json' },
			agent: function (parsedURL) {
				if (parsedURL.protocol == 'http:') {
					return this.httpAgent;
				}

				return this.httpsAgent;
			}.bind(this)
		}).catch((error) => {
			console.log('Firebase webhook dispatch failed', webhookUrl, error);
		});
	}

}

class HttpRequestProcessor {

	init(options) {

		const subject = options.subject;
		const publicVapidKey = options.publicVapidKey;
		const privateVapidKey = options.privateVapidKey;

		webpush.setVapidDetails(subject, publicVapidKey, privateVapidKey);

		this.webpush = webpush;

		const maxWebhookSockets = options.maxWebhookSockets || 10;

		this.httpAgent = new http.Agent({ maxSockets: maxWebhookSockets });
		this.httpsAgent = new https.Agent({ maxSockets: maxWebhookSockets });

		if (options.apple) {

			const applePushRequestProcessor = new ApplePushRequestProcessor();

			applePushRequestProcessor.init(options.apple, {
				httpAgent: this.httpAgent,
				httpsAgent: this.httpsAgent
			});

			this.applePushRequestProcessor = applePushRequestProcessor;
		}

		if (options.firebase) {

			const firebasePushRequestProcessor = new FirebasePushRequestProcessor();

			firebasePushRequestProcessor.init(options.firebase, {
				httpAgent: this.httpAgent,
				httpsAgent: this.httpsAgent
			});

			this.firebasePushRequestProcessor = firebasePushRequestProcessor;
		}

	}

	handleHttpRequest(request, response) {

		if (request.method === 'GET') {

			if (request.url === '/health/live' || request.url === '/health/ready') {
				response.writeHead(200, { 'Content-Type': 'application/json' });
				response.end(JSON.stringify({ status: 'ok' }));
				return;
			}

			response.writeHead(404, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'not found' }));
			return;
		}

		if (request.method !== 'POST') {
			response.writeHead(405, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'method not allowed' }));
			return;
		}

		let body = '';

		request.on('data', chunk => {
			body += chunk.toString();
		});

		request.on('end', () => {

			const path = (request.url || '').split('?')[0];

			let requestObject;

			try {
				requestObject = JSON.parse(body);
			} catch (e) {
				response.writeHead(400, { 'Content-Type': 'application/json' });
				response.end(JSON.stringify({ error: 'invalid JSON body' }));
				return;
			}

			if (!requestObject || !Array.isArray(requestObject.messages)) {
				response.writeHead(400, { 'Content-Type': 'application/json' });
				response.end(JSON.stringify({ error: 'messages array required' }));
				return;
			}

			if ((path === '/dispatch/apple')) {
				this.handleAppleDispatch(requestObject, response);
				return;
			}

			if ((path === '/dispatch/firebase')) {
				this.handleFirebaseDispatch(requestObject, response);
				return;
			}

			if ((path === '/') || (path === '/dispatch') || (path === '/dispatch/web')) {
				this.handleWebDispatch(requestObject, response);
				return;
			}

			response.writeHead(404, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'not found' }));
		});

	}

	handleWebDispatch(requestObject, response) {

		if (requestObject.messages.length === 0) {
			response.writeHead(400, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'empty messages array' }));
			return;
		}

		console.log('Sending ' + requestObject.messages.length + (requestObject.messages.length == 1 ? ' web push message' : ' web push messages'));

		this.sendPushMessages(requestObject.messages);

		response.writeHead(202, { 'Content-Type': 'application/json' });
		response.end(JSON.stringify({ status: 'accepted', accepted: requestObject.messages.length, rejected: 0, errors: [] }));
	}

	handleAppleDispatch(requestObject, response) {

		if (requestObject.messages.length === 0) {
			response.writeHead(400, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'empty messages array' }));
			return;
		}

		if (!this.applePushRequestProcessor || !this.applePushRequestProcessor.hasRequiredConfig()) {
			response.writeHead(503, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'apple dispatcher not configured' }));
			return;
		}

		const requestId = crypto.randomUUID();
		const queueResult = this.applePushRequestProcessor.queueMessages(requestObject.messages, requestId);
		console.log('Apple dispatch batch submitted: requestId=' + requestId + ', accepted=' + queueResult.accepted + ', rejected=' + queueResult.rejected);

		response.writeHead(202, { 'Content-Type': 'application/json' });
		response.end(JSON.stringify({
			status: 'accepted',
			requestId: requestId,
			accepted: queueResult.accepted,
			rejected: queueResult.rejected,
			errors: queueResult.errors
		}));
	}

	handleFirebaseDispatch(requestObject, response) {

		if (requestObject.messages.length === 0) {
			response.writeHead(400, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'empty messages array' }));
			return;
		}

		if (!this.firebasePushRequestProcessor || !this.firebasePushRequestProcessor.hasRequiredConfig()) {
			response.writeHead(503, { 'Content-Type': 'application/json' });
			response.end(JSON.stringify({ error: 'firebase dispatcher not configured' }));
			return;
		}

		const requestId = crypto.randomUUID();
		const queueResult = this.firebasePushRequestProcessor.queueMessages(requestObject.messages, requestId);
		console.log('Firebase dispatch batch submitted: requestId=' + requestId + ', accepted=' + queueResult.accepted + ', rejected=' + queueResult.rejected);

		response.writeHead(202, { 'Content-Type': 'application/json' });
		response.end(JSON.stringify({
			status: 'accepted',
			requestId: requestId,
			accepted: queueResult.accepted,
			rejected: queueResult.rejected,
			errors: queueResult.errors
		}));
	}

	sendPushMessages(messages) {

		if (messages && (messages.length > 0)) {

			for (var i = 0; i < messages.length; i++) {

				var message = messages[i];

				var subscription = message.subscription;
				var payload = message.payload;
				var webhookUrl = message.webhook;

				// console.log('sending push payload', payload, subscription);

				var onSuccess = this.processPushSuccess.bind(this, webhookUrl);
				var onError = this.processPushError.bind(this, webhookUrl);

				this.webpush.sendNotification(subscription, payload).then(onSuccess).catch(onError);

			}

		}

	}

	processPushSuccess(webhookUrl, result) {

		if (webhookUrl) {

			this.sendPushResult(webhookUrl, result);

		} else {

			console.log('push succeeded', webhookUrl, result);

		}

	}

	processPushError(webhookUrl, error) {

		if (webhookUrl) {

			this.sendPushResult(webhookUrl, error);

		}

		console.log('push failed', webhookUrl, error);

	}

	sendPushResult(webhookUrl, result) {

		fetch(webhookUrl, {
			method: 'POST',
			body: JSON.stringify(result),
			headers: { 'Content-Type': 'application/json' },
			agent: function (parsedURL) {
				if (parsedURL.protocol == 'http:') {
					return this.httpAgent;
				}

				return this.httpsAgent;
			}.bind(this)
		}).catch((error) => {
			console.log('Webhook dispatch failed', webhookUrl, error);
		});

	}

}

const options = {};

if (process.env.WEBPUSH_SUBJECT) {

	options.subject = process.env.WEBPUSH_SUBJECT;

} else {

	console.log("Error: WEBPUSH_SUBJECT environment parameter required");
	process.exit(1);

}

if (process.env.WEBPUSH_PUBLIC_KEY) {

	options.publicVapidKey = process.env.WEBPUSH_PUBLIC_KEY;

} else {

	console.log("Error: WEBPUSH_PUBLIC_KEY environment parameter required");
	process.exit(1);

}

if (process.env.WEBPUSH_PRIVATE_KEY) {

	options.privateVapidKey = process.env.WEBPUSH_PRIVATE_KEY;

} else {

	console.log("Error: WEBPUSH_PRIVATE_KEY environment parameter required");
	process.exit(1);

}

var listenPortHttp = process.env.HTTP_PORT || 8080;

options.maxWebhookSockets = process.env.MAX_WEBHOOK_SOCKETS || 20;

const appleConfigProvided =
	!!process.env.APNS_TEAM_ID ||
	!!process.env.APNS_KEY_ID ||
	!!process.env.APNS_TOPIC ||
	!!process.env.APNS_AUTH_KEY_PATH;

if (appleConfigProvided) {

	if (!process.env.APNS_TEAM_ID || !process.env.APNS_KEY_ID || !process.env.APNS_TOPIC || !process.env.APNS_AUTH_KEY_PATH) {
		console.log('Error: APNS_TEAM_ID, APNS_KEY_ID, APNS_TOPIC and APNS_AUTH_KEY_PATH must all be set when enabling Apple dispatch');
		process.exit(1);
	}

	options.apple = {
		teamId: process.env.APNS_TEAM_ID,
		keyId: process.env.APNS_KEY_ID,
		topic: process.env.APNS_TOPIC,
		authKeyPath: process.env.APNS_AUTH_KEY_PATH,
		environment: process.env.APNS_ENVIRONMENT || 'production',
		maxConcurrency: Number(process.env.APNS_MAX_CONCURRENCY || 100),
		requestTimeoutMs: Number(process.env.APNS_REQUEST_TIMEOUT_MS || 10000),
		retryMax: Number(process.env.APNS_RETRY_MAX || 5),
		retryBaseMs: Number(process.env.APNS_RETRY_BASE_MS || 300)
	};
}

const firebaseConfigProvided =
	!!process.env.FCM_PROJECT_ID ||
	!!process.env.FCM_CREDENTIALS_PATH ||
	!!process.env.FCM_CREDENTIALS_JSON;

if (firebaseConfigProvided) {

	if (!process.env.FCM_PROJECT_ID || (!process.env.FCM_CREDENTIALS_PATH && !process.env.FCM_CREDENTIALS_JSON)) {
		console.log('Error: FCM_PROJECT_ID and one of FCM_CREDENTIALS_PATH or FCM_CREDENTIALS_JSON must be set when enabling Firebase dispatch');
		process.exit(1);
	}

	options.firebase = {
		projectId: process.env.FCM_PROJECT_ID,
		credentialsPath: process.env.FCM_CREDENTIALS_PATH,
		credentialsJson: process.env.FCM_CREDENTIALS_JSON,
		maxConcurrency: Number(process.env.FCM_MAX_CONCURRENCY || 100),
		requestTimeoutMs: Number(process.env.FCM_REQUEST_TIMEOUT_MS || 10000),
		retryMax: Number(process.env.FCM_RETRY_MAX || 5),
		retryBaseMs: Number(process.env.FCM_RETRY_BASE_MS || 300)
	};
}

var httpRequestProcessor = new HttpRequestProcessor();
httpRequestProcessor.init(options);

var handleHttpRequest = httpRequestProcessor.handleHttpRequest.bind(httpRequestProcessor);

// create servers
var httpServer = http.createServer(handleHttpRequest);

httpServer.listen(listenPortHttp, function () {
	// Callback triggered when http server is successfully listening.
	console.log("HTTP server listening on port " + listenPortHttp);
});

export default httpServer;