/** OSINT feeds (AISStream, OpenSky) → cotStore. */ import WebSocket from 'ws' import { updateFromCot } from './cotStore.js' import { broadcastSubscriberSnapshots, getSubscriberBboxUnion, } from './cotSubscribers.js' import { getCotSnapshotOpts } from './cotSnapshot.js' import { openSkyStateToCot, aisStreamMessageToCot, clampBbox } from './cotEntityUtils.js' import { OPENSKY_CACHE_MS, TRACKING_FEED_DEBOUNCE_MS, MAX_OPENSKY_BBOX_DEGREES } from './constants.js' const COALESCE_MS = 150 const state = { aisSocket: null, aisReconnectTimer: null, aisBackoffMs: 1000, openSkyTimer: null, bboxDebounceTimer: null, coalesceTimer: null, openSkyToken: null, openSkyTokenExpiresAt: 0, /** @type {Map} */ openSkyCache: new Map(), lastAisBbox: null, stopped: false, } function getConfig() { return useRuntimeConfig() } function openskyPollMs() { return OPENSKY_CACHE_MS } async function fetchOpenSkyToken(clientId, clientSecret) { const now = Date.now() if (state.openSkyToken && state.openSkyTokenExpiresAt > now + 60_000) { return state.openSkyToken } const body = new URLSearchParams({ grant_type: 'client_credentials', client_id: clientId, client_secret: clientSecret, }) const res = await fetch('https://auth.opensky-network.org/auth/realms/opensky-network/protocol/openid-connect/token', { method: 'POST', headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, body, }) if (!res.ok) throw new Error(`OpenSky auth failed: ${res.status}`) const json = await res.json() state.openSkyToken = json.access_token state.openSkyTokenExpiresAt = now + (Number(json.expires_in) || 1800) * 1000 return state.openSkyToken } function cacheKeyForBbox(bbox) { const round = n => Math.round(n * 10) / 10 return `${round(bbox.west)},${round(bbox.south)},${round(bbox.east)},${round(bbox.north)}` } async function fetchOpenSkyForBbox(bbox) { const config = getConfig() const clientId = config.openskyClientId const clientSecret = config.openskyClientSecret const cacheMs = openskyPollMs() const key = cacheKeyForBbox(bbox) const cached = state.openSkyCache.get(key) if (cached && Date.now() - cached.fetchedAt < cacheMs) { return cached.states } const params = new URLSearchParams({ lamin: String(bbox.south), lomin: String(bbox.west), lamax: String(bbox.north), lomax: String(bbox.east), }) const headers = {} if (clientId && clientSecret) { const token = await fetchOpenSkyToken(clientId, clientSecret) headers.Authorization = `Bearer ${token}` } const res = await fetch(`https://opensky-network.org/api/states/all?${params}`, { headers }) if (!res.ok) { console.error('[trackingFeed] OpenSky fetch failed:', res.status) return [] } const json = await res.json() const states = Array.isArray(json?.states) ? json.states : [] state.openSkyCache.set(key, { fetchedAt: Date.now(), states }) return states } export function scheduleCoalescedSnapshot() { if (state.coalesceTimer) clearTimeout(state.coalesceTimer) state.coalesceTimer = setTimeout(() => { state.coalesceTimer = null broadcastSubscriberSnapshots(getCotSnapshotOpts()).catch(() => {}) }, COALESCE_MS) } async function ingestOpenSkyBbox(bbox) { try { const states = await fetchOpenSkyForBbox(bbox) for (const row of states) { const parsed = openSkyStateToCot(row) if (parsed) await updateFromCot(parsed, { silent: true }) } if (states.length > 0) scheduleCoalescedSnapshot() } catch (err) { console.error('[trackingFeed] OpenSky error:', err?.message) } } async function pollOpenSky() { if (state.stopped) return const union = getSubscriberBboxUnion() const bbox = clampBbox(union, MAX_OPENSKY_BBOX_DEGREES) if (!bbox) return await ingestOpenSkyBbox(bbox) } function stopOpenSkyPoll() { if (state.openSkyTimer) { clearInterval(state.openSkyTimer) state.openSkyTimer = null } } /** Tests only. @internal */ export function scheduleOpenSkyPollForTests() { scheduleOpenSkyPoll() } function scheduleOpenSkyPoll() { if (state.openSkyTimer || state.stopped) return if (!getSubscriberBboxUnion()) return const intervalMs = openskyPollMs() state.openSkyTimer = setInterval(() => { pollOpenSky().catch((err) => { console.error('[trackingFeed] OpenSky poll error:', err?.message) }) }, intervalMs) } function aisBboxKey(bbox) { if (!bbox) return null return `${bbox.west},${bbox.south},${bbox.east},${bbox.north}` } function closeAisSocket() { if (state.aisSocket) { try { state.aisSocket.removeAllListeners() state.aisSocket.close() } catch { /* ignore */ } state.aisSocket = null } } function scheduleAisReconnect() { if (state.stopped || state.aisReconnectTimer) return state.aisReconnectTimer = setTimeout(() => { state.aisReconnectTimer = null connectAisStream() }, state.aisBackoffMs) state.aisBackoffMs = Math.min(state.aisBackoffMs * 2, 60_000) } function subscribeAisBbox(ws, bbox) { const apiKey = getConfig().aisstreamApiKey if (!apiKey || !bbox) return const key = aisBboxKey(bbox) if (key === state.lastAisBbox) return state.lastAisBbox = key ws.send(JSON.stringify({ APIKey: apiKey, BoundingBoxes: [[[bbox.south, bbox.west], [bbox.north, bbox.east]]], })) } function connectAisStream() { const apiKey = getConfig().aisstreamApiKey if (!apiKey || state.stopped) return closeAisSocket() const ws = new WebSocket('wss://stream.aisstream.io/v0/stream') state.aisSocket = ws ws.on('open', () => { state.aisBackoffMs = 1000 const union = getSubscriberBboxUnion() const bbox = clampBbox(union, MAX_OPENSKY_BBOX_DEGREES) if (bbox) subscribeAisBbox(ws, bbox) }) ws.on('message', (data) => { try { const message = JSON.parse(data.toString()) const parsed = aisStreamMessageToCot(message) if (parsed) updateFromCot(parsed).catch(() => {}) } catch { /* ignore malformed AIS messages */ } }) ws.on('close', () => { state.aisSocket = null if (!state.stopped) scheduleAisReconnect() }) ws.on('error', (err) => { console.error('[trackingFeed] AISStream error:', err?.message) }) } function refreshFeedBboxes() { if (state.stopped) return const union = getSubscriberBboxUnion() const bbox = clampBbox(union, MAX_OPENSKY_BBOX_DEGREES) if (!bbox) { stopOpenSkyPoll() return } if (getConfig().aisstreamApiKey) { if (state.aisSocket?.readyState === WebSocket.OPEN) { subscribeAisBbox(state.aisSocket, bbox) } else if (!state.aisSocket) { connectAisStream() } } pollOpenSky().catch(() => {}) scheduleOpenSkyPoll() } export function scheduleTrackingFeedRefresh() { if (state.bboxDebounceTimer) clearTimeout(state.bboxDebounceTimer) state.bboxDebounceTimer = setTimeout(() => { state.bboxDebounceTimer = null refreshFeedBboxes() }, TRACKING_FEED_DEBOUNCE_MS) } export function startTrackingFeed() { state.stopped = false } export function stopTrackingFeed() { state.stopped = true if (state.bboxDebounceTimer) clearTimeout(state.bboxDebounceTimer) if (state.coalesceTimer) clearTimeout(state.coalesceTimer) state.coalesceTimer = null stopOpenSkyPoll() if (state.aisReconnectTimer) clearTimeout(state.aisReconnectTimer) state.aisReconnectTimer = null closeAisSocket() state.openSkyCache.clear() state.lastAisBbox = null } /** Tests only. @internal */ export function resetTrackingFeedForTests() { stopTrackingFeed() state.stopped = false } /** Tests only. @internal */ export function isOpenSkyPollActive() { return state.openSkyTimer != null }