%% Copyright (c) 2019-2023, Loïc Hoguin <essen@ninenines.eu>
%%
%% Permission to use, copy, modify, and/or distribute this software for any
%% purpose with or without fee is hereby granted, provided that the above
%% copyright notice and this permission notice appear in all copies.
%%
%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

-module(flow_SUITE).
-compile(export_all).
-compile(nowarn_export_all).

-import(ct_helper, [doc/1]).

all() ->
	[{group, flow}].

groups() ->
	[{flow, [parallel], ct_helper:all(?MODULE)}].

%% Tests.

default_flow_http(_) ->
	doc("Confirm flow control default can be changed and overriden for HTTP/1.1."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, date}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		%% First we check that we can set the flow for the entire connection.
		{ok, ConnPid1} = gun:open("localhost", Port, #{
			http_opts => #{flow => 1}
		}),
		{ok, http} = gun:await_up(ConnPid1),
		StreamRef1 = gun:get(ConnPid1, "/"),
		{response, nofin, 200, _} = gun:await(ConnPid1, StreamRef1),
		{data, nofin, _} = gun:await(ConnPid1, StreamRef1),
		{error, timeout} = gun:await(ConnPid1, StreamRef1, 1500),
		gun:close(ConnPid1),
		%% Then we confirm that we can override it per request.
		{ok, ConnPid2} = gun:open("localhost", Port, #{
			http_opts => #{flow => 1}
		}),
		{ok, http} = gun:await_up(ConnPid2),
		StreamRef2 = gun:get(ConnPid2, "/", [], #{flow => 2}),
		{response, nofin, 200, _} = gun:await(ConnPid2, StreamRef2),
		{data, nofin, _} = gun:await(ConnPid2, StreamRef2),
		{data, nofin, _} = gun:await(ConnPid2, StreamRef2),
		{error, timeout} = gun:await(ConnPid2, StreamRef2, 1500),
		gun:close(ConnPid2)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

default_flow_http2(_) ->
	doc("Confirm flow control default can be changed and overriden for HTTP/2."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, 40000}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		%% First we check that we can set the flow for the entire connection.
		{ok, ConnPid} = gun:open("localhost", Port, #{
			http2_opts => #{
				flow => 1,
				%% We set the max frame size to the same as the initial
				%% window size in order to reduce the number of data messages.
				initial_connection_window_size => 65535,
				initial_stream_window_size => 65535,
				max_frame_size_received => 65535
			},
			protocols => [http2]
		}),
		{ok, http2} = gun:await_up(ConnPid),
		StreamRef1 = gun:get(ConnPid, "/"),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef1),
		%% We set the flow to 1 therefore we will receive *2* data messages,
		%% and then nothing because the window was fully consumed.
		{data, nofin, _} = gun:await(ConnPid, StreamRef1),
		{data, nofin, _} = gun:await(ConnPid, StreamRef1),
		{error, timeout} = gun:await(ConnPid, StreamRef1, 1500),
		%% Then we confirm that we can override it per request.
		StreamRef2 = gun:get(ConnPid, "/", [], #{flow => 2}),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef2),
		%% We set the flow to 2 but due to the ensure_window algorithm
		%% we end up receiving *4* data messages before we consume
		%% the window.
		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
		{data, nofin, _} = gun:await(ConnPid, StreamRef2),
		{error, timeout} = gun:await(ConnPid, StreamRef2, 1500),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

flow_http(_) ->
	doc("Confirm flow control works as intended for HTTP/1.1."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, date}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port),
		{ok, http} = gun:await_up(ConnPid),
		StreamRef = gun:get(ConnPid, "/", [], #{flow => 1}),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
		%% We set the flow to 1 therefore we will receive 1 data message,
		%% and then nothing because Gun doesn't read from the socket.
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 3000),
		%% We then update the flow and get 2 more data messages but no more.
		gun:update_flow(ConnPid, StreamRef, 2),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 1000),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

