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.
//!
//! ## Dual transport
//!
//! The stream is parameterized by a `Transport` enum; both arms share
//! all the scaffolding:
//!
//! - **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 the official Tiingo backup.
//! - **Tiingo** (`api.tiingo.com/iex`): official IEX feed at
//!   `thresholdLevel: 6` (derived `tngoLast` reference price; needs no
//!   IEX license agreement - unlike raw levels 0/5 - so it works on ANY
//!   Tiingo account, free tier included). Subscribe carries the API
//!   key; each frame is a JSON `"A"` message whose `data` is
//!   `[datetime, ticker, price]`. Each connect spends one request
//!   against the hourly quota (a stable stream is then free), so it's
//!   the default for Power subscribers (reconnect headroom) but any
//!   keyed account can opt in.
//!
//! Which one runs is chosen by `DataService.startLiveStream` from
//! `Config.effectiveLiveQuoteProvider()`, mirroring the REST live-quote
//! provider. The per-transport surface is exactly three things: the
//! connect target + handshake headers (`transportConfig`), the
//! subscribe message (`buildSubscribe`), 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 loop 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` is 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 is COOPERATIVE: the read loop polls with a short read
//! timeout (`read_poll_ms`) and re-checks `should_stop` between reads,
//! so `stop()` just sets `should_stop` and `join`s - the read thread
//! returns on its OWN thread and tears the client down there. We do NOT
//! close the socket from another thread: closing an Io-aware TLS stream
//! while a read is in flight crashes the std TLS reader (the read
//! thread faults dereferencing the half-torn-down client). The only
//! cost is up to one `read_poll_ms` of latency on quit, plus (if a quit
//! lands during the blocking connect/handshake) that operation's own
//! timeout.

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

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";

const tiingo_host = "api.tiingo.com";
const tiingo_path = "/iex";

/// 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;
/// Read-poll timeout (ms). The read loop wakes at least this often to
/// re-check `should_stop`, bounding quit latency while streaming. Short
/// enough to feel instant, long enough to be free when idle. Untyped:
/// coerces to the `readTimeout` argument type at the call site.
const read_poll_ms = 250;
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.
///
/// - `.yahoo`: keyless, free, all tiers; base64+protobuf "pricing"
///   frames. Unofficial - can break without notice.
/// - `.tiingo`: official IEX feed via `wss://api.tiingo.com/iex` at
///   `thresholdLevel: 6` (the derived `tngoLast` reference price, which
///   - unlike raw levels 0/5 - needs no IEX license agreement, so it
///   works on ANY Tiingo account incl. the free tier). Requires an API
///   key in the subscribe message; each connect spends one request
///   against the hourly quota, so it's the default for Power
///   subscribers but any keyed account can opt in.
pub const Transport = enum { yahoo, tiingo };

/// 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`. 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),
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 and join it, then free the
/// subscribed-ticker list. Cooperative: the read loop notices
/// `should_stop` within one `read_poll_ms` and returns on its own
/// thread (we never touch the socket/client from here - see the
/// shutdown note at the top). Safe to call when not running.
pub fn stop(self: *LiveStream) void {
    self.should_stop.store(true, .release);

    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 });
        }
    }
}

/// Parse one frame and update `prices`. Never fails: a bad/unparseable
/// frame is simply ignored (returns without touching `prices`) so the
/// connection stays up. Returns `void` precisely because there's no
/// error worth propagating to the read loop.
pub fn serverMessage(self: *LiveStream, data: []const u8) void {
    const tick = switch (self.transport) {
        .yahoo => parseYahooMessage(self.allocator, data),
        .tiingo => parseTiingoMessage(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,
        },
        .tiingo => .{
            .host = tiingo_host,
            .port = tls_port,
            .path = tiingo_path,
            .headers = "Host: " ++ tiingo_host,
        },
    };
}

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

    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;
    };
    // The client is created, used, and torn down entirely on THIS (read)
    // thread - no other thread ever touches it. That's what makes the
    // cooperative shutdown safe (see the shutdown note at the top): on
    // `should_stop`, `readUntilStopped` returns and this defer closes +
    // frees the client here, not from the quitting thread.
    defer {
        cptr.deinit();
        self.allocator.destroy(cptr);
    }

    try cptr.handshake(cfg.path, .{
        .timeout_ms = handshake_timeout_ms,
        .headers = cfg.headers,
    });
    try self.sendSubscribe(cptr);
    try self.readUntilStopped(cptr);
}

