Introduction

Real-time WebSocket feed for crypto exchange listing announcements.

What is this?

This service delivers instant notifications when a cryptocurrency exchange publishes an announcement – listings and other exchange announcements. You connect via WebSocket and receive structured JSON messages in real time.

Currently supported exchanges:

Supported announcement types:

How it works

Exchange publishes announcement
        |
        v
   Detection engine (< 5ms)
        |
        v
   Dispatch server (< 1ms)
        |
        v
   Your WebSocket client

1. Our detection engine monitors exchange announcement pages continuously.
2. When a new listing is detected, it is parsed, classified, and forwarded to the dispatch server.
3. The dispatch server broadcasts the announcement to all connected WebSocket subscribers simultaneously.
4. You receive a JSON message with the ticker, exchange, listing type, and precise timestamps.

Key features

• Ultra-low latency – sub-millisecond dispatch from detection to your connection.
• Precise timestamps – microsecond-resolution timestamps at every stage (publish, detect, dispatch) so you can measure your exact latency.
• Exchange filtering – subscribe only to the exchanges you care about.
• Always-on heartbeat – know immediately if your connection drops.
• Zero-copy broadcast – announcements are serialized once and shared across all subscribers for maximum throughput.

Quick links

Quick Start

Get connected and receiving announcements in under 5 minutes.

1. Get your API key

You will receive an API key from the administrator. It looks like this:

dsk_a1b2c3d4e5f67890abcdef1234567890abcdef1234567890abcdef1234567890ab

All keys start with the dsk_ prefix followed by 64 hexadecimal characters.

2. Connect to the WebSocket

wss://cryptolisting.ws

Pass your API key via the X-API-Key HTTP header during the WebSocket handshake.

pip install websocket-client

# pip install websocket-client
import json, websocket
from datetime import datetime

URL    = "wss://cryptolisting.ws"
HEADER = ["X-API-Key: dsk_your_key_here"]

def on_open(ws):
    print("Connected!")

def on_message(ws, message):
    data = json.loads(message)
    ts   = datetime.now().strftime("%H:%M:%S.%f")[:-3]

    if data["type"] == "announcement":
        print(f"[{ts}] {data['listingType']} | {data['ticker']} on {data['publisher']}")
        print(f"        {data['title']}")
    elif data["type"] == "welcome":
        print(f"[{ts}] Welcome — tier={data['tier']}, cex={data['allowedCex']}")

def on_close(ws, code, msg):
    print(f"Disconnected ({code}). Reconnecting...")
    connect()

def connect():
    websocket.WebSocketApp(
        URL, header=HEADER,
        on_open=on_open, on_message=on_message, on_close=on_close
    ).run_forever(ping_interval=30, ping_timeout=10)

connect()

npm install ws

const WebSocket = require("ws");

const URL = "wss://cryptolisting.ws";

function connect() {
  const ws = new WebSocket(URL, {
    headers: { "X-API-Key": "dsk_your_key_here" },
  });

  ws.on("open", () => console.log("Connected!"));

  ws.on("message", (raw) => {
    const data = JSON.parse(raw);
    if (data.type === "announcement") {
      console.log(`${data.listingType} | ${data.ticker} on ${data.publisher}`);
      console.log(`  ${data.title}`);
    } else if (data.type === "welcome") {
      console.log(`Welcome — tier=${data.tier}, cex=${data.allowedCex}`);
    }
  });

  ws.on("close", () => {
    console.log("Disconnected. Reconnecting...");
    setTimeout(connect, 1000);
  });
}

connect();

3. Receive messages

Once connected, you will immediately receive a welcome message confirming your subscription:

{
  "type": "welcome",
  "tier": "premium",
  "maxConnections": 5,
  "allowedCex": "*",
  "expiresInSecs": 86400
}

Then you will receive:

• Heartbeats every 30 seconds – confirms the connection is alive.
• Announcements whenever a listing or other exchange announcement is detected.

{
  "type": "announcement",
  "title": "Binance Will List TOKEN (TOKEN)",
  "ticker": "TOKEN",
  "publisher": "binance",
  "listingType": "spot_listing",
  "publishTimestampUs": 1710345000000000,
  "detectedTimestampUs": 1710345000005000,
  "dispatchTimestampUs": 1710345000006000
}

4. Filter by exchange (optional)