flow_http2(_) ->
	doc("Confirm flow control works as intended for HTTP/2."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, 40000}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port, #{
			%% We set the max frame size to the same as the initial
			%% window size in order to reduce the number of data messages.
			http2_opts => #{
				initial_connection_window_size => 65535,
				initial_stream_window_size => 65535,
				max_frame_size_received => 65535
			},
			protocols => [http2]
		}),
		{ok, http2} = gun:await_up(ConnPid),
		StreamRef = gun:get(ConnPid, "/", [], #{flow => 1}),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
		%% We set the flow to 1 therefore we will receive *2* data messages,
		%% and then nothing because the window was fully consumed.
		{data, nofin, D1} = gun:await(ConnPid, StreamRef),
		{data, nofin, D2} = gun:await(ConnPid, StreamRef),
		%% We consumed all the window available.
		65535 = byte_size(D1) + byte_size(D2),
		{error, timeout} = gun:await(ConnPid, StreamRef, 3500),
		%% We then update the flow and get *5* more data messages but no more.
		gun:update_flow(ConnPid, StreamRef, 2),
		{data, nofin, D3} = gun:await(ConnPid, StreamRef),
		{data, nofin, D4} = gun:await(ConnPid, StreamRef),
		{data, nofin, D5} = gun:await(ConnPid, StreamRef),
		{data, nofin, D6} = gun:await(ConnPid, StreamRef),
		{data, nofin, D7} = gun:await(ConnPid, StreamRef),
		%% We consumed all the window available again.
		%% D3 is the end of the truncated D2, D4, D5 and D6 are full and D7 truncated.
		131070 = byte_size(D3) + byte_size(D4) + byte_size(D5) + byte_size(D6) + byte_size(D7),
		{error, timeout} = gun:await(ConnPid, StreamRef, 1000),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

flow_ws(_) ->
	doc("Confirm flow control works as intended for Websocket."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", ws_echo_h, []}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port),
		{ok, http} = gun:await_up(ConnPid),
		StreamRef = gun:ws_upgrade(ConnPid, "/", [], #{flow => 1}),
		{upgrade, [<<"websocket">>], _} = gun:await(ConnPid, StreamRef),
		%% We send 2 frames with some time in between to make sure that
		%% Gun handles them in separate Protocol:handle calls.
		Frame = {text, <<"Hello!">>},
		gun:ws_send(ConnPid, StreamRef, Frame),
		timer:sleep(500),
		gun:ws_send(ConnPid, StreamRef, Frame),
		%% We set the flow to 1 therefore we will receive 1 data message,
		%% and then nothing because Gun doesn't read from the socket.
		{ws, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 3000),
		%% We then update the flow, send 2 frames with some time in between
		%% and get 2 more data messages but no more.
		gun:update_flow(ConnPid, StreamRef, 2),
		gun:ws_send(ConnPid, StreamRef, Frame),
		timer:sleep(500),
		gun:ws_send(ConnPid, StreamRef, Frame),
		{ws, _} = gun:await(ConnPid, StreamRef),
		{ws, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 1000),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

no_flow_http(_) ->
	doc("Ignore flow updates for no-flow streams for HTTP/1.1."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, date}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port),
		{ok, http} = gun:await_up(ConnPid),
		StreamRef = gun:get(ConnPid, "/", []),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
		gun:update_flow(ConnPid, StreamRef, 2),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

no_flow_http2(_) ->
	doc("Ignore flow updates for no-flow streams for HTTP/2."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, date}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port, #{
			protocols => [http2]
		}),
		{ok, http2} = gun:await_up(ConnPid),
		StreamRef = gun:get(ConnPid, "/", []),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
		gun:update_flow(ConnPid, StreamRef, 2),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		{data, nofin, _} = gun:await(ConnPid, StreamRef),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

no_flow_ws(_) ->
	doc("Ignore flow updates for no-flow streams for Websocket."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", ws_echo_h, []}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port),
		{ok, http} = gun:await_up(ConnPid),
		StreamRef = gun:ws_upgrade(ConnPid, "/", []),
		{upgrade, [<<"websocket">>], _} = gun:await(ConnPid, StreamRef),
		gun:update_flow(ConnPid, StreamRef, 2),
		Frame = {text, <<"Hello!">>},
		gun:ws_send(ConnPid, StreamRef, Frame),
		timer:sleep(100),
		gun:ws_send(ConnPid, StreamRef, Frame),
		{ws, _} = gun:await(ConnPid, StreamRef),
		{ws, _} = gun:await(ConnPid, StreamRef),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

sse_flow_http(_) ->
	doc("Confirm flow control works as intended for HTTP/1.1 "
		"when using the gun_sse_h content handler."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, date}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port, #{
			http_opts => #{content_handlers => [gun_sse_h, gun_data_h]}
		}),
		{ok, http} = gun:await_up(ConnPid),
		StreamRef = gun:get(ConnPid, "/", [], #{flow => 1}),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
		%% We set the flow to 1 therefore we will receive 1 event message,
		%% and then nothing because Gun doesn't read from the socket. We
		%% set the timeout to 2500 to ensure there is only going to be one
		%% message queued up.
		{sse, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 2500),
		%% We then update the flow and get 2 more event messages but no more.
		gun:update_flow(ConnPid, StreamRef, 2),
		{sse, _} = gun:await(ConnPid, StreamRef),
		{sse, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 1000),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.

sse_flow_http2(_) ->
	doc("Confirm flow control works as intended for HTTP/2 "
		"when using the gun_sse_h content handler."),
	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [], #{env => #{
		dispatch => cowboy_router:compile([{'_', [{"/", sse_clock_h, 40000}]}])
	}}),
	Port = ranch:get_port(?FUNCTION_NAME),
	try
		{ok, ConnPid} = gun:open("localhost", Port, #{
			%% We set the max frame size to the same as the initial
			%% window size in order to reduce the number of data messages.
			http2_opts => #{
				content_handlers => [gun_sse_h, gun_data_h],
				initial_connection_window_size => 65535,
				initial_stream_window_size => 65535,
				max_frame_size_received => 65535
			},
			protocols => [http2]
		}),
		{ok, http2} = gun:await_up(ConnPid),
		StreamRef = gun:get(ConnPid, "/", [], #{flow => 1}),
		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
		%% We set the flow to 1 therefore we will receive 1 event message,
		%% and then nothing because the window was fully consumed before
		%% the second event was fully received.
		{sse, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 3000),
		%% We then update the flow and get 3 more event messages but no more.
		%% We get an extra message because of the ensure_window algorithm.
		gun:update_flow(ConnPid, StreamRef, 2),
		{sse, _} = gun:await(ConnPid, StreamRef),
		{sse, _} = gun:await(ConnPid, StreamRef),
		{sse, _} = gun:await(ConnPid, StreamRef),
		{error, timeout} = gun:await(ConnPid, StreamRef, 1000),
		gun:close(ConnPid)
	after
		cowboy:stop_listener(?FUNCTION_NAME)
	end.