/// Cooperative read loop. Polls with a `read_poll_ms` timeout so a
/// no-data wait returns `null` (WouldBlock) and we re-check
/// `should_stop` between reads - letting the thread exit on its own
/// when asked to stop. Replaces websocket.zig's `readLoop`, which
/// blocks indefinitely (read timeout 0) and trips `unreachable` on a
/// timeout, so it can only be unblocked by a cross-thread close - the
/// exact thing that crashes the Io-aware TLS reader. We replicate its
/// frame handling: text/binary -> parser; ping -> pong; close -> done.
fn readUntilStopped(self: *LiveStream, c: *websocket.Client) !void {
    try c.readTimeout(read_poll_ms);
    while (!self.should_stop.load(.acquire)) {
        const message = c.read() catch |err| switch (err) {
            error.Closed => return,
            else => return err,
        } orelse continue; // timeout / no data yet -> re-check should_stop
        defer c.done(message);
        switch (message.type) {
            .text, .binary => self.serverMessage(message.data),
            // @constCast is safe: data points into our own read buffer.
            .ping => c.writeFrame(.pong, @constCast(message.data)) catch |err| log.debug("pong: {t}", .{err}),
            .close => return,
            .pong => {},
        }
    }
}

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.api_key, 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. `api_key` is used
/// only by the Tiingo transport (its subscribe carries authorization);
/// Yahoo ignores it.
fn buildSubscribe(transport: Transport, allocator: std.mem.Allocator, api_key: ?[]const u8, 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, "]}");
        },
        .tiingo => {
            // {"eventName":"subscribe","authorization":"<key>",
            //  "eventData":{"thresholdLevel":6,"tickers":["spy",...]}}
            // The key and tickers are alphanumeric (plus '.'/'-'), so no
            // JSON string escaping is required.
            try buf.appendSlice(allocator, "{\"eventName\":\"subscribe\",\"authorization\":\"");
            try buf.appendSlice(allocator, api_key orelse "");
            try buf.appendSlice(allocator, "\",\"eventData\":{\"thresholdLevel\":6,\"tickers\":[");
            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
}

// ── Tiingo frame parsing ──────────────────────────────────────

/// Parse one Tiingo IEX text frame. Returns a tick only for
/// `thresholdLevel: 6` "A" (new-data) reference-price messages, whose
/// `data` array is `[datetime, ticker, referencePrice]`. Heartbeats
/// ("H"), subscription acks ("I"), and any malformed/unknown frame
/// yield null. Pure and allocator-scoped (parse memory freed before
/// return), so fully fixture-testable.
fn parseTiingoMessage(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 "A" carries price data; "H" (heartbeat) / "I" (ack) do not.
    const mt = jsonStr(obj.get("messageType")) orelse return null;
    if (!std.mem.eql(u8, mt, "A")) return null;

    const data = switch (obj.get("data") orelse return null) {
        .array => |a| a.items,
        else => return null,
    };
    // Reference format: data[0]=datetime, data[1]=ticker, data[2]=price.
    if (data.len < 3) return null;
    const tkr = jsonStr(data[1]) orelse return null;
    const price = optFloat(data[2]) orelse return null;

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

// ── 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();

    s.serverMessage(
        \\{"type":"pricing","message":"CgRBQVBMFUgBjUMY4JKAz+JnKgNOTVMwCDgBRQisH79InK3UKmUAj+K/sAFQ2AEE"}
    );
    s.serverMessage(
        \\{"type":"pricing","message":"CgNTUFkVUhg5RBiwooDP4mcqA1BDWDAUOAFF9v3HP0jQ+bonZYA9NkGwAVDYAQQ="}
    );
    // A non-pricing frame is ignored (early return, no entry, no crash).
    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"}
    ;
    s.serverMessage(frame);
    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, null, &.{"SPY"}, &buf);
    try std.testing.expectEqualStrings("{\"subscribe\":[\"SPY\"]}", buf.items);

    buf.clearRetainingCapacity();
    try buildSubscribe(.yahoo, a, null, &.{ "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, null, &.{}, &buf);
    try std.testing.expectEqualStrings("{\"subscribe\":[]}", buf.items);
}

test "buildSubscribe: tiingo message carries auth + thresholdLevel 6" {
    const a = std.testing.allocator;
    var buf: std.ArrayList(u8) = .empty;
    defer buf.deinit(a);

    try buildSubscribe(.tiingo, a, "secret-key", &.{ "spy", "aapl" }, &buf);
    try std.testing.expectEqualStrings(
        "{\"eventName\":\"subscribe\",\"authorization\":\"secret-key\",\"eventData\":{\"thresholdLevel\":6,\"tickers\":[\"spy\",\"aapl\"]}}",
        buf.items,
    );
}

test "parseTiingoMessage: threshold-6 A message yields ticker + price" {
    const body =
        \\{"messageType":"A","service":"iex","data":["2026-06-29T13:33:45.383-05:00","spy",612.595]}
    ;
    const tick = parseTiingoMessage(std.testing.allocator, body) orelse return error.TestExpectedTick;
    try std.testing.expectEqualStrings("spy", tick.ticker());
    try std.testing.expectApproxEqAbs(@as(f64, 612.595), tick.price, 0.0001);
}

test "parseTiingoMessage: heartbeat / ack / malformed frames yield null" {
    const a = std.testing.allocator;
    // Heartbeat "H" and ack "I" carry no price data.
    try std.testing.expect(parseTiingoMessage(a,
        \\{"messageType":"H","response":{"code":200,"message":"HeartBeat"}}
    ) == null);
    try std.testing.expect(parseTiingoMessage(a,
        \\{"messageType":"I","response":{"code":200},"data":{"subscriptionId":1}}
    ) == null);
    // "A" but data too short.
    try std.testing.expect(parseTiingoMessage(a,
        \\{"messageType":"A","data":["2026-06-29","spy"]}
    ) == null);
    // "A" but price isn't a number.
    try std.testing.expect(parseTiingoMessage(a,
        \\{"messageType":"A","data":["2026-06-29","spy","oops"]}
    ) == null);
    // Not JSON / array root.
    try std.testing.expect(parseTiingoMessage(a, "not json") == null);
    try std.testing.expect(parseTiingoMessage(a, "[1,2,3]") == null);
}