//! subscribe filter options
//!
//! v1 (/subscribe): wantedCollections / wantedDids, quirks included
//! (docs/upstream-subscribe-protocol.md): prefix patterns get no NSID
//! validation of the head, caps fire post-dedupe, identity/account events
//! bypass wantedCollections, empty-collection commits bypass the filter.
//!
//! v2 (network.bsky.jetstream.subscribeEvents): kinds / dids / collections —
//! three orthogonal AND-composed predicates, each match-all when omitted.
//! collections constrains only commit events; kinds is validated against the
//! lexicon enum; the legacy v1 names are rejected upstream of this parser.

const std = @import("std");
const zat = @import("zat");
const wire = @import("wire.zig");

const Allocator = std.mem.Allocator;

pub const max_wanted_collections = 100;
pub const max_wanted_dids = 10_000;

pub const ParseError = error{
    InvalidCollection,
    TooManyCollections,
    InvalidDid,
    InvalidEncoding,
    InvalidKind,
    CollectionsWithoutCommits,
    TooManyDids,
    OutOfMemory,
};

pub const KindSet = std.enums.EnumSet(wire.Kind);

pub const Filter = struct {
    /// exact collection NSIDs; null means match-all (no collection filter)
    collections: ?std.StringHashMapUnmanaged(void) = null,
    /// prefixes ending in '.', e.g. "app.bsky."
    prefixes: std.ArrayList([]const u8) = .empty,
    /// wanted DIDs; null means match-all
    dids: ?std.StringHashMapUnmanaged(void) = null,
    /// v2 kinds predicate; full set means match-all (v1 always full)
    kinds: KindSet = .full,
    arena: std.heap.ArenaAllocator,

    pub fn init(allocator: Allocator) Filter {
        return .{ .arena = std.heap.ArenaAllocator.init(allocator) };
    }

    pub fn deinit(self: *Filter) void {
        self.arena.deinit();
    }

    /// parse the query string of a /subscribe URL (the part after '?').
    pub fn parseQuery(allocator: Allocator, query: []const u8) ParseError!Filter {
        var f = Filter.init(allocator);
        errdefer f.deinit();
        var it = std.mem.splitScalar(u8, query, '&');
        while (it.next()) |pair| {
            if (pair.len == 0) continue;
            const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue;
            const key = pair[0..eq];
            const value = try percentDecodeAlloc(f.arena.allocator(), pair[eq + 1 ..]);
            if (std.mem.eql(u8, key, "wantedCollections")) {
                try f.addCollection(value);
            } else if (std.mem.eql(u8, key, "wantedDids")) {
                try f.addDid(value);
            }
        }
        return f;
    }

    /// parse a network.bsky.jetstream.subscribeEvents query: kinds / dids /
    /// collections. an out-of-enum kind and a collections filter that can
    /// never apply (kinds present but excluding commit) are both errors —
    /// the caller maps them to pre-upgrade HTTP 400s.
    pub fn parseQueryV2(allocator: Allocator, query: []const u8) ParseError!Filter {
        var f = Filter.init(allocator);
        errdefer f.deinit();
        var saw_kinds = false;
        var saw_collections = false;
        var kinds = KindSet.empty;
        var it = std.mem.splitScalar(u8, query, '&');
        while (it.next()) |pair| {
            if (pair.len == 0) continue;
            const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue;
            const key = pair[0..eq];
            const value = try percentDecodeAlloc(f.arena.allocator(), pair[eq + 1 ..]);
            if (std.mem.eql(u8, key, "collections")) {
                saw_collections = true;
                try f.addCollection(value);
            } else if (std.mem.eql(u8, key, "dids")) {
                try f.addDid(value);
            } else if (std.mem.eql(u8, key, "kinds")) {
                saw_kinds = true;
                const kind = std.meta.stringToEnum(wire.Kind, value) orelse return error.InvalidKind;
                kinds.insert(kind);
            }
        }
        if (saw_kinds) {
            if (saw_collections and !kinds.contains(.commit)) return error.CollectionsWithoutCommits;
            f.kinds = kinds;
        }
        return f;
    }

    pub fn addCollection(self: *Filter, raw: []const u8) ParseError!void {
        const alloc = self.arena.allocator();
        if (std.mem.endsWith(u8, raw, ".*")) {
            // prefix pattern: trim the '*', keep the trailing '.'
            // v1 quirk: no NSID validation of the head
            const prefix = raw[0 .. raw.len - 1];
            for (self.prefixes.items) |p| {
                if (std.mem.eql(u8, p, prefix)) return; // dedupe
            }
            try self.prefixes.append(alloc, try alloc.dupe(u8, prefix));
        } else {
            if (zat.Nsid.parse(raw) == null) return error.InvalidCollection;
            if (self.collections == null) self.collections = .empty;
            const gop = self.collections.?.getOrPut(alloc, raw) catch return error.OutOfMemory;
            if (!gop.found_existing) gop.key_ptr.* = alloc.dupe(u8, raw) catch return error.OutOfMemory;
        }
        // cap fires post-dedupe on the unique count
        const unique = (if (self.collections) |c| c.count() else 0) + self.prefixes.items.len;
        if (unique > max_wanted_collections) return error.TooManyCollections;
    }

    pub fn addDid(self: *Filter, raw: []const u8) ParseError!void {
        const alloc = self.arena.allocator();
        if (zat.Did.parse(raw) == null) return error.InvalidDid;
        if (self.dids == null) self.dids = .empty;
        const gop = self.dids.?.getOrPut(alloc, raw) catch return error.OutOfMemory;
        if (!gop.found_existing) gop.key_ptr.* = alloc.dupe(u8, raw) catch return error.OutOfMemory;
        if (self.dids.?.count() > max_wanted_dids) return error.TooManyDids;
    }

    /// does this filter want a frame?
    pub fn wants(self: *const Filter, kind: wire.Kind, did: []const u8, collection: []const u8) bool {
        if (!self.kinds.contains(kind)) return false;
        if (self.dids) |dids| {
            if (!dids.contains(did)) return false;
        }
        // identity/account/sync events bypass wantedCollections entirely;
        // commits with an empty collection also bypass the filter
        if (kind != .commit or collection.len == 0) return true;
        if (self.collections == null and self.prefixes.items.len == 0) return true;
        if (self.collections) |c| {
            if (c.contains(collection)) return true;
        }
        for (self.prefixes.items) |prefix| {
            if (std.mem.startsWith(u8, collection, prefix)) return true;
        }
        return false;
    }
};