To receive announcements only from specific exchanges, add the cex query parameter:

wss://cryptolisting.ws?cex=binance
wss://cryptolisting.ws?cex=binance,upbit

Omit the parameter to receive announcements from all exchanges.

Next steps

• Message Reference – full field-by-field documentation
• Error Handling & Reconnection – build a resilient client
• Code Examples – production-ready implementations

Authentication

Every WebSocket connection must be authenticated with a valid API key.

API key format

dsk_<64 hex characters>

Example:

dsk_a1b2c3d4e5f67890abcdef1234567890abcdef1234567890abcdef1234567890ab

• Prefix: dsk_ (always)
• Body: 64 hexadecimal characters (32 random bytes)

Passing your API key

Method 1: HTTP header (recommended)

Set the X-API-Key header during the WebSocket handshake:

X-API-Key: dsk_your_key_here

Method 2: Query parameter

Append the key as a query parameter:

wss://cryptolisting.ws?api_key=dsk_your_key_here

Key properties

Each API key has the following properties, configured by the administrator:

Key lifecycle

1. Created – The administrator generates a key. The full key is shown exactly once.
2. Active – The key can be used to connect. Connections are subject to the key’s limits.
3. Expired – If the key has an expiration date and that date has passed, connections are refused and active sessions are disconnected with close code 1000 and reason key_expired.
4. Revoked – The administrator revokes the key. All active sessions are immediately disconnected with close code 1000 and reason key_revoked. The key can no longer be used.

Security

• Keys are hashed with SHA-256 before storage. Even if the database is compromised, your raw key cannot be recovered.
• Keys are transmitted over TLS-encrypted WebSocket connections (WSS).
• Invalid keys are cached server-side for 30 seconds to prevent brute-force attempts.

WebSocket API

Connection endpoint

wss://cryptolisting.ws

The server uses binary WebSocket frames containing UTF-8 JSON.

Query parameters

Connection lifecycle

Client                              Server
  |                                   |
  |--- WSS handshake + API key -----→ |
  |                                   |-- Validate key
  |                                   |-- Check rate limits
  |                                   |-- Check connection limits
  |←---- 101 Switching Protocols -----|
  |                                   |
  |←---- Welcome message -------------|  (immediate)
  |                                   |
  |←---- Heartbeat -------------------|  (every 30s)
  |←---- Heartbeat -------------------|
  |                                   |
  |←---- Announcement ----------------|  (when detected)
  |                                   |
  |--- {"type":"test"} -------------→ |  (optional)
  |←---- Test Announcement -----------|  (only to you)
  |                                   |
  |←---- PING ----------------------- |  (every 15s)
  |--- PONG ------------------------→ |
  |                                   |
  |←---- Close (1000, key_expired) ---|  (if key expires)
  |                                   |

1. Handshake

The server validates your API key during the HTTP upgrade handshake. If validation fails, you receive an HTTP error response (not a WebSocket frame):

2. Welcome message

Immediately after a successful handshake, the server sends a welcome message confirming your subscription parameters. See Message Reference.

3. Heartbeats

The server sends a JSON heartbeat message every 30 seconds. If you do not receive a heartbeat within ~35 seconds, your connection may be dead.

4. Ping/Pong

In addition to JSON heartbeats, the server sends WebSocket PING frames every 15 seconds. Your WebSocket library handles PONG responses automatically. If the server receives no PONG within 30 seconds, it closes your connection.

5. Announcements

When a listing or other exchange announcement is detected, the server sends an announcement message to all subscribers whose exchange filter matches. See Message Reference.

6. Disconnection

The server may close your connection with a WebSocket close frame. The close code and reason indicate why. See Error Handling.

Client-to-server messages

Client messages are subject to rate limiting (3 messages/minute) and frame size limits (1 KB max). The only supported client message is the test request:

{"type":"test"}

This returns a fake announcement to verify your integration is working. See Message Reference — Test Announcement for details.

Message Reference

All messages are delivered as binary WebSocket frames containing UTF-8 JSON. Every message has a type field.

Welcome

Sent once, immediately after a successful connection.

{
  "type": "welcome",
  "tier": "premium",
  "maxConnections": 5,
  "allowedCex": "*",
  "expiresInSecs": 2592000
}

Announcement

Sent whenever a listing or other exchange announcement is detected on a monitored exchange.

