kestrelos/server/utils/trackingFeed.js

/** 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<string, { fetchedAt: number, states: unknown[][] }>} */
  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
}