Merge pull request #33 from sebastiankrll/enhance/ws-bandwith

Enhance/ws bandwith

Author: sebastiankrll

apps/api/src/index.ts

@@ -169,7 +169,10 @@ app.get(
 			return;
 		}
 
-		res.json(all);
+		const buffer = Buffer.from(all, "base64");
+		res.setHeader("Content-Type", "application/octet-stream");
+		res.setHeader("Content-Encoding", "gzip");
+		res.send(buffer);
 	}),
 );
 

apps/ingestion/src/airport.ts

@@ -31,9 +31,8 @@ const TAF_URL = "https://aviationweather.gov/data/cache/tafs.cache.xml.gz";
 const WEATHER_FETCH_INTERVAL = 600_000;
 
 let cached: AirportLong[] = [];
-let deleted: string[] = [];
 let updated: AirportShort[] = [];
-let added: AirportShort[] = [];
+let added: Required<AirportShort>[] = [];
 
 export async function mapAirports(pilotsLong: PilotLong[]): Promise<AirportLong[]> {
 	await updateWeather();
@@ -126,31 +125,48 @@ export async function mapAirports(pilotsLong: PilotLong[]): Promise<AirportLong[
 	}
 
 	const airportsLong = Object.values(airportRecord);
+	setAirportDelta(airportsLong);
 
-	const deletedLong = cached.filter((a) => !airportsLong.some((b) => b.icao === a.icao));
-	const addedLong = airportsLong.filter((a) => !cached.some((b) => b.icao === a.icao));
-	const updatedLong = airportsLong.filter((a) => cached.some((b) => b.icao === a.icao));
+	return airportsLong;
+}
 
-	deleted = deletedLong.map((a) => a.icao);
-	added = addedLong.map(getAirportShort);
-	updated = updatedLong.map(getAirportShort);
-	// console.log(airportsLong[0])
+function setAirportDelta(airportsLong: AirportLong[]): void {
+	updated = [];
+	added = [];
+
+	for (const a of airportsLong) {
+		const cachedAirport = cached.find((c) => c.icao === a.icao);
+
+		if (!cachedAirport) {
+			added.push(getAirportShort(a) as Required<AirportShort>);
+		} else {
+			const airportShort = getAirportShort(a, cachedAirport);
+			updated.push(airportShort);
+		}
+	}
 
 	cached = airportsLong;
-	return airportsLong;
 }
 