Single ticker:

{
  "type": "announcement",
  "title": "Binance Will List TOKEN (TOKEN)",
  "ticker": "TOKEN",
  "publisher": "binance",
  "listingType": "spot_listing",
  "publishTimestampUs": 1710345000000000,
  "detectedTimestampUs": 1710345000005000,
  "dispatchTimestampUs": 1710345000006000
}

Multiple tickers (one announcement listing several assets at once):

{
  "type": "announcement",
  "title": "Binance Will List ABC, DEF and GHI",
  "ticker": "ABC,DEF,GHI",
  "publisher": "binance",
  "listingType": "spot_listing",
  "publishTimestampUs": 1710345000000000,
  "detectedTimestampUs": 1710345000005000,
  "dispatchTimestampUs": 1710345000006000
}

tickers = data["ticker"].split(",")  # ["ABC", "DEF", "GHI"]

Supported exchanges (publisher)

Listing types

Measuring your latency

All timestamps are in microseconds since UNIX epoch (not milliseconds, not nanoseconds).

Detection delay    = detectedTimestampUs - publishTimestampUs
Dispatch delay     = dispatchTimestampUs - detectedTimestampUs
Your network delay = your_receive_time_us - dispatchTimestampUs
Total end-to-end   = your_receive_time_us - publishTimestampUs

Python example:

import time

def on_announcement(msg):
    now_us = int(time.time() * 1_000_000)
    detection_ms = (msg["detectedTimestampUs"] - msg["publishTimestampUs"]) / 1000
    dispatch_ms  = (msg["dispatchTimestampUs"] - msg["detectedTimestampUs"]) / 1000
    network_ms   = (now_us - msg["dispatchTimestampUs"]) / 1000
    total_ms     = (now_us - msg["publishTimestampUs"]) / 1000

    print(f"Detection: {detection_ms:.2f}ms")
    print(f"Dispatch:  {dispatch_ms:.2f}ms")
    print(f"Network:   {network_ms:.2f}ms")
    print(f"Total:     {total_ms:.2f}ms")

Test Announcement

You can request a fake announcement to verify your integration. Send a JSON message to the server:

{"type":"test"}

The server replies with a test_announcement — identical to a real announcement except for the type field:

{
  "type": "test_announcement",
  "title": "Binance Will List MOCKX (MOCKX)",
  "ticker": "MOCKX",
  "publisher": "binance",
  "listingType": "spot_listing",
  "publishTimestampUs": 1743850000000000,
  "detectedTimestampUs": 1743850001999800,
  "dispatchTimestampUs": 1743850002000000
}

The fields are the same as a regular Announcement. The timestamps are simulated: publishTimestampUs is 1 second before dispatch, and detectedTimestampUs is 200µs before dispatch.

Test rate limit

Test requests are rate-limited to 1 per minute per API key (shared across all connections using the same key). If you exceed the limit, the server responds with an error instead of a test announcement:

{
  "type": "error",
  "code": "test_rate_limited",
  "retryAfterSecs": 42
}

Heartbeat

Sent every 30 seconds to indicate the server is alive.

{
  "type": "heartbeat",
  "timestampNs": 1710345030000000000,
  "timeUtc": "2024-03-13T10:30:30.000000Z"
}

Exchange Filtering

You can choose to receive announcements from specific exchanges only, rather than receiving everything.

How filtering works

Filtering is applied at two levels:

1. Key-level restriction (set by administrator)

When your API key is created, the administrator can restrict it to specific exchanges. For example, a key may be limited to binance,upbit only.

You can check your key’s allowed exchanges in the welcome message you receive upon connection:

{
  "type": "welcome",
  "allowedCex": "binance,upbit",
  ...
}

A value of "*" means your key has access to all exchanges.

2. Client-side preference (set by you)

When connecting, you can further narrow your subscription using the cex query parameter:

wss://cryptolisting.ws?cex=binance

Multiple exchanges:

wss://cryptolisting.ws?cex=binance,upbit

All exchanges (default):

wss://cryptolisting.ws

Effective filter

The effective filter is the intersection of your key’s allowed exchanges and your client-side preference:

Heartbeats are always delivered

Heartbeat messages are always sent regardless of your exchange filter. They are not tied to any specific exchange.

Rate Limits & Security

Connection limits

