import json
import time
import queue
import threading
import signal
import sys
import logging
import websocket

logging.basicConfig(
    format='%(asctime)s.%(msecs)03d %(levelname)s: %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    level=logging.INFO,
    stream=sys.stdout
)
logger = logging.getLogger(__name__)

SITE_ID = 'your-site-id'
API_TOKEN = 'your-api-token'  # use secrets manager in production

message_queue = queue.Queue()

def on_open(ws):
    ws.send(json.dumps({'subscribe': f'/sites/{SITE_ID}/stats/clients'}))
    logger.info("Subscribed to channel")

def on_message(ws, message):
    message_queue.put(message)  # never block here

def on_error(ws, error):
    logger.error(f"WebSocket error: {error}")

def on_close(ws, code, msg):
    logger.info(f"WebSocket closed — code: {code}, msg: {msg}")
    if code == 1006:
        logger.warning("Abnormal closure — will reconnect")

def on_ping(ws, _):
    logger.debug("Ping received")

def on_pong(ws, _):
    logger.debug("Pong received")

def worker():
    count = 0
    while True:
        msg = message_queue.get(block=True)
        if msg is None:
            continue
        try:
            data = json.loads(msg)
            # TODO: send to your downstream sink
            count += 1
            if count % 100 == 0:
                logger.info(f"Processed {count} messages. Queue depth: {message_queue.qsize()}")
        except Exception as e:
            logger.error(f"Error processing message: {e}")

def establish_ws():
    ws = websocket.WebSocketApp(
        'wss://api-ws.mist.com/api-ws/v1/stream',
        header=[f'Authorization: Token {API_TOKEN}'],
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
        on_ping=on_ping,
        on_pong=on_pong,
    )
    ws.run_forever(ping_interval=60, ping_timeout=45)

if __name__ == '__main__':
    signal.signal(signal.SIGINT, lambda s, f: sys.exit(0))
    signal.signal(signal.SIGTERM, lambda s, f: sys.exit(0))
    threading.Thread(target=worker, daemon=True).start()
    while True:
        try:
            establish_ws()
        except Exception as e:
            logger.error(f"Connection error: {e}")
        logger.info("Reconnecting in 5 seconds...")
        time.sleep(5)

// Requires: npm install ws
const WebSocket = require('ws');

const API_TOKEN = process.env.MIST_API_TOKEN;
const SITE_ID = process.env.MIST_SITE_ID;

if (!API_TOKEN || !SITE_ID) {
    throw new Error('Set MIST_API_TOKEN and MIST_SITE_ID before running this script.');
}

const WS_URL = 'wss://api-ws.mist.com/api-ws/v1/stream';
const SUBSCRIPTION = `/sites/${SITE_ID}/stats/clients`;
const RECONNECT_DELAY_MS = 5000;
const SUBSCRIBE_ACK_TIMEOUT_MS = 10000;
const LOG_EVERY_N_MESSAGES = 100;
const QUEUE_WARN_THRESHOLD = 5000;

const messageQueue = [];
let workerScheduled = false;
let processedCount = 0;
let reconnectTimer = null;
let ws = null;

function log(message, extra = '') {
    const suffix = extra ? ` ${extra}` : '';
    console.log(`${new Date().toISOString()} ${message}${suffix}`);
}

function scheduleWorker() {
    if (workerScheduled) return;
    workerScheduled = true;
    setImmediate(processQueue);
}

async function processQueue() {
    workerScheduled = false;

    while (messageQueue.length > 0) {
        const raw = messageQueue.shift();
        try {
            const event = JSON.parse(raw);
            void event;
            // TODO: send to downstream sink (Kafka/Kinesis/DB/etc.)
            processedCount += 1;
            if (processedCount % LOG_EVERY_N_MESSAGES === 0) {
                log('worker throughput', `processed=${processedCount} queue_depth=${messageQueue.length}`);
            }
        } catch (err) {
            log('worker processing error', err.message);
        }

        // Yield so socket callbacks keep running under load.
        await new Promise((resolve) => setImmediate(resolve));
    }
}

function connect() {
    let subscriptionConfirmed = false;
    let subscribeAckTimer = null;

    ws = new WebSocket(WS_URL, {
        headers: {
            Authorization: `Token ${API_TOKEN}`,
        },
    });

    ws.on('open', () => {
        log('websocket opened');
        ws.send(JSON.stringify({ subscribe: SUBSCRIPTION }));

        subscribeAckTimer = setTimeout(() => {
            if (!subscriptionConfirmed) {
                log('subscription ack timeout, reconnecting');
                ws.close();
            }
        }, SUBSCRIBE_ACK_TIMEOUT_MS);
    });

    ws.on('message', (buffer) => {
        const raw = buffer.toString();

        // Keep receive path fast: enqueue first, process elsewhere.
        messageQueue.push(raw);
        if (messageQueue.length === QUEUE_WARN_THRESHOLD || messageQueue.length % 10000 === 0) {
            log('queue depth warning', `queue_depth=${messageQueue.length}`);
        }
        scheduleWorker();

        // Lightweight control-plane handling in receive path.
        try {
            const msg = JSON.parse(raw);
            if (msg.event === 'channel_subscribed' && msg.channel === SUBSCRIPTION) {
                subscriptionConfirmed = true;
                clearTimeout(subscribeAckTimer);
                log('channel_subscribed', msg.channel);
            } else if (msg.event === 'subscribe_failed') {
                log('subscribe_failed', `${msg.channel || ''} ${msg.detail || ''}`.trim());
                ws.close();
            } else if (msg.event === 'data') {
                // TODO: send to your downstream sink
            }
        } catch (err) {
            log('control-plane parse error', err.message);
        }
    });

    ws.on('ping', () => {
        log('ping received');
    });

    ws.on('pong', () => {
        log('pong received');
    });

    ws.on('error', (err) => {
        log('websocket error', err.message);
    });

    ws.on('close', (code, reasonBuffer) => {
        clearTimeout(subscribeAckTimer);
        const reason = reasonBuffer ? reasonBuffer.toString() : '';
        log('websocket closed', `code=${code} reason=${reason}`.trim());

        if (!reconnectTimer) {
            reconnectTimer = setTimeout(() => {
                reconnectTimer = null;
                log('reconnecting');
                connect();
            }, RECONNECT_DELAY_MS);
        }
    });
}

function shutdown() {
    log('shutdown requested');
    if (reconnectTimer) {
        clearTimeout(reconnectTimer);
        reconnectTimer = null;
    }
    if (ws && ws.readyState === WebSocket.OPEN) {
        ws.close(1000, 'shutdown');
    }
    process.exit(0);
}

process.on('SIGINT', shutdown);
process.on('SIGTERM', shutdown);

connect();