-- gRPC streaming over fiber channels: server-streaming, client-streaming,
-- and bidirectional flavors, run end-to-end through the loopback transport.
--
-- Parameterized over both codegen modes since the generated client/server
-- bodies differ between full and runtime mode (different per-message
-- encode/decode call sites).

local t     = require('luatest')
local fiber = require('fiber')
local pb    = require('pb')

local MODES = {'full', 'runtime'}

for _, mode in ipairs(MODES) do
    local hello = require(mode .. '.hello.hello_pb')
    local g = t.group('grpc_stream.' .. mode)

    -- ---------------------------------------------------------------------
    -- Server-streaming: one request in, many replies out.
    -- ---------------------------------------------------------------------

    g.test_server_stream_yields_replies_in_order = function()
        local impl = {
            StreamHellos = function(req, stream, _)
                for i = 1, 3 do
                    stream:send({greeting = req.name .. '#' .. tostring(i)})
                end
            end,
        }
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))

        local stream = client.StreamHellos({name = 'X'}, {})
        local replies = {}
        while true do
            local r, err = stream:recv()
            if r == nil then
                t.assert_equals(err, nil, 'unexpected stream error: ' .. tostring(err))
                break
            end
            replies[#replies + 1] = r.greeting
        end
        t.assert_equals(replies, {'X#1', 'X#2', 'X#3'})
    end

    g.test_server_stream_handler_can_be_empty = function()
        -- A handler that returns without sending anything is a valid
        -- (empty) stream — recv() must return nil on the first call.
        local impl = {StreamHellos = function(_, _, _) end}
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))
        local stream = client.StreamHellos({name = 'x'}, {})
        local r, err = stream:recv()
        t.assert_equals(r, nil)
        t.assert_equals(err, nil)
    end

    g.test_server_stream_handler_error_surfaces = function()
        local impl = {
            StreamHellos = function(_, stream, _)
                stream:send({greeting = 'first'})
                error('boom from handler')
            end,
        }
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))
        local stream = client.StreamHellos({name = 'x'}, {})

        -- First message comes through cleanly.
        local r1 = stream:recv()
        t.assert_equals(r1.greeting, 'first')

        -- Then end-of-stream with the handler's error string.
        local r2, err = stream:recv()
        t.assert_equals(r2, nil)
        t.assert_str_contains(tostring(err), 'boom from handler')
    end

    g.test_missing_streaming_handler_errors = function()
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server({})))
        local stream = client.StreamHellos({name = 'x'}, {})
        local r, err = stream:recv()
        t.assert_equals(r, nil)
        t.assert_str_contains(tostring(err), 'StreamHellos: handler missing')
    end

    -- ---------------------------------------------------------------------
    -- Client-streaming: many requests in, one reply out.
    -- ---------------------------------------------------------------------

    g.test_client_stream_collects_and_replies = function()
        local impl = {
            CollectHellos = function(stream, _)
                local names = {}
                while true do
                    local req = stream:recv()
                    if req == nil then break end
                    names[#names + 1] = req.name
                end
                return {greeting = 'collected: ' .. table.concat(names, ',')}
            end,
        }
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))

        local call = client.CollectHellos({})
        call:send({name = 'a'})
        call:send({name = 'b'})
        call:send({name = 'c'})
        call:close_send()

        local reply, err = call:recv()
        t.assert_equals(err, nil)
        t.assert_equals(reply.greeting, 'collected: a,b,c')

        -- recv after the response should return nil (stream ended).
        local tail, err2 = call:recv()
        t.assert_equals(tail, nil)
        t.assert_equals(err2, nil)
    end

    g.test_client_stream_empty_is_valid = function()
        local impl = {
            CollectHellos = function(stream, _)
                t.assert_equals(stream:recv(), nil)
                return {greeting = 'empty'}
            end,
        }
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))
        local call = client.CollectHellos({})
        call:close_send()
        local reply = call:recv()
        t.assert_equals(reply.greeting, 'empty')
    end

    -- ---------------------------------------------------------------------
    -- Bidirectional streaming: both sides send + recv independently.
    -- ---------------------------------------------------------------------

    g.test_bidi_echo_loop = function()
        -- Server echoes each request back with a "you said: " prefix.
        local impl = {
            Chat = function(stream, _)
                while true do
                    local req = stream:recv()
                    if req == nil then break end
                    stream:send({greeting = 'you said: ' .. req.name})
                end
            end,
        }
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))

        local call = client.Chat({})

        -- Drive the client side from a separate fiber so we can interleave
        -- sends and receives without deadlocking.
        local reader_done = fiber.channel(1)
        local received = {}
        fiber.create(function()
            while true do
                local r, err = call:recv()
                if r == nil then
                    reader_done:put(err or false)
                    break
                end
                received[#received + 1] = r.greeting
            end
        end)

        call:send({name = 'foo'})
        call:send({name = 'bar'})
        call:send({name = 'baz'})
        call:close_send()

        local terminal = reader_done:get(5)
        t.assert(terminal == false, 'reader saw error: ' .. tostring(terminal))
        t.assert_equals(received, {
            'you said: foo', 'you said: bar', 'you said: baz',
        })
    end

    g.test_bidi_handler_error_surfaces = function()
        local impl = {
            Chat = function(_, _)
                error('chat exploded')
            end,
        }
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))
        local call = client.Chat({})
        call:close_send()
        local r, err = call:recv()
        t.assert_equals(r, nil)
        t.assert_str_contains(tostring(err), 'chat exploded')
    end

    g.test_bidi_cancel_stops_further_recv = function()
        -- Server sends forever; client cancels after one message.
        local server_finished = fiber.channel(1)
        local impl = {
            Chat = function(stream, _)
                local i = 0
                while stream:send({greeting = 'tick:' .. tostring(i)}) do
                    i = i + 1
                    if i > 100 then break end -- safety
                    fiber.yield()
                end
                server_finished:put(i)
            end,
        }
        local client = hello.Greeter_client(pb.grpc.loopback(
            hello.Greeter_server(impl)))
        local call = client.Chat({})
        local first = call:recv()
        t.assert_str_contains(first.greeting, 'tick:')
        call:cancel()

        -- Give the server fiber a chance to observe the cancel.
        local final_i = server_finished:get(5)
        t.assert(final_i ~= nil, 'server did not stop after cancel')
        t.assert(final_i <= 100, 'server kept running past safety bound')
    end
