zfin/src/net/LiveStream.zig

//! Live-price websocket stream (push-based intraday quotes).
//!
//! A `LiveStream` holds a single persistent websocket on a background
//! thread, parses incoming price ticks, and exposes the latest price
//! per ticker via `snapshotInto`. The TUI/CLI poll the snapshot; they
//! never block on the socket.
//!
//! ## Transport-ready
//!
//! The stream is parameterized by a `Transport` enum so a second
//! provider slots in without reshaping the scaffolding. Only `.yahoo`
//! is implemented today:
//!
//! - **Yahoo** (`streamer.finance.yahoo.com`): keyless, free, works on
//!   every tier. Subscribe with `{"subscribe":[...]}`; each frame is
//!   `{"type":"pricing","message":"<base64 protobuf>"}` whose protobuf
//!   carries field 1 = ticker (string) and field 2 = price (fixed32
//!   float). It is an UNOFFICIAL feed (the browser uses it) and can
//!   break without notice - hence "could die any time", which is why a
//!   future Tiingo transport is planned as the paid, official backup.
//!
//! The per-transport surface is exactly three things: the connect
//! target + handshake headers (`transportConfig`), the subscribe
//! message (`sendSubscribe`), and the frame parser (`serverMessage`).
//! Everything else - thread, lock, snapshot, reconnect/backoff, client
//! lifecycle - is shared.
//!
//! ## Why a raw background thread (not std.Io tasks)
//!
//! The read loop is a single long-lived blocking loop (`readLoop`),
//! so it runs on its own OS thread that coexists with the app's
//! `Io.Threaded` event loop. The pinned `websocket.zig` client is
//! itself Io-aware (connect, TLS, reads, writes go through `std.Io`,
//! which is safe to call from this thread); we thread `io` in for it.
//!
//! ## Threading and shutdown
//!
//! `prices` and `client` are guarded by `mutex` (a `std.atomic.Mutex`
//! spinlock, since the read loop runs outside the Io task model and
//! critical sections are single hashmap ops held for microseconds).
//! Shutdown: `stop()` sets `should_stop` and calls `client.close()`
//! from the caller's thread, which unblocks the read thread's
//! `readLoop`, after which we `join`. The blocking connect (DNS + TCP
//! + TLS) runs WITHOUT holding `mutex`, so a quit during a slow connect
//! isn't blocked.

const std = @import("std");
const websocket = @import("websocket");
const json_utils = @import("../providers/json_utils.zig");
const jsonStr = json_utils.jsonStr;

const log = std.log.scoped(.live_stream);

const LiveStream = @This();

const tls_port: u16 = 443;

const yahoo_host = "streamer.finance.yahoo.com";
const yahoo_path = "/?version=2";
const yahoo_origin = "https://finance.yahoo.com";

/// Tickers are short; 24 bytes covers stocks, class suffixes, index
/// (`^GSPC`), crypto (`BTC-USD`), and futures (`ES=F`) shapes.
const max_ticker_len = 24;
const handshake_timeout_ms = 10_000;
const max_message_size = 512 * 1024;
const read_buffer_size = 32 * 1024;
const reconnect_backoff_ms = 2_000;
/// Yahoo pricing protobufs are ~60-90 bytes decoded; 1 KiB is ample.
/// A frame whose decoded size exceeds this is skipped (returns null).
const max_decoded_message = 1024;

/// Which live-quote feed the stream connects to. Only `.yahoo` is
/// wired today; the enum exists so a paid Tiingo transport is a
/// fill-in (add the arm to `transportConfig`/`sendSubscribe`/the
/// `serverMessage` parser) rather than a refactor.
pub const Transport = enum { yahoo };

/// A single parsed price update. The ticker is stored inline so the
/// value owns no heap memory.
pub const StreamTick = struct {
    ticker_buf: [max_ticker_len]u8 = @splat(0),
    ticker_len: usize = 0,
    price: f64 = 0,

    pub fn ticker(self: *const StreamTick) []const u8 {
        return self.ticker_buf[0..self.ticker_len];
    }
};

const TransportConfig = struct {
    host: []const u8,
    port: u16,
    path: []const u8,
    /// Raw handshake header block (CRLF-separated, no trailing CRLF).
    /// The library does not auto-add Host, so it is included here.
    headers: []const u8,
};

