gigafibre-fsm/services/targo-hub/server.js

'use strict'
const http = require('http')
const { URL } = require('url')
const cfg = require('./lib/config')
const { log, json, parseBody } = require('./lib/helpers')
const sse = require('./lib/sse')
const twilio = require('./lib/twilio')
const pbx = require('./lib/pbx')
const telephony = require('./lib/telephony')
const devices = require('./lib/devices')
const provision = require('./lib/provision')
const auth = require('./lib/auth')
const conversation = require('./lib/conversation')
const traccar = require('./lib/traccar')
const dispatch = require('./lib/dispatch')
const ical = require('./lib/ical')
const vision = require('./lib/vision')
let voiceAgent
try { voiceAgent = require('./lib/voice-agent') } catch (e) { voiceAgent = null; console.log('Voice agent module not loaded:', e.message) }
// Oktopus stack is decommissioned. Set OKTOPUS_DISABLED=1 (default) to skip
// loading the module and the MQTT monitor — both would otherwise attempt
// connections to a dead broker and spam reconnect errors into stdout. The
// Mongo `devices` collection updates from MQTT heartbeats are no longer
// produced; nothing in ops or the field app reads from that collection.
// Re-enable by setting OKTOPUS_DISABLED=0 in the hub env.
const OKTOPUS_DISABLED = process.env.OKTOPUS_DISABLED !== '0'
let oktopus = null
let oktopusMqtt = null
if (!OKTOPUS_DISABLED) {
  try { oktopus = require('./lib/oktopus') } catch (e) { console.log('Oktopus module not loaded:', e.message) }
  try { oktopusMqtt = require('./lib/oktopus-mqtt') } catch (e) { console.log('Oktopus MQTT monitor not loaded:', e.message) }
} else {
  console.log('Oktopus integration disabled (OKTOPUS_DISABLED=1)')
}