-export function getAirportShort(a: AirportLong): AirportShort {
-	return {
-		icao: a.icao,
-		dep_traffic: a.dep_traffic,
-		arr_traffic: a.arr_traffic,
-	};
+export function getAirportShort(a: AirportLong, c?: AirportLong): AirportShort {
+	if (!c) {
+		return {
+			icao: a.icao,
+			dep_traffic: a.dep_traffic,
+			arr_traffic: a.arr_traffic,
+		};
+	} else {
+		const airportShort: AirportShort = { icao: a.icao };
+
+		if (JSON.stringify(a.dep_traffic) !== JSON.stringify(c.dep_traffic)) airportShort.dep_traffic = a.dep_traffic;
+		if (JSON.stringify(a.arr_traffic) !== JSON.stringify(c.arr_traffic)) airportShort.arr_traffic = a.arr_traffic;
+
+		return airportShort;
+	}
 }
 
 export function getAirportDelta(): AirportDelta {
 	return {
-		deleted,
 		added,
 		updated,
 	};

apps/ingestion/src/controller.ts

@@ -3,28 +3,30 @@ import type { FIRFeature, SimAwareTraconFeature } from "@sr24/types/db";
 import type { ControllerDelta, ControllerLong, ControllerMerged, ControllerShort, PilotLong, VatsimData } from "@sr24/types/vatsim";
 import { haversineDistance } from "./utils/helpers.js";
 
-let cachedMerged: ControllerMerged[] = [];
-let deleted: string[] = [];
+let cached: ControllerMerged[] = [];
 let updated: ControllerMerged[] = [];
 let added: ControllerMerged[] = [];
 
 export async function mapControllers(vatsimData: VatsimData, pilotsLong: PilotLong[]): Promise<[ControllerLong[], ControllerMerged[]]> {
-	const controllersLong: ControllerLong[] = vatsimData.controllers.map((controller) => {
-		return {
-			callsign: controller.callsign,
-			frequency: parseFrequencyToKHz(controller.frequency),
-			facility: controller.facility,
-			atis: controller.text_atis,
-			connections: 0,
-			cid: controller.cid,
-			name: controller.name,
-			rating: controller.rating,
-			server: controller.server,
-			visual_range: controller.visual_range,
-			logon_time: new Date(controller.logon_time),
-			timestamp: new Date(controller.last_updated),
-		};
-	});
+	const controllersLong: ControllerLong[] = vatsimData.controllers
+		.map((controller) => {
+			if (controller.facility === 0 && !controller.callsign.includes("OBS")) return null;
+			return {
+				callsign: controller.callsign,
+				frequency: parseFrequencyToKHz(controller.frequency),
+				facility: controller.facility,
+				atis: controller.text_atis,
+				connections: 0,
+				cid: controller.cid,
+				name: controller.name,
+				rating: controller.rating,
+				server: controller.server,
+				visual_range: controller.visual_range,
+				logon_time: new Date(controller.logon_time),
+				timestamp: new Date(controller.last_updated),
+			};
+		})
+		.filter((c) => c !== null);
 
 	getConnectionsCount(vatsimData, controllersLong, pilotsLong);
 
@@ -46,19 +48,63 @@ export async function mapControllers(vatsimData: VatsimData, pilotsLong: PilotLo
 	});
 
 	const merged = await mergeControllers(controllersLong);
+	setControllerDelta(merged);
 
-	const deletedMerged = cachedMerged.filter((a) => !merged.some((b) => b.id === a.id));
-	deleted = deletedMerged.map((c) => c.id);
-	added = merged.filter((a) => !cachedMerged.some((b) => b.id === a.id));
-	updated = merged.filter((a) => cachedMerged.some((b) => b.id === a.id));
-
-	cachedMerged = merged;
 	return [controllersLong, merged];
 }
 
+function setControllerDelta(merged: ControllerMerged[]): void {
+	added = [];
+	updated = [];
+
+	for (const m of merged) {
+		const cachedMerged = cached.find((c) => c.id === m.id);
+		if (!cachedMerged) {
+			added.push(m);
+		} else {
+			const updatedControllers: ControllerShort[] = [];
+
+			for (const controller of m.controllers) {
+				const cachedController = cachedMerged.controllers.find((c) => c.callsign === controller.callsign);
+				const controllerShort = getControllerShort(controller, cachedController);
+				updatedControllers.push(controllerShort);
+			}
+
+			if (updatedControllers.length > 0) {
+				updated.push({
+					id: m.id,
+					facility: m.facility,
+					controllers: updatedControllers,
+				});
+			}
+		}
+	}
+
+	cached = merged;
+}
+
+function getControllerShort(controller: ControllerShort, cachedController?: ControllerShort): ControllerShort {
+	if (!cachedController) {
+		return {
+			callsign: controller.callsign,
+			frequency: controller.frequency,
+			facility: controller.facility,
+			atis: controller.atis,
+			connections: controller.connections,
+		};
+	} else {
+		const controllerShort: ControllerShort = { callsign: controller.callsign, facility: controller.facility };
+
+		if (controller.frequency !== cachedController.frequency) controllerShort.frequency = controller.frequency;
+		if (JSON.stringify(controller.atis) !== JSON.stringify(cachedController.atis)) controllerShort.atis = controller.atis;
+		if (controller.connections !== cachedController.connections) controllerShort.connections = controller.connections;
+
+		return controllerShort;
+	}
+}
+
 export function getControllerDelta(): ControllerDelta {
 	return {
-		deleted,
 		added,
 		updated,
 	};

apps/ingestion/src/index.ts

@@ -1,7 +1,9 @@
 import "dotenv/config";
+import { promisify } from "node:util";
+import { gzip } from "node:zlib";
 import { pgDeleteStalePilots, pgUpsertPilots } from "@sr24/db/pg";
 import { rdsConnect, rdsPub, rdsSetMultiple, rdsSetMultipleTimeSeries, rdsSetSingle } from "@sr24/db/redis";
-import type { TrackPoint, VatsimData, VatsimTransceivers, WsAll, WsDelta } from "@sr24/types/vatsim";
+import type { AirportShort, PilotShort, TrackPoint, VatsimData, VatsimTransceivers, WsAll, WsDelta } from "@sr24/types/vatsim";
 import axios from "axios";
 import { getAirportDelta, getAirportShort, mapAirports } from "./airport.js";
 import { getControllerDelta, mapControllers } from "./controller.js";
@@ -12,6 +14,8 @@ const VATSIM_DATA_URL = "https://data.vatsim.net/v3/vatsim-data.json";
 const VATSIM_TRANSCEIVERS_URL = "https://data.vatsim.net/v3/transceivers-data.json";
 const FETCH_INTERVAL = 5_000;
 
+const gzipAsync = promisify(gzip);
+
 let dbsInitialized = false;
 let updating = false;
 let lastVatsimUpdate = 0;
@@ -47,15 +51,17 @@ async function fetchVatsimData(): Promise<void> {
 				airports: getAirportDelta(),
 				controllers: getControllerDelta(),
 			};
-			rdsPub("ws:delta", delta);
+			const gzDelta = await gzipAsync(JSON.stringify(delta));
+			rdsPub("ws:delta", gzDelta.toString("base64"));
 
 			// Set full websocket data on redis ws:all
 			const all: WsAll = {
-				pilots: pilotsLong.map(getPilotShort),
-				airports: airportsLong.map(getAirportShort),
+				pilots: pilotsLong.map((p) => getPilotShort(p) as Required<PilotShort>),
+				airports: airportsLong.map((a) => getAirportShort(a) as Required<AirportShort>),
 				controllers: controllersMerged,
 			};
-			rdsSetSingle("ws:all", all);
+			const gzAll = await gzipAsync(JSON.stringify(all));
+			rdsSetSingle("ws:all", gzAll.toString("base64"));
 
 			// Set pilots, controllers and airports data in redis
 			rdsSetMultiple(pilotsLong, "pilot", (p) => p.id, "pilots:live", 120);

apps/ingestion/src/pilot.ts

@@ -80,15 +80,10 @@ const MILITARY_RATINGS = [
 ];
 
 let cached: PilotLong[] = [];
-let deleted: string[] = [];
 let updated: PilotShort[] = [];
-let added: PilotShort[] = [];
+let added: Required<PilotShort>[] = [];
 
 export async function mapPilots(latestVatsimData: VatsimData): Promise<[PilotLong[], PilotLong[]]> {
-	deleted = [];
-	updated = [];
-	added = [];
-
 	const pilotsLongPromises: Promise<PilotLong>[] = latestVatsimData.pilots.map(async (pilot) => {
 		const id = getPilotId(pilot);
 		const cachedPilot = cached.find((c) => c.id === id);
@@ -145,19 +140,12 @@ export async function mapPilots(latestVatsimData: VatsimData): Promise<[PilotLon
 	});
 
 	const pilotsLong = await Promise.all(pilotsLongPromises);
-
 	const deletedLong = cached.filter((a) => !pilotsLong.some((b) => b.id === a.id));
 	deletedLong.forEach((p) => {
 		p.live = false;
 	});
 
-	const addedLong = pilotsLong.filter((a) => !cached.some((b) => b.id === a.id));
-	const updatedLong = pilotsLong.filter((a) => cached.some((b) => b.id === a.id));
-
-	added = addedLong.map(getPilotShort);
-	updated = updatedLong.map(getPilotShort);
-	deleted = deletedLong.map((p) => p.id);
-	cached = pilotsLong;
+	setPilotDelta(pilotsLong);
 
 	return [pilotsLong, deletedLong];
 }
@@ -175,31 +163,67 @@ function getPilotId(pilot: VatsimPilot): string {
 	return b64url.slice(0, 10);
 }
 
+function setPilotDelta(pilotsLong: PilotLong[]): void {
+	added = [];
+	updated = [];
+
+	for (const p of pilotsLong) {
+		const cachedPilot = cached.find((c) => c.id === p.id);
+		if (!cachedPilot) {
+			added.push(getPilotShort(p) as Required<PilotShort>);
+		} else {
+			const pilotShort = getPilotShort(p, cachedPilot);
+			updated.push(pilotShort);
+		}
+	}
+
+	cached = pilotsLong;
+}
+
 export function getPilotDelta(): PilotDelta {
 	return {
-		deleted,
 		added,
 		updated,
 	};
 }
 
-export function getPilotShort(p: PilotLong): PilotShort {
-	return {
-		id: p.id,
-		latitude: p.latitude,
-		longitude: p.longitude,
-		altitude_agl: p.altitude_agl,
-		altitude_ms: p.altitude_ms,
-		groundspeed: p.groundspeed,
-		vertical_speed: p.vertical_speed,
-		heading: p.heading,
-		callsign: p.callsign,
-		aircraft: p.aircraft,
-		transponder: p.transponder,
-		frequency: p.frequency,
-		route: p.route,
-		ghost: p.ghost,
-	};
+export function getPilotShort(p: PilotLong, c?: PilotLong): PilotShort {
+	if (!c) {
+		return {
+			id: p.id,
+			latitude: p.latitude,
+			longitude: p.longitude,
+			altitude_agl: p.altitude_agl,
+			altitude_ms: p.altitude_ms,
+			groundspeed: p.groundspeed,
+			vertical_speed: p.vertical_speed,
+			heading: p.heading,
+			callsign: p.callsign,
+			aircraft: p.aircraft,
+			transponder: p.transponder,
+			frequency: p.frequency,
+			route: p.route,
+			ghost: p.ghost,
+		};
+	} else {
+		const pilotShort: PilotShort = { id: p.id };
+
+		if (p.latitude !== c.latitude) pilotShort.latitude = p.latitude;
+		if (p.longitude !== c.longitude) pilotShort.longitude = p.longitude;
+		if (p.altitude_agl !== c.altitude_agl) pilotShort.altitude_agl = p.altitude_agl;
+		if (p.altitude_ms !== c.altitude_ms) pilotShort.altitude_ms = p.altitude_ms;
+		if (p.groundspeed !== c.groundspeed) pilotShort.groundspeed = p.groundspeed;
+		if (p.vertical_speed !== c.vertical_speed) pilotShort.vertical_speed = p.vertical_speed;
+		if (p.heading !== c.heading) pilotShort.heading = p.heading;
+		if (p.callsign !== c.callsign) pilotShort.callsign = p.callsign;
+		if (p.aircraft !== c.aircraft) pilotShort.aircraft = p.aircraft;
+		if (p.transponder !== c.transponder) pilotShort.transponder = p.transponder;
+		if (p.frequency !== c.frequency) pilotShort.frequency = p.frequency;
+		if (p.route !== c.route) pilotShort.route = p.route;
+		if (p.ghost !== c.ghost) pilotShort.ghost = p.ghost;
+
+		return pilotShort;
+	}
 }
 
 function calculateVerticalSpeed(current: PilotLong, cache: PilotLong | undefined): number {