// ── Fields ────────────────────────────────────────────────────

allocator: std.mem.Allocator,
/// Threaded into the websocket client (the pinned `websocket.zig` is
/// Io-aware: connect, TLS, reads, and writes all go through it).
/// `std.Io` is safe to use from the read thread.
io: std.Io,
transport: Transport,
/// Owned copy of the API token, or null for keyless transports
/// (Yahoo). Decoupled from Config's lifetime.
api_key: ?[]u8 = null,
/// Owned, duped tickers to subscribe. Set in `start`, freed in `stop`.
tickers: [][]u8 = &.{},
/// Guards `prices` and `client`. Spinlock (not an Io mutex) because the
/// read loop is a raw OS thread outside the Io task model; critical
/// sections are one hashmap op or a short pointer read.
mutex: std.atomic.Mutex = .unlocked,
/// Latest price per ticker, keyed by UPPERCASED ticker (owned dup).
/// Written by the read thread, read via `snapshotInto`.
prices: std.StringHashMap(f64),
/// The live client, published so `stop()` can close it cross-thread.
/// Heap-owned by the read thread for the duration of one connection.
client: ?*websocket.Client = null,
thread: ?std.Thread = null,
should_stop: std.atomic.Value(bool) = .init(false),

// ── Lifecycle ─────────────────────────────────────────────────

/// Allocate and initialize a stream. The returned pointer is stable
/// (the background thread captures it), so callers must keep it
/// heap-allocated and call `destroy` (not move it). `api_key` is copied
/// when non-null; pass null for keyless transports.
pub fn create(allocator: std.mem.Allocator, io: std.Io, transport: Transport, api_key: ?[]const u8) !*LiveStream {
    const self = try allocator.create(LiveStream);
    errdefer allocator.destroy(self);
    const key_dup: ?[]u8 = if (api_key) |k| try allocator.dupe(u8, k) else null;
    self.* = .{
        .allocator = allocator,
        .io = io,
        .transport = transport,
        .api_key = key_dup,
        .prices = std.StringHashMap(f64).init(allocator),
    };
    return self;
}

/// Stop the stream (if running) and free everything.
pub fn destroy(self: *LiveStream) void {
    self.stop();
    var it = self.prices.keyIterator();
    while (it.next()) |k| self.allocator.free(k.*);
    self.prices.deinit();
    if (self.api_key) |k| self.allocator.free(k);
    self.allocator.destroy(self);
}

/// Start streaming `symbols`. Idempotent: a no-op if already running
/// (the symbol set is fixed for the stream's lifetime; change it with
/// `stop` + `start`). `symbols` is copied.
pub fn start(self: *LiveStream, symbols: []const []const u8) !void {
    if (self.thread != null) return;

    const tk = try self.allocator.alloc([]u8, symbols.len);
    errdefer self.allocator.free(tk);
    var n: usize = 0;
    errdefer for (tk[0..n]) |t| self.allocator.free(t);
    for (symbols) |s| {
        tk[n] = try self.allocator.dupe(u8, s);
        n += 1;
    }
    self.tickers = tk;

    self.should_stop.store(false, .release);
    self.thread = try std.Thread.spawn(.{}, threadMain, .{self});
}

/// Signal the read thread to stop, close the connection to unblock it,
/// join, and free the subscribed-ticker list. Safe to call when not
/// running.
pub fn stop(self: *LiveStream) void {
    self.should_stop.store(true, .release);
    // Close the socket so the read thread's blocking readLoop returns.
    // close() is thread-safe per the library.
    self.lock();
    if (self.client) |c| c.close(.{}) catch |err| log.debug("close on stop: {t}", .{err});
    self.unlock();

    if (self.thread) |t| {
        t.join();
        self.thread = null;
    }

    for (self.tickers) |t| self.allocator.free(t);
    self.allocator.free(self.tickers);
    self.tickers = &.{};
}

pub fn isRunning(self: *LiveStream) bool {
    return self.thread != null;
}