If you exceed any of these limits, the server responds with HTTP 429 Too Many Requests during the handshake:

{
  "error": "max_connections_reached",
  "limit": 5,
  "current": 5
}

Client message limits

The test request rate limit is shared across all connections using the same API key. Unlike the other limits, exceeding it does not disconnect you — you receive an error message with a retryAfterSecs field instead. See Message Reference — Test Announcement.

TLS

All connections use TLS encryption by default (WSS on port 9201).

API key security

• Keys are hashed before storage – raw keys cannot be recovered from the database.
• Keys can be revoked instantly by an administrator, immediately disconnecting all active sessions.
• Keys can have expiration dates – expired keys are automatically rejected.

Error Handling & Reconnection

HTTP errors (during handshake)

If authentication or rate limiting fails, the server rejects the WebSocket upgrade with an HTTP error:

WebSocket close codes

Once connected, the server may close your connection with a close frame. The reason field tells you why:

Recommended reconnection strategy

Exponential backoff

Attempt 1: wait 1 second
Attempt 2: wait 2 seconds
Attempt 3: wait 4 seconds
Attempt 4: wait 8 seconds
...
Cap: 300 seconds (5 minutes)

Decision logic

On disconnect:
  |
  |-- Close reason = "key_expired" or "key_revoked"?
  |     → Stop. Your key is no longer valid.
  |
  |-- Close reason = "rate_limit_exceeded"?
  |     → Wait 60s, then reconnect. Stop sending messages.
  |
  |-- Close reason = "too_slow"?
  |     → Reconnect immediately. Process messages faster.
  |
  |-- HTTP 429?
  |     → Back off. You're connecting too frequently.
  |
  |-- Unexpected disconnect / network error?
        → Reconnect with exponential backoff.

Heartbeat-based health check

Monitor heartbeats to detect silent disconnections:

import asyncio

HEARTBEAT_TIMEOUT = 35  # seconds

async def monitor_connection(ws):
    while True:
        try:
            msg = await asyncio.wait_for(ws.recv(), timeout=HEARTBEAT_TIMEOUT)
            data = json.loads(msg)
            if data["type"] == "announcement":
                handle_announcement(data)
        except asyncio.TimeoutError:
            print("No heartbeat received -- reconnecting")
            break  # Exit loop and reconnect

Code Examples

Production-ready client implementations with reconnection, heartbeat monitoring, and latency tracking.

import asyncio
import json
import ssl
import time
import websockets

API_KEY = "dsk_your_key_here"
WS_URL  = "wss://cryptolisting.ws"
# Filter specific exchanges (optional):
# WS_URL = "wss://cryptolisting.ws?cex=binance,upbit"

HEARTBEAT_TIMEOUT = 35  # seconds
MAX_RETRIES = 20


def on_announcement(msg: dict):
    now_us = int(time.time() * 1_000_000)
    network_ms = (now_us - msg["dispatchTimestampUs"]) / 1000
    total_ms   = (now_us - msg["publishTimestampUs"]) / 1000

    print(f"[{msg['listingType']}] {msg['ticker']} on {msg['publisher']}")
    print(f"  Title:   {msg['title']}")
    print(f"  Latency: {total_ms:.2f}ms total, {network_ms:.2f}ms network")


async def connect():
    ssl_ctx = ssl.create_default_context()
    ssl_ctx.check_hostname = False
    ssl_ctx.verify_mode = ssl.CERT_NONE

    headers = {"X-API-Key": API_KEY}

    for attempt in range(MAX_RETRIES):
        try:
            async with websockets.connect(
                WS_URL, extra_headers=headers, ssl=ssl_ctx
            ) as ws:
                print("Connected!")

                while True:
                    try:
                        raw = await asyncio.wait_for(
                            ws.recv(), timeout=HEARTBEAT_TIMEOUT
                        )
                    except asyncio.TimeoutError:
                        print("No heartbeat -- reconnecting")
                        break

                    msg = json.loads(raw)

                    if msg["type"] == "welcome":
                        print(f"Welcome: tier={msg['tier']}, "
                              f"cex={msg['allowedCex']}")
                        # Request a test announcement to verify integration
                        await ws.send(json.dumps({"type": "test"}))

                    elif msg["type"] in ("announcement", "test_announcement"):
                        prefix = "[TEST] " if msg["type"] == "test_announcement" else ""
                        print(f"{prefix}", end="")
                        on_announcement(msg)

                    elif msg["type"] == "heartbeat":
                        pass  # Connection alive

        except websockets.ConnectionClosed as e:
            if e.rcvd:
                reason = e.rcvd.reason
                if reason in ("key_expired", "key_revoked"):
                    print(f"Key is no longer valid: {reason}")
                    return
                print(f"Disconnected: {reason}")

        except Exception as e:
            print(f"Connection error: {e}")

        backoff = min(2 ** attempt, 300)
        print(f"Reconnecting in {backoff}s (attempt {attempt + 1})")
        await asyncio.sleep(backoff)

    print("Max retries exceeded")