// === tests ===

const testing = std.testing;

test "match-all when no params" {
    var f = try Filter.parseQuery(testing.allocator, "");
    defer f.deinit();
    try testing.expect(f.wants(.commit, "did:plc:x", "app.bsky.feed.post"));
    try testing.expect(f.wants(.identity, "did:plc:x", ""));
}

fn percentDecodeAlloc(allocator: Allocator, input: []const u8) ParseError![]const u8 {
    const out = allocator.alloc(u8, input.len) catch return error.OutOfMemory;
    var read: usize = 0;
    var write: usize = 0;
    while (read < input.len) {
        switch (input[read]) {
            '%' => {
                if (read + 2 >= input.len) return error.InvalidEncoding;
                out[write] = std.fmt.parseInt(u8, input[read + 1 .. read + 3], 16) catch
                    return error.InvalidEncoding;
                read += 3;
            },
            '+' => {
                out[write] = ' ';
                read += 1;
            },
            else => |c| {
                out[write] = c;
                read += 1;
            },
        }
        write += 1;
    }
    return out[0..write];
}

test "v2 query: kinds, dids, collections compose as AND predicates" {
    var f = try Filter.parseQueryV2(testing.allocator, "kinds=commit&kinds=account&collections=app.bsky.feed.post&collections=sh.tangled.*");
    defer f.deinit();
    try testing.expect(f.wants(.commit, "did:plc:x", "app.bsky.feed.post"));
    try testing.expect(f.wants(.commit, "did:plc:x", "sh.tangled.repo"));
    try testing.expect(!f.wants(.commit, "did:plc:x", "app.bsky.graph.follow"));
    // collections constrains only commits; account passes because kinds allows it
    try testing.expect(f.wants(.account, "did:plc:x", ""));
    // identity excluded by kinds even though collections would not apply
    try testing.expect(!f.wants(.identity, "did:plc:x", ""));

    try testing.expectError(error.InvalidKind, Filter.parseQueryV2(testing.allocator, "kinds=bogus"));
    try testing.expectError(error.CollectionsWithoutCommits, Filter.parseQueryV2(testing.allocator, "kinds=identity&collections=app.bsky.feed.post"));

    // omitted kinds = match-all, sync included (v2 emits sync)
    var all = try Filter.parseQueryV2(testing.allocator, "");
    defer all.deinit();
    try testing.expect(all.wants(.sync, "did:plc:x", ""));
}

test "collection exact and prefix matching" {
    var f = try Filter.parseQuery(testing.allocator, "wantedCollections=app.bsky.feed.post&wantedCollections=sh.tangled.*");
    defer f.deinit();
    try testing.expect(f.wants(.commit, "d", "app.bsky.feed.post"));
    try testing.expect(!f.wants(.commit, "d", "app.bsky.feed.like"));
    try testing.expect(f.wants(.commit, "d", "sh.tangled.repo"));
    // identity/account bypass collections
    try testing.expect(f.wants(.identity, "d", ""));
    try testing.expect(f.wants(.account, "d", ""));
    // empty-collection commit bypasses the filter
    try testing.expect(f.wants(.commit, "d", ""));
}

test "two-segment prefix is accepted (v1 quirk)" {
    var f = Filter.init(testing.allocator);
    defer f.deinit();
    try f.addCollection("app.bsky.*");
    try testing.expect(f.wants(.commit, "d", "app.bsky.feed.post"));
}

test "invalid collection and did rejected" {
    var f = Filter.init(testing.allocator);
    defer f.deinit();
    try testing.expectError(error.InvalidCollection, f.addCollection("not a nsid"));
    try testing.expectError(error.InvalidCollection, f.addCollection("app.bsky.fo*"));
    try testing.expectError(error.InvalidDid, f.addDid("bob"));
}

test "query values are percent decoded" {
    var f = try Filter.parseQuery(testing.allocator, "wantedDids=did%3Aplc%3Aabc&wantedCollections=app.bsky.feed.post");
    defer f.deinit();
    try testing.expect(f.wants(.commit, "did:plc:abc", "app.bsky.feed.post"));
    try testing.expect(!f.wants(.commit, "did:plc:other", "app.bsky.feed.post"));
}

test "did filtering applies to all kinds" {
    var f = try Filter.parseQuery(testing.allocator, "wantedDids=did:plc:aaa");
    defer f.deinit();
    try testing.expect(f.wants(.commit, "did:plc:aaa", "c"));
    try testing.expect(!f.wants(.commit, "did:plc:bbb", "c"));
    try testing.expect(!f.wants(.identity, "did:plc:bbb", ""));
}

test "caps fire post-dedupe" {
    var f = Filter.init(testing.allocator);
    defer f.deinit();
    // the same collection many times never trips the cap
    var i: usize = 0;
    while (i < 200) : (i += 1) try f.addCollection("app.bsky.feed.post");
    try testing.expectEqual(@as(u32, 1), f.collections.?.count());
}