/// Spin-acquire `mutex`. Critical sections are O(1) hashmap ops (or an
/// O(symbols) snapshot scan), so the spin is effectively uncontended.
fn lock(self: *LiveStream) void {
    while (!self.mutex.tryLock()) std.atomic.spinLoopHint();
}

fn unlock(self: *LiveStream) void {
    self.mutex.unlock();
}

/// Copy the latest streamed prices for `symbols` into `out`, keyed by
/// the caller's original slices (so `out`'s keys borrow `symbols`).
/// Matching is case-insensitive (we store UPPERCASED tickers). Symbols
/// with no streamed price yet are left absent (caller falls back to the
/// cached close).
pub fn snapshotInto(self: *LiveStream, symbols: []const []const u8, out: *std.StringHashMap(f64)) void {
    self.lock();
    defer self.unlock();
    var upper: [max_ticker_len]u8 = undefined;
    for (symbols) |s| {
        if (s.len > upper.len) continue;
        const key = std.ascii.upperString(upper[0..], s);
        if (self.prices.get(key)) |p| {
            out.put(s, p) catch |err| log.warn("snapshot put({s}): {t}", .{ s, err });
        }
    }
}

/// Read-thread handler hook (called by `websocket.Client.readLoop`).
/// Parses one frame and updates `prices`. Never propagates an error (a
/// bad frame must not tear down the stream); failures are swallowed so
/// the connection stays up.
pub fn serverMessage(self: *LiveStream, data: []const u8) !void {
    const tick = switch (self.transport) {
        .yahoo => parseYahooMessage(self.allocator, data),
    } orelse return;
    var upper: [max_ticker_len]u8 = undefined;
    if (tick.ticker().len > upper.len) return;
    const key = std.ascii.upperString(upper[0..], tick.ticker());

    self.lock();
    defer self.unlock();
    const gop = self.prices.getOrPut(key) catch return;
    if (!gop.found_existing) {
        gop.key_ptr.* = self.allocator.dupe(u8, key) catch {
            // Roll back the entry whose key is still the stack slice.
            _ = self.prices.remove(key);
            return;
        };
    }
    gop.value_ptr.* = tick.price;
}

// ── Read-thread internals ─────────────────────────────────────

fn threadMain(self: *LiveStream) void {
    while (!self.should_stop.load(.acquire)) {
        self.connectAndRun() catch |err| log.debug("stream connection ended: {t}", .{err});
        if (self.should_stop.load(.acquire)) break;
        self.backoffSleep(reconnect_backoff_ms);
    }
}

fn transportConfig(self: *LiveStream) TransportConfig {
    return switch (self.transport) {
        .yahoo => .{
            .host = yahoo_host,
            .port = tls_port,
            .path = yahoo_path,
            .headers = "Host: " ++ yahoo_host ++ "\r\nOrigin: " ++ yahoo_origin,
        },
    };
}

fn connectAndRun(self: *LiveStream) !void {
    const cfg = self.transportConfig();

    // Blocking connect (DNS + TCP + TLS) happens BEFORE publishing the
    // client, so a concurrent stop() isn't blocked waiting on it.
    const cptr = try self.allocator.create(websocket.Client);
    cptr.* = websocket.Client.init(self.io, self.allocator, .{
        .host = cfg.host,
        .port = cfg.port,
        .tls = true,
        .max_size = max_message_size,
        .buffer_size = read_buffer_size,
    }) catch |err| {
        self.allocator.destroy(cptr);
        return err;
    };
    // From here cptr owns a live connection; always tear down + unpublish.
    defer {
        self.lock();
        self.client = null;
        self.unlock();
        cptr.deinit();
        self.allocator.destroy(cptr);
    }

    // Publish so stop() can close the socket to unblock readLoop.
    self.lock();
    self.client = cptr;
    self.unlock();

    try cptr.handshake(cfg.path, .{
        .timeout_ms = handshake_timeout_ms,
        .headers = cfg.headers,
    });
    try self.sendSubscribe(cptr);
    // Blocks until the connection closes (or stop() closes it). The
    // library auto-handles ping/pong and close frames; our
    // serverMessage handles the text data frames.
    try cptr.readLoop(self);
}