end

-- ---------------------------------------------------------------------------
-- pb.grpc.new_stream_pair: unit-level coverage of the primitive itself
-- (independent of generated code).
-- ---------------------------------------------------------------------------

local u = t.group('grpc_stream.primitive')

u.test_send_recv_byte_passthrough = function()
    local client, server = pb.grpc.new_stream_pair()
    fiber.create(function()
        local req = server:recv()
        server:send('reply:' .. req)
        server:_finish(nil)
    end)
    client:send('ping')
    client:close_send()
    local out = client:recv()
    t.assert_equals(out, 'reply:ping')
    t.assert_equals(client:recv(), nil)
end

u.test_finish_with_error_propagates = function()
    local client, server = pb.grpc.new_stream_pair()
    fiber.create(function()
        server:_finish('handler error')
    end)
    local b, err = client:recv()
    t.assert_equals(b, nil)
    t.assert_equals(err, 'handler error')
end

u.test_send_after_close_send_errors = function()
    local client, _ = pb.grpc.new_stream_pair()
    client:close_send()
    t.assert_error(function() client:send('nope') end)
end

-- ---------------------------------------------------------------------------
-- multiplex: streaming methods route correctly across multiple servers.
-- ---------------------------------------------------------------------------

local m = t.group('grpc_stream.multiplex')

m.test_multiplex_routes_streams_to_right_server = function()
    local hello = require('full.hello.hello_pb')
    -- Two independent Greeter servers; multiplex would normally complain
    -- about the duplicate paths, so for this test we keep just one with a
    -- streaming impl and verify routing works via the multiplex transport.
    local impl = {
        StreamHellos = function(req, stream, _)
            stream:send({greeting = 'mp:' .. req.name})
        end,
    }
    local mux = pb.grpc.multiplex({hello.Greeter_server(impl)})
    local client = hello.Greeter_client(mux)
    local s = client.StreamHellos({name = 'X'}, {})
    local r = s:recv()
    t.assert_equals(r.greeting, 'mp:X')
    t.assert_equals(s:recv(), nil)
end

m.test_multiplex_duplicate_streaming_path_errors = function()
    -- Construct two minimal server tables that share a streaming path so
    -- we hit the duplicate-stream branch without also tripping the
    -- duplicate-unary branch.
    local stream_entry = {
        kind = 'server_stream',
        handler = function() end,
    }
    local s1 = {methods = {}, streams = {['/dup/Path'] = stream_entry}}
    local s2 = {methods = {}, streams = {['/dup/Path'] = stream_entry}}
    t.assert_error_msg_contains('duplicate streaming route', function()
        pb.grpc.multiplex({s1, s2})
    end)
end