music-atmosphere-feed / src / jetstream.zig

const std = @import("std");
const mem = std.mem;
const json = std.json;
const posix = std.posix;
const Allocator = mem.Allocator;
const websocket = @import("websocket");
const filter = @import("feed/filter.zig");
const stats = @import("server/stats.zig");
const posts = @import("store/posts.zig");

// turbostream: hydrated jetstream from graze.social
const TURBOSTREAM_HOST = "api.graze.social";
const TURBOSTREAM_PATH = "/app/api/v1/turbostream/turbostream";

pub const Post = struct {
    uri: []const u8,
    cid: []const u8,
    text: []const u8,
    did: []const u8,
    rkey: []const u8,
};

pub fn consumer(allocator: Allocator) void {
    var backoff: u64 = 1;
    const max_backoff: u64 = 60;

    while (true) {
        connect(allocator) catch |err| {
            std.debug.print("turbostream error: {}, reconnecting in {}s...\n", .{ err, backoff });
        };
        posix.nanosleep(backoff, 0);
        backoff = @min(backoff * 2, max_backoff);
    }
}

fn connect(allocator: Allocator) !void {
    const host = posix.getenv("TURBOSTREAM_HOST") orelse TURBOSTREAM_HOST;
    const path = TURBOSTREAM_PATH;

    std.debug.print("connecting to wss://{s}{s}\n", .{ host, path });

    var client = websocket.Client.init(allocator, .{
        .host = host,
        .port = 443,
        .tls = true,
        .max_size = 1024 * 1024 * 4, // 4MB for batched messages
    }) catch |err| {
        std.debug.print("websocket client init failed: {}\n", .{err});
        return err;
    };
    defer client.deinit();

    var host_header_buf: [256]u8 = undefined;
    const host_header = std.fmt.bufPrint(&host_header_buf, "Host: {s}\r\n", .{host}) catch host;

    client.handshake(path, .{ .headers = host_header }) catch |err| {
        std.debug.print("websocket handshake failed: {}\n", .{err});
        return err;
    };

    std.debug.print("turbostream connected!\n", .{});
    stats.get().recordConnected();

    var handler = Handler{ .allocator = allocator };
    client.readLoop(&handler) catch |err| {
        std.debug.print("websocket read loop error: {}\n", .{err});
        return err;
    };
}

const Handler = struct {
    allocator: Allocator,
    msg_count: usize = 0,
    last_summary_time: i64 = 0,

    pub fn serverMessage(self: *Handler, data: []const u8) !void {
        self.msg_count += 1;

        self.processRecord(data) catch |err| {
            if (err != error.NotAPost and err != error.NoMatch) {
                std.debug.print("processing error: {}\n", .{err});
            }
        };

        // periodic summary every 30 seconds
        const now = std.time.timestamp();
        if (now - self.last_summary_time >= 30) {
            self.last_summary_time = now;
            const s = stats.get();
            const lag_ms = s.getPostLagMs();
            const lag_sec = @divTrunc(lag_ms, 1000);
            const status = s.getStatus();
            std.debug.print("turbostream: msgs={} matches={} lag={}s status={s}\n", .{
                s.getMessages(),
                s.getMatches(),
                lag_sec,
                status,
            });
        }
    }

    pub fn close(_: *Handler) void {
        std.debug.print("turbostream connection closed\n", .{});
    }

    fn processRecord(self: *Handler, payload: []const u8) !void {
        const parsed = json.parseFromSlice(json.Value, self.allocator, payload, .{}) catch return error.ParseError;
        defer parsed.deinit();

        if (parsed.value != .object) return error.NotAPost;
        const record_obj = parsed.value.object;

        // turbostream format: { at_uri, did, time_us, message: {...}, hydrated_metadata: {...} }
        const message = record_obj.get("message") orelse return error.NotAPost;
        if (message != .object) return error.NotAPost;
        const msg = message.object;

        // check kind
        const kind = msg.get("kind") orelse return error.NotAPost;
        if (kind != .string or !mem.eql(u8, kind.string, "commit")) return error.NotAPost;

        // get commit
        const commit = msg.get("commit") orelse return error.NotAPost;
        if (commit != .object) return error.NotAPost;

        // check collection - turbostream sends all collections, filter to posts
        const collection = commit.object.get("collection") orelse return error.NotAPost;
        if (collection != .string or !mem.eql(u8, collection.string, "app.bsky.feed.post")) return error.NotAPost;

        // check operation (create only)
        const operation = commit.object.get("operation") orelse return error.NotAPost;
        if (operation != .string or !mem.eql(u8, operation.string, "create")) return error.NotAPost;

        // get rkey and parse TID for timestamp
        const rkey_val = commit.object.get("rkey") orelse return error.NotAPost;
        if (rkey_val != .string) return error.NotAPost;
        const rkey = rkey_val.string;

        const post_time_ms = posts.parseTidTimestamp(rkey);

        // get time_us from message
        const time_us: i64 = if (msg.get("time_us")) |t| (if (t == .integer) t.integer else 0) else 0;
        stats.get().recordEvent(time_us, post_time_ms);

        // get cid
        const cid_val = commit.object.get("cid") orelse return error.NotAPost;
        if (cid_val != .string) return error.NotAPost;

        // get post record
        const post_record = commit.object.get("record") orelse return error.NotAPost;
        if (post_record != .object) return error.NotAPost;

        // check if post matches filter criteria
        if (!filter.matches(post_record.object)) return error.NoMatch;

        // get uri from turbostream (it provides at_uri directly)
        const at_uri = record_obj.get("at_uri") orelse return error.NotAPost;
        if (at_uri != .string) return error.NotAPost;

        // add to feed
        posts.add(at_uri.string, cid_val.string) catch |err| {
            std.debug.print("failed to add post: {}\n", .{err});
            return;
        };

        const s = stats.get();
        s.recordMatch();

        // track all platforms in post
        const platforms = filter.detectAllPlatformsFromRecord(post_record.object);
        if (platforms.soundcloud) s.recordPlatform(.soundcloud);
        if (platforms.bandcamp) s.recordPlatform(.bandcamp);
        if (platforms.spotify) s.recordPlatform(.spotify);
        if (platforms.plyr) s.recordPlatform(.plyr);
        if (platforms.apple) s.recordPlatform(.apple);

        // track match types
        if (filter.isQuoteMatch(post_record.object)) s.recordQuoteMatch();
        if (platforms.count() >= 2) s.recordMultiPlatform();

        std.debug.print("added: {s}\n", .{at_uri.string});
    }
};