fn sendSubscribe(self: *LiveStream, c: *websocket.Client) !void {
    var buf: std.ArrayList(u8) = .empty;
    defer buf.deinit(self.allocator);
    try buildSubscribe(self.transport, self.allocator, self.tickers, &buf);
    // write() masks the buffer in place, so it must be a mutable buffer
    // we own; we discard it right after.
    try c.write(buf.items);
}

/// Build the transport's subscribe message into `buf`. Pure (no I/O),
/// so it's unit-testable without a live connection.
fn buildSubscribe(transport: Transport, allocator: std.mem.Allocator, tickers: []const []const u8, buf: *std.ArrayList(u8)) !void {
    switch (transport) {
        .yahoo => {
            // {"subscribe":["SPY","AAPL"]}
            // Tickers are alphanumeric plus '.'/'-'/'^'/'='/'=X', so no
            // JSON string escaping is required.
            try buf.appendSlice(allocator, "{\"subscribe\":[");
            for (tickers, 0..) |t, i| {
                if (i != 0) try buf.append(allocator, ',');
                try buf.append(allocator, '"');
                try buf.appendSlice(allocator, t);
                try buf.append(allocator, '"');
            }
            try buf.appendSlice(allocator, "]}");
        },
    }
}

/// Sleep `ms` in small chunks, bailing early if asked to stop, so a
/// quit during reconnect backoff doesn't wait the full interval. Uses
/// the `.awake` (monotonic) clock so a wall-clock jump can't skip or
/// extend the backoff.
fn backoffSleep(self: *LiveStream, ms: u64) void {
    var remaining = ms;
    while (remaining > 0 and !self.should_stop.load(.acquire)) {
        const chunk = @min(remaining, 200);
        self.io.sleep(.fromMilliseconds(@intCast(chunk)), .awake) catch return;
        remaining -= chunk;
    }
}

// ── Yahoo frame parsing ───────────────────────────────────────

/// Parse one Yahoo streamer frame. Returns a tick only for
/// `{"type":"pricing","message":"<base64 protobuf>"}` frames whose
/// protobuf carries field 1 (ticker) and field 2 (price). Any other
/// type, or a malformed/oversized frame, yields null. Pure and
/// allocator-scoped (parse memory freed before return), so fully
/// fixture-testable.
fn parseYahooMessage(allocator: std.mem.Allocator, body: []const u8) ?StreamTick {
    const parsed = std.json.parseFromSlice(std.json.Value, allocator, body, .{}) catch return null;
    defer parsed.deinit();

    const obj = switch (parsed.value) {
        .object => |o| o,
        else => return null,
    };

    // Only "pricing" frames carry a price payload.
    const typ = jsonStr(obj.get("type")) orelse return null;
    if (!std.mem.eql(u8, typ, "pricing")) return null;
    const b64 = jsonStr(obj.get("message")) orelse return null;

    // Base64-decode the protobuf into a stack buffer. Standard alphabet
    // with padding (Yahoo emits '+', '/', '=').
    const dec = std.base64.standard.Decoder;
    const decoded_len = dec.calcSizeForSlice(b64) catch return null;
    if (decoded_len > max_decoded_message) return null;
    var buf: [max_decoded_message]u8 = undefined;
    dec.decode(buf[0..decoded_len], b64) catch return null;

    return parsePricingProto(buf[0..decoded_len]);
}

/// Walk a Yahoo PricingData protobuf for field 1 (ticker, length-
/// delimited string) and field 2 (price, fixed32 little-endian float),
/// skipping every other field. Returns null if either is absent or the
/// buffer is malformed.
fn parsePricingProto(pb: []const u8) ?StreamTick {
    var i: usize = 0;
    var ticker: ?[]const u8 = null;
    var price: ?f64 = null;

    while (i < pb.len) {
        const tag = readVarint(pb, &i) orelse break;
        const field = tag >> 3;
        const wire: u3 = @intCast(tag & 0x7);
        switch (wire) {
            0 => _ = readVarint(pb, &i) orelse break, // varint
            1 => { // fixed64
                if (i + 8 > pb.len) break;
                i += 8;
            },
            2 => { // length-delimited
                const len = readVarint(pb, &i) orelse break;
                if (i + len > pb.len) break;
                const bytes = pb[i .. i + len];
                i += len;
                if (field == 1) ticker = bytes;
            },
            5 => { // fixed32
                if (i + 4 > pb.len) break;
                const raw = std.mem.readInt(u32, pb[i..][0..4], .little);
                i += 4;
                if (field == 2) price = @as(f64, @as(f32, @bitCast(raw)));
            },
            else => break, // groups (3/4) are deprecated/unsupported
        }
        if (ticker != null and price != null) break;
    }

    const t = ticker orelse return null;
    const p = price orelse return null;
    var tick: StreamTick = .{ .price = p };
    const n = @min(t.len, tick.ticker_buf.len);
    @memcpy(tick.ticker_buf[0..n], t[0..n]);
    tick.ticker_len = n;
    return tick;
}