if __name__ == "__main__":
    asyncio.run(connect())

const WebSocket = require("ws");

const API_KEY = "dsk_your_key_here";
const WS_URL = "wss://cryptolisting.ws";
// Filter specific exchanges (optional):
// const WS_URL = "wss://cryptolisting.ws?cex=binance,upbit";

const HEARTBEAT_TIMEOUT_MS = 35_000;
const MAX_RETRIES = 20;

function onAnnouncement(msg) {
  const nowUs = Date.now() * 1000;
  const networkMs = (nowUs - msg.dispatchTimestampUs) / 1000;
  const totalMs = (nowUs - msg.publishTimestampUs) / 1000;

  console.log(`[${msg.listingType}] ${msg.ticker} on ${msg.publisher}`);
  console.log(`  Title:   ${msg.title}`);
  console.log(`  Latency: ${totalMs.toFixed(2)}ms total, ${networkMs.toFixed(2)}ms network`);
}

function connect(attempt = 0) {
  if (attempt >= MAX_RETRIES) {
    console.log("Max retries exceeded");
    return;
  }

  const ws = new WebSocket(WS_URL, {
    headers: { "X-API-Key": API_KEY },
    rejectUnauthorized: false,
  });

  let heartbeatTimer = null;

  function resetHeartbeat() {
    clearTimeout(heartbeatTimer);
    heartbeatTimer = setTimeout(() => {
      console.log("No heartbeat -- reconnecting");
      ws.terminate();
    }, HEARTBEAT_TIMEOUT_MS);
  }

  ws.on("open", () => {
    console.log("Connected!");
    resetHeartbeat();
  });

  ws.on("message", (data) => {
    resetHeartbeat();
    const msg = JSON.parse(data);

    if (msg.type === "welcome") {
      console.log(`Welcome: tier=${msg.tier}, cex=${msg.allowedCex}`);
      // Request a test announcement to verify integration
      ws.send(JSON.stringify({ type: "test" }));
    } else if (msg.type === "announcement" || msg.type === "test_announcement") {
      const prefix = msg.type === "test_announcement" ? "[TEST] " : "";
      console.log(prefix);
      onAnnouncement(msg);
    }
    // heartbeat: no action needed
  });

  ws.on("close", (code, reason) => {
    clearTimeout(heartbeatTimer);
    const reasonStr = reason.toString();

    if (reasonStr === "key_expired" || reasonStr === "key_revoked") {
      console.log(`Key is no longer valid: ${reasonStr}`);
      return;
    }

    const backoff = Math.min(2 ** attempt, 300);
    console.log(`Disconnected (${code}). Reconnecting in ${backoff}s...`);
    setTimeout(() => connect(attempt + 1), backoff * 1000);
  });

  ws.on("error", (err) => {
    console.error("WebSocket error:", err.message);
  });
}

connect();

package main

import (
	"crypto/tls"
	"encoding/json"
	"fmt"
	"log"
	"math"
	"net/http"
	"time"

	"github.com/gorilla/websocket"
)

const (
	apiKey           = "dsk_your_key_here"
	wsURL            = "wss://cryptolisting.ws"
	// Filter specific exchanges (optional):
	// wsURL = "wss://cryptolisting.ws?cex=binance,upbit"
	heartbeatTimeout = 35 * time.Second
	maxRetries       = 20
)