const server = http.createServer(async (req, res) => {
  const url = new URL(req.url, `http://localhost:${cfg.PORT}`)
  const path = url.pathname
  const method = req.method

  res.setHeader('Access-Control-Allow-Origin', '*')
  res.setHeader('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS')
  res.setHeader('Access-Control-Allow-Headers', 'Content-Type, Authorization, X-Authentik-Email, X-Authentik-Groups')
  if (method === 'OPTIONS') { res.writeHead(204); return res.end() }

  try {
    if (path === '/health') {
      return json(res, 200, { ok: true, clients: sse.clientCount(), topics: sse.topicCount(), uptime: Math.floor(process.uptime()) })
    }

    if (path === '/sse' && method === 'GET') {
      const email = req.headers['x-authentik-email'] || 'anonymous'
      const topics = (url.searchParams.get('topics') || '').split(',').map(t => t.trim()).filter(Boolean)
      if (!topics.length) return json(res, 400, { error: 'Missing topics parameter' })
      res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive', 'X-Accel-Buffering': 'no' })
      res.write(': connected\n\n')
      sse.addClient(topics, res, email)
      const keepalive = setInterval(() => { try { res.write(': ping\n\n') } catch { clearInterval(keepalive) } }, 25000)
      res.on('close', () => clearInterval(keepalive))
      return
    }

    if (path === '/broadcast' && method === 'POST') {
      const a = req.headers.authorization || ''
      if (cfg.INTERNAL_TOKEN && a !== 'Bearer ' + cfg.INTERNAL_TOKEN) return json(res, 401, { error: 'Unauthorized' })
      const body = await parseBody(req)
      if (!body.topic || !body.event) return json(res, 400, { error: 'Missing topic or event' })
      return json(res, 200, { ok: true, delivered: sse.broadcast(body.topic, body.event, body.data || {}) })
    }

    if (path === '/webhook/twilio/sms-incoming' && method === 'POST') return twilio.smsIncoming(req, res)
    if (path === '/webhook/twilio/sms-status' && method === 'POST') return twilio.smsStatus(req, res)
    if (path === '/send/sms' && method === 'POST') return twilio.sendSms(req, res)
    if (path === '/voice/token' && method === 'GET') return twilio.voiceToken(req, res, url)
    if (path === '/voice/sip-config' && method === 'GET') return twilio.sipConfig(req, res)
    if (path === '/voice/twiml' && method === 'POST') return twilio.voiceTwiml(req, res)
    if (path === '/voice/status' && method === 'POST') return twilio.voiceStatus(req, res)
    if (path === '/voice/inbound' && method === 'POST') return voiceAgent.handleInboundCall(req, res)
    if (path === '/voice/gather' && method === 'POST') return voiceAgent.handleGather(req, res)
    if (path === '/voice/connect-agent' && method === 'POST') return voiceAgent.handleConnectAgent(req, res)
    if (path === '/webhook/3cx/call-event' && method === 'POST') return pbx.callEvent(req, res)

    // Uptime-Kuma webhook — synthesized outage alerts
    if (path === '/webhook/kuma' && method === 'POST') {
      const body = await parseBody(req)
      const { handleKumaWebhook } = require('./lib/outage-monitor')
      const result = await handleKumaWebhook(body)
      return json(res, 200, result)
    }
    if (path.startsWith('/telephony/')) return telephony.handle(req, res, method, path, url)
    if (path.startsWith('/devices')) return devices.handle(req, res, method, path, url)
    if (path.startsWith('/acs/')) return devices.handleACSConfig(req, res, method, path, url)
    if (path.startsWith('/provision/')) return provision.handle(req, res, method, path)
    if (path.startsWith('/oktopus/')) {
      if (!oktopus) return require('./lib/helpers').json(res, 410, { error: 'Oktopus integration removed' })
      return oktopus.handle(req, res, method, path)
    }
    if (path.startsWith('/auth/')) return auth.handle(req, res, method, path, url)
    if (path.startsWith('/conversations')) return conversation.handle(req, res, method, path, url)
    if (path.startsWith('/traccar')) return traccar.handle(req, res, method, path)
    if (path.startsWith('/address/')) return require('./lib/address-validate').handle(req, res, method, path)
    if (path.startsWith('/magic-link')) return require('./lib/magic-link').handle(req, res, method, path)
    if (path.startsWith('/portal/')) return require('./lib/portal-auth').handle(req, res, method, path)
    // Lightweight tech mobile page: /t/{token}[/action]
    if (path.startsWith('/t/')) return require('./lib/tech-mobile').route(req, res, method, path)
    if (path.startsWith('/accept')) return require('./lib/acceptance').handle(req, res, method, path)
    if (path.startsWith('/api/catalog') || path.startsWith('/api/checkout') || path.startsWith('/api/accept-for-client') || path.startsWith('/api/order') || path.startsWith('/api/address') || path.startsWith('/api/otp')) return require('./lib/checkout').handle(req, res, method, path)
    if (path.startsWith('/api/referral/')) return require('./lib/referral').handle(req, res, method, path)
    // iCal token: /dispatch/ical-token/TECH-001 (auth required — returns token for building URL)
    const icalTokenMatch = path.match(/^\/dispatch\/ical-token\/(.+)$/)
    if (icalTokenMatch && method === 'GET') {
      const techId = icalTokenMatch[1]
      const token = ical.generateToken(techId)
      return json(res, 200, { techId, token, url: `/dispatch/calendar/${techId}.ics?token=${token}` })
    }
    // iCal feed: /dispatch/calendar/TECH-001.ics?token=xxx (token auth, no SSO)
    const icalMatch = path.match(/^\/dispatch\/calendar\/(.+)\.ics$/)
    if (icalMatch && method === 'GET') return ical.handleCalendar(req, res, icalMatch[1], url.searchParams)
    if (path.startsWith('/dispatch')) return dispatch.handle(req, res, method, path)
    if (path.startsWith('/admin/pollers')) return require('./lib/poller-control').handle(req, res, method, path)
    if (path.startsWith('/reports')) return require('./lib/reports').handle(req, res, method, path, url)
    if (path.startsWith('/campaigns')) return require('./lib/campaigns').handle(req, res, method, path)
    // Gift redirect wrapper — short public URLs in campaign emails that
    // 302 to the underlying Giftbit shortlink (subject to our expiry/revoke).
    if (path.startsWith('/g/') && method === 'GET') return require('./lib/campaigns').handleGiftRedirect(req, res, path)
    if (path.startsWith('/contract')) return require('./lib/contracts').handle(req, res, method, path)
    if (path.startsWith('/payments') || path === '/webhook/stripe') return require('./lib/payments').handle(req, res, method, path, url)
    if (path === '/vision/barcodes' && method === 'POST') return vision.handleBarcodes(req, res)
    if (path === '/vision/equipment' && method === 'POST') return vision.handleEquipment(req, res)
    if (path === '/vision/invoice'   && method === 'POST') return vision.handleInvoice(req, res)
    if (path.startsWith('/ai/')) return require('./lib/ai').handle(req, res, method, path)
    if (path.startsWith('/modem')) return require('./lib/modem-bridge').handleModemRequest(req, res, path)
    if (path.startsWith('/network/')) return require('./lib/network-intel').handle(req, res, method, path)
    if (path.startsWith('/agent/')) return require('./lib/agent').handleAgentApi(req, res, method, path)
    if (path.startsWith('/flow/templates')) return require('./lib/flow-templates').handle(req, res, method, path, url)
    if (path.startsWith('/flow/runs') || path === '/flow/start' || path === '/flow/advance' || path === '/flow/complete' || path === '/flow/event') {
      return require('./lib/flow-api').handle(req, res, method, path, url)
    }

    if (path.startsWith('/c/') && method === 'GET') {
      const fs = require('fs')
      const chatPath = require('path').join(__dirname, 'public', 'chat.html')
      try {
        const html = fs.readFileSync(chatPath, 'utf8')
        res.writeHead(200, { 'Content-Type': 'text/html; charset=utf-8', 'Cache-Control': 'no-cache' })
        return res.end(html)
      } catch { return json(res, 404, { error: 'Chat page not found' }) }
    }

    if (path.startsWith('/olt')) {
      try {
        const oltSnmp = require('./lib/olt-snmp')
        const oltParts = path.replace('/olt', '').split('/').filter(Boolean)

        if ((!oltParts.length || oltParts[0] === 'stats') && method === 'GET') {
          return json(res, 200, { olts: oltSnmp.getOltStats() })
        }
        if (oltParts[0] === 'onus' && method === 'GET') {
          const serial = url.searchParams.get('serial')
          if (serial) return json(res, 200, oltSnmp.getOnuBySerial(serial) || { error: 'ONU not found' })
          return json(res, 200, { onus: oltSnmp.getAllOnus() })
        }
        if (oltParts[0] === 'poll' && method === 'POST') {
          oltSnmp.pollAllOlts().catch(e => log('Manual OLT poll error:', e.message))
          return json(res, 200, { ok: true, message: 'OLT poll triggered' })
        }
        if ((oltParts[0] === 'register' || oltParts[0] === 'config') && method === 'POST') {
          const body = await parseBody(req)
          if (!body.host || !body.community) return json(res, 400, { error: 'host and community required' })
          oltSnmp.registerOlt(body)
          return json(res, 200, { ok: true })
        }
        return json(res, 404, { error: 'OLT endpoint not found' })
      } catch (e) {
        return json(res, 500, { error: 'OLT module error: ' + e.message })
      }
    }

    json(res, 404, { error: 'Not found' })
  } catch (e) {
    log('ERROR:', e.message)
    json(res, 500, { error: 'Internal error' })
  }
})

// WebSocket upgrade for Twilio Media Streams
server.on('upgrade', (req, socket, head) => {
  const url = new URL(req.url, `http://localhost:${cfg.PORT}`)
  if (url.pathname === '/voice/ws' && voiceAgent) {
    const { WebSocketServer } = require('ws')
    const wss = new WebSocketServer({ noServer: true })
    wss.handleUpgrade(req, socket, head, (ws) => voiceAgent.handleMediaStream(ws, req))
  } else {
    socket.destroy()
  }
})

server.listen(cfg.PORT, '0.0.0.0', () => {
  log(`targo-hub listening on :${cfg.PORT}`)
  if (voiceAgent) log('Voice agent: enabled')
  pbx.startPoller()
  devices.startPoller()
  try { require('./lib/olt-snmp').startOltPoller() }
  catch (e) { log('OLT SNMP poller failed to start:', e.message) }
  if (oktopusMqtt) oktopusMqtt.start()
  else log('Oktopus MQTT monitor: skipped (disabled)')
  // Start PPA (pre-authorized payment) cron scheduler
  try { require('./lib/payments').startPPACron() }
  catch (e) { log('PPA cron failed to start:', e.message) }
})