/// Decode a base-128 protobuf varint starting at `i.*`, advancing `i`
/// past it. Returns null on truncation or an over-long (> 64-bit)
/// encoding.
fn readVarint(buf: []const u8, i: *usize) ?u64 {
    var result: u64 = 0;
    var shift: usize = 0;
    while (i.* < buf.len) {
        const byte = buf[i.*];
        i.* += 1;
        if (shift >= 64) return null; // malformed: more than 10 bytes
        result |= @as(u64, byte & 0x7f) << @intCast(shift);
        if (byte & 0x80 == 0) return result;
        shift += 7;
    }
    return null; // truncated
}

// ── Tests ─────────────────────────────────────────────────────

test "parseYahooMessage: AAPL pricing frame yields ticker + price" {
    const body =
        \\{"type":"pricing","message":"CgRBQVBMFUgBjUMY4JKAz+JnKgNOTVMwCDgBRQisH79InK3UKmUAj+K/sAFQ2AEE"}
    ;
    const tick = parseYahooMessage(std.testing.allocator, body) orelse return error.TestExpectedTick;
    try std.testing.expectEqualStrings("AAPL", tick.ticker());
    // field 2 is a fixed32 float: exactly 282.010009765625.
    try std.testing.expectApproxEqAbs(@as(f64, 282.010009765625), tick.price, 0.0001);
}

test "parseYahooMessage: SPY pricing frame yields ticker + price" {
    const body =
        \\{"type":"pricing","message":"CgNTUFkVUhg5RBiwooDP4mcqA1BDWDAUOAFF9v3HP0jQ+bonZYA9NkGwAVDYAQQ="}
    ;
    const tick = parseYahooMessage(std.testing.allocator, body) orelse return error.TestExpectedTick;
    try std.testing.expectEqualStrings("SPY", tick.ticker());
    try std.testing.expectApproxEqAbs(@as(f64, 740.3800048828125), tick.price, 0.0001);
}

test "parseYahooMessage: non-pricing and malformed frames yield null" {
    const a = std.testing.allocator;
    // Not JSON.
    try std.testing.expect(parseYahooMessage(a, "not json") == null);
    // JSON array root, not object.
    try std.testing.expect(parseYahooMessage(a, "[1,2,3]") == null);
    // Pricing-shaped but wrong type.
    try std.testing.expect(parseYahooMessage(a,
        \\{"type":"heartbeat","message":"CgRBQVBM"}
    ) == null);
    // Missing message field.
    try std.testing.expect(parseYahooMessage(a,
        \\{"type":"pricing"}
    ) == null);
    // message present but not valid base64.
    try std.testing.expect(parseYahooMessage(a,
        \\{"type":"pricing","message":"!!!not-base64!!!"}
    ) == null);
}

test "parsePricingProto: missing field 1 or field 2 yields null" {
    // Only field 1 (ticker "AB"), no price field.
    const ticker_only = [_]u8{ 0x0a, 0x02, 'A', 'B' };
    try std.testing.expect(parsePricingProto(&ticker_only) == null);
    // Only field 2 (price), no ticker.
    var price_only: [5]u8 = undefined;
    price_only[0] = 0x15; // field 2, wire type 5 (fixed32)
    std.mem.writeInt(u32, price_only[1..5], @bitCast(@as(f32, 12.5)), .little);
    try std.testing.expect(parsePricingProto(&price_only) == null);
}