type Message struct {
	Type                string `json:"type"`
	Title               string `json:"title,omitempty"`
	Ticker              string `json:"ticker,omitempty"`
	Publisher           string `json:"publisher,omitempty"`
	ListingType         string `json:"listingType,omitempty"`
	PublishTimestampUs  uint64 `json:"publishTimestampUs,omitempty"`
	DetectedTimestampUs uint64 `json:"detectedTimestampUs,omitempty"`
	DispatchTimestampUs uint64 `json:"dispatchTimestampUs,omitempty"`
	Tier                string `json:"tier,omitempty"`
	MaxConnections      int    `json:"maxConnections,omitempty"`
	AllowedCex          string `json:"allowedCex,omitempty"`
	TimestampNs         uint64 `json:"timestampNs,omitempty"`
	TimeUtc             string `json:"timeUtc,omitempty"`
}

func onAnnouncement(msg Message) {
	nowUs := uint64(time.Now().UnixMicro())
	networkMs := float64(nowUs-msg.DispatchTimestampUs) / 1000
	totalMs := float64(nowUs-msg.PublishTimestampUs) / 1000

	fmt.Printf("[%s] %s on %s\n", msg.ListingType, msg.Ticker, msg.Publisher)
	fmt.Printf("  Title:   %s\n", msg.Title)
	fmt.Printf("  Latency: %.2fms total, %.2fms network\n", totalMs, networkMs)
}

func connect() {
	dialer := websocket.Dialer{
		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
	}
	headers := http.Header{"X-API-Key": {apiKey}}

	for attempt := 0; attempt < maxRetries; attempt++ {
		conn, resp, err := dialer.Dial(wsURL, headers)
		if err != nil {
			if resp != nil {
				log.Printf("Connection failed (HTTP %d): %v", resp.StatusCode, err)
				if resp.StatusCode == 403 {
					log.Println("API key is invalid, expired, or revoked. Stopping.")
					return
				}
			} else {
				log.Printf("Connection failed: %v", err)
			}
			backoff := math.Min(math.Pow(2, float64(attempt)), 300)
			log.Printf("Reconnecting in %.0fs (attempt %d)", backoff, attempt+1)
			time.Sleep(time.Duration(backoff) * time.Second)
			continue
		}

		log.Println("Connected!")
		attempt = 0 // reset backoff on successful connection
		conn.SetReadDeadline(time.Now().Add(heartbeatTimeout))

		shouldStop := false
		for {
			_, rawMsg, err := conn.ReadMessage()
			if err != nil {
				if closeErr, ok := err.(*websocket.CloseError); ok {
					reason := closeErr.Text
					if reason == "key_expired" || reason == "key_revoked" {
						log.Printf("Key is no longer valid: %s. Stopping.", reason)
						shouldStop = true
					} else {
						log.Printf("Disconnected: %s (code %d)", reason, closeErr.Code)
					}
				} else {
					log.Printf("Read error: %v", err)
				}
				break
			}

			conn.SetReadDeadline(time.Now().Add(heartbeatTimeout))

			var msg Message
			json.Unmarshal(rawMsg, &msg)

			switch msg.Type {
			case "welcome":
				log.Printf("Welcome: tier=%s, cex=%s", msg.Tier, msg.AllowedCex)
				// Request a test announcement to verify integration
				conn.WriteMessage(websocket.TextMessage, []byte(`{"type":"test"}`))
			case "announcement", "test_announcement":
				if msg.Type == "test_announcement" {
					fmt.Print("[TEST] ")
				}
				onAnnouncement(msg)
			case "heartbeat":
				// Connection alive
			}
		}
		conn.Close()

		if shouldStop {
			return
		}

		backoff := math.Min(math.Pow(2, float64(attempt)), 300)
		log.Printf("Reconnecting in %.0fs (attempt %d)", backoff, attempt+1)
		time.Sleep(time.Duration(backoff) * time.Second)
	}

	log.Println("Max retries exceeded")
}

func main() {
	connect()
}

[dependencies]
tokio = { version = "1", features = ["full"] }
tokio-tungstenite = { version = "0.21", features = ["native-tls"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
native-tls = "0.2"

use futures_util::StreamExt;
use serde::Deserialize;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio_tungstenite::{connect_async_tls_with_config, Connector};
use tokio_tungstenite::tungstenite::protocol::CloseFrame;
use tokio_tungstenite::tungstenite::Message as WsMessage;

const API_KEY: &str = "dsk_your_key_here";
const WS_URL: &str = "wss://cryptolisting.ws";
// Filter specific exchanges (optional):
// const WS_URL: &str = "wss://cryptolisting.ws?cex=binance,upbit";

const HEARTBEAT_TIMEOUT: Duration = Duration::from_secs(35);
const MAX_RETRIES: u32 = 20;

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct Msg {
    r#type: String,
    #[serde(default)] title: String,
    #[serde(default)] ticker: String,
    #[serde(default)] publisher: String,
    #[serde(default)] listing_type: String,
    #[serde(default)] publish_timestamp_us: u64,
    #[serde(default)] dispatch_timestamp_us: u64,
    #[serde(default)] tier: String,
    #[serde(default)] allowed_cex: String,
}

fn on_announcement(msg: &Msg) {
    let now_us = SystemTime::now()
        .duration_since(UNIX_EPOCH).unwrap()
        .as_micros() as u64;
    let network_ms = (now_us - msg.dispatch_timestamp_us) as f64 / 1000.0;
    let total_ms = (now_us - msg.publish_timestamp_us) as f64 / 1000.0;

    println!("[{}] {} on {}", msg.listing_type, msg.ticker, msg.publisher);
    println!("  Title:   {}", msg.title);
    println!("  Latency: {:.2}ms total, {:.2}ms network", total_ms, network_ms);
}

#[tokio::main]
async fn main() {
    let tls = native_tls::TlsConnector::builder()
        .danger_accept_invalid_certs(true)
        .build()
        .unwrap();

    for attempt in 0..MAX_RETRIES {
        let request = http::Request::builder()
            .uri(WS_URL)
            .header("X-API-Key", API_KEY)
            .header("Connection", "Upgrade")
            .header("Upgrade", "websocket")
            .header("Sec-WebSocket-Version", "13")
            .header("Sec-WebSocket-Key",
                tokio_tungstenite::tungstenite::handshake::client::generate_key())
            .body(())
            .unwrap();

        let connector = Connector::NativeTls(tls.clone());
        let ws = match connect_async_tls_with_config(
            request, None, false, Some(connector)
        ).await {
            Ok((ws, _)) => ws,
            Err(e) => {
                eprintln!("Connection failed: {}", e);
                let backoff = Duration::from_secs((2u64.pow(attempt)).min(300));
                eprintln!("Reconnecting in {}s (attempt {})", backoff.as_secs(), attempt + 1);
                tokio::time::sleep(backoff).await;
                continue;
            }
        };

        println!("Connected!");
        let (_, mut read) = ws.split();
        let mut should_stop = false;

        loop {
            match tokio::time::timeout(HEARTBEAT_TIMEOUT, read.next()).await {
                Err(_) => {
                    eprintln!("No heartbeat -- reconnecting");
                    break;
                }
                Ok(None) => {
                    eprintln!("Connection closed");
                    break;
                }
                Ok(Some(Err(e))) => {
                    eprintln!("Read error: {}", e);
                    break;
                }
                Ok(Some(Ok(WsMessage::Close(Some(CloseFrame { reason, .. }))))) => {
                    if reason == "key_expired" || reason == "key_revoked" {
                        eprintln!("Key is no longer valid: {}. Stopping.", reason);
                        should_stop = true;
                    } else {
                        eprintln!("Disconnected: {}", reason);
                    }
                    break;
                }
                Ok(Some(Ok(frame))) => {
                    if let Ok(text) = frame.to_text() {
                        if let Ok(msg) = serde_json::from_str::<Msg>(text) {
                            match msg.r#type.as_str() {
                                "welcome" => {
                                    println!("Welcome: tier={}, cex={}", msg.tier, msg.allowed_cex);
                                    // Request a test announcement to verify integration
                                    // (requires access to write half)
                                }
                                "announcement" | "test_announcement" => {
                                    if msg.r#type == "test_announcement" {
                                        print!("[TEST] ");
                                    }
                                    on_announcement(&msg);
                                }
                                "heartbeat" => {} // Connection alive
                                _ => {}
                            }
                        }
                    }
                }
            }
        }

        if should_stop {
            return;
        }

        let backoff = Duration::from_secs((2u64.pow(attempt)).min(300));
        eprintln!("Reconnecting in {}s (attempt {})", backoff.as_secs(), attempt + 1);
        tokio::time::sleep(backoff).await;
    }

    eprintln!("Max retries exceeded");
}