test "parsePricingProto: skips intervening fields to reach price" {
    // field 1 = "X", field 3 varint (skipped), field 2 = 9.5 fixed32.
    var pb: std.ArrayList(u8) = .empty;
    defer pb.deinit(std.testing.allocator);
    const a = std.testing.allocator;
    try pb.appendSlice(a, &.{ 0x0a, 0x01, 'X' }); // field 1, len 1, "X"
    try pb.appendSlice(a, &.{ 0x18, 0x96, 0x01 }); // field 3 varint = 150
    try pb.append(a, 0x15); // field 2, wire 5
    var price_bytes: [4]u8 = undefined;
    std.mem.writeInt(u32, &price_bytes, @bitCast(@as(f32, 9.5)), .little);
    try pb.appendSlice(a, &price_bytes);

    const tick = parsePricingProto(pb.items) orelse return error.TestExpectedTick;
    try std.testing.expectEqualStrings("X", tick.ticker());
    try std.testing.expectApproxEqAbs(@as(f64, 9.5), tick.price, 0.0001);
}

test "parsePricingProto: skips a fixed64 field before reaching the price" {
    var pb: std.ArrayList(u8) = .empty;
    defer pb.deinit(std.testing.allocator);
    const a = std.testing.allocator;
    try pb.appendSlice(a, &.{ 0x0a, 0x01, 'Y' }); // field 1, "Y"
    try pb.append(a, 0x21); // field 4, wire type 1 (fixed64)
    try pb.appendSlice(a, &.{ 0, 0, 0, 0, 0, 0, 0, 0 }); // 8 bytes, skipped
    try pb.append(a, 0x15); // field 2, wire type 5 (fixed32)
    var price_bytes: [4]u8 = undefined;
    std.mem.writeInt(u32, &price_bytes, @bitCast(@as(f32, 3.25)), .little);
    try pb.appendSlice(a, &price_bytes);

    const tick = parsePricingProto(pb.items) orelse return error.TestExpectedTick;
    try std.testing.expectEqualStrings("Y", tick.ticker());
    try std.testing.expectApproxEqAbs(@as(f64, 3.25), tick.price, 0.0001);
}

test "parsePricingProto: unsupported wire type (group) stops the walk" {
    // field 1 = "Z", then a group-start tag (wire type 3) -> break.
    // Price is never seen, so the result is null.
    const pb = [_]u8{ 0x0a, 0x01, 'Z', 0x0b }; // 0x0b = field 1, wire type 3
    try std.testing.expect(parsePricingProto(&pb) == null);
}

test "parsePricingProto: over-long ticker is truncated to capacity" {
    var pb: std.ArrayList(u8) = .empty;
    defer pb.deinit(std.testing.allocator);
    const a = std.testing.allocator;
    const long = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; // 36 chars
    try pb.append(a, 0x0a); // field 1, wire 2
    try pb.append(a, @intCast(long.len));
    try pb.appendSlice(a, long);
    try pb.append(a, 0x15); // field 2, wire 5
    var price_bytes: [4]u8 = undefined;
    std.mem.writeInt(u32, &price_bytes, @bitCast(@as(f32, 1.0)), .little);
    try pb.appendSlice(a, &price_bytes);

    const tick = parsePricingProto(pb.items) orelse return error.TestExpectedTick;
    try std.testing.expectEqual(@as(usize, max_ticker_len), tick.ticker().len);
    try std.testing.expectEqualStrings("ABCDEFGHIJKLMNOPQRSTUVWX", tick.ticker());
}

test "readVarint: multi-byte and truncated encodings" {
    var i: usize = 0;
    // 150 = 0x96 0x01
    const ok = [_]u8{ 0x96, 0x01 };
    try std.testing.expectEqual(@as(?u64, 150), readVarint(&ok, &i));
    try std.testing.expectEqual(@as(usize, 2), i);
    // Truncated (continuation bit set, no follow-on byte).
    i = 0;
    const trunc = [_]u8{0x80};
    try std.testing.expect(readVarint(&trunc, &i) == null);
}

test "LiveStream: create/destroy lifecycle, keyless" {
    const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null);
    try std.testing.expect(!s.isRunning());
    try std.testing.expect(s.api_key == null);
    s.destroy(); // testing.allocator would flag any leak
}

test "LiveStream: create dupes and frees an api key" {
    // Yahoo ignores the key, but the dupe/free branches must be exercised.
    const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, "secret-token");
    try std.testing.expect(s.api_key != null);
    try std.testing.expectEqualStrings("secret-token", s.api_key.?);
    s.destroy();
}

test "LiveStream: serverMessage stores prices; snapshotInto matches case-insensitively" {
    const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null);
    defer s.destroy();

    try s.serverMessage(
        \\{"type":"pricing","message":"CgRBQVBMFUgBjUMY4JKAz+JnKgNOTVMwCDgBRQisH79InK3UKmUAj+K/sAFQ2AEE"}
    );
    try s.serverMessage(
        \\{"type":"pricing","message":"CgNTUFkVUhg5RBiwooDP4mcqA1BDWDAUOAFF9v3HP0jQ+bonZYA9NkGwAVDYAQQ="}
    );
    // A non-pricing frame is ignored (early return, no entry, no crash).
    try s.serverMessage("not a frame");

    var out = std.StringHashMap(f64).init(std.testing.allocator);
    defer out.deinit();
    // lowercase request hits the UPPERCASED stored key; an unseen symbol
    // is absent; an over-long symbol is skipped without overflowing.
    const syms = [_][]const u8{ "aapl", "spy", "TSLA", "THIS_TICKER_IS_TOO_LONG_TO_FIT_IN_BUF" };
    s.snapshotInto(&syms, &out);
    try std.testing.expectApproxEqAbs(@as(f64, 282.010009765625), out.get("aapl").?, 0.0001);
    try std.testing.expectApproxEqAbs(@as(f64, 740.3800048828125), out.get("spy").?, 0.0001);
    try std.testing.expect(out.get("TSLA") == null);
}

test "LiveStream: serverMessage overwrites an existing ticker (found_existing path)" {
    const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null);
    defer s.destroy();
    const frame =
        \\{"type":"pricing","message":"CgRBQVBMFUgBjUMY4JKAz+JnKgNOTVMwCDgBRQisH79InK3UKmUAj+K/sAFQ2AEE"}
    ;
    try s.serverMessage(frame);
    try s.serverMessage(frame); // second time reuses the existing key
    var out = std.StringHashMap(f64).init(std.testing.allocator);
    defer out.deinit();
    const syms = [_][]const u8{"AAPL"};
    s.snapshotInto(&syms, &out);
    try std.testing.expectEqual(@as(usize, 1), out.count());
}

test "LiveStream: yahoo transportConfig carries the Origin header and version path" {
    const s = try LiveStream.create(std.testing.allocator, std.testing.io, .yahoo, null);
    defer s.destroy();
    const cfg = s.transportConfig();
    try std.testing.expectEqualStrings(yahoo_host, cfg.host);
    try std.testing.expectEqualStrings(yahoo_path, cfg.path);
    try std.testing.expectEqual(@as(u16, 443), cfg.port);
    try std.testing.expect(std.mem.indexOf(u8, cfg.headers, "Origin: https://finance.yahoo.com") != null);
    try std.testing.expect(std.mem.indexOf(u8, cfg.headers, "Host: " ++ yahoo_host) != null);
}

test "buildSubscribe: yahoo message for one and many tickers" {
    const a = std.testing.allocator;
    var buf: std.ArrayList(u8) = .empty;
    defer buf.deinit(a);

    try buildSubscribe(.yahoo, a, &.{"SPY"}, &buf);
    try std.testing.expectEqualStrings("{\"subscribe\":[\"SPY\"]}", buf.items);

    buf.clearRetainingCapacity();
    try buildSubscribe(.yahoo, a, &.{ "SPY", "AAPL", "BTC-USD" }, &buf);
    try std.testing.expectEqualStrings("{\"subscribe\":[\"SPY\",\"AAPL\",\"BTC-USD\"]}", buf.items);

    // No tickers -> empty array (degenerate but well-formed).
    buf.clearRetainingCapacity();
    try buildSubscribe(.yahoo, a, &.{}, &buf);
    try std.testing.expectEqualStrings("{\"subscribe\":[]}", buf.items);
}