chore: solve on_response unatural return (#60)

## Problem

N/A

## Solution

fixes #59 

## Rationale

N/A

Author: zoedsoupe

.wakatime-project

@@ -0,0 +1 @@
+supabase-ex

lib/supabase/fetcher.ex

@@ -218,10 +218,7 @@ defmodule Supabase.Fetcher do
 
   def stream(%Request{http_client: http_client} = builder, on_response, opts)
       when not is_nil(builder.url) do
-    with {:ok, resp} <- http_client.stream(builder, on_response, opts) do
-      {:ok, ResponseAdapter.from(resp)}
-    end
-    |> handle_response(builder)
+    http_client.stream(builder, on_response, opts)
   rescue
     exception -> handle_exception(exception, __STACKTRACE__, builder)
   end

lib/supabase/fetcher/adapter/finch.ex

@@ -89,7 +89,12 @@ defmodule Supabase.Fetcher.Adapter.Finch do
       end
 
     if is_function(on_response, 1) do
-      on_response.({status, headers, stream})
+      case on_response.({status, headers, stream}) do
+        :ok -> :ok
+        {:ok, body} -> {:ok, body}
+        {:error, %Supabase.Error{} = err} -> {:error, err}
+        unexpected -> Supabase.Error.new(service: b.service, metadata: %{raw_error: unexpected})
+      end
     else
       %Finch.Response{
         status: status,

lib/supabase/fetcher/behaviour.ex

@@ -4,12 +4,20 @@ defmodule Supabase.Fetcher.Behaviour do
   alias Supabase.Fetcher.Request
   alias Supabase.Fetcher.Response
 
+  @type on_response_input ::
+          {Supabase.Fetcher.status(), Supabase.Fetcher.headers(), body :: Enumerable.t()}
+  @typedoc """
+  The response handler for streaming responses. It receives the response status, headers, and body as input.
+
+  Note that here only the status and headers are consumed from the stream and so the body reamins unconsumed for custom operations, receiving each chunk of the body as it arrives.
+
+  It needs to return either `:ok` or `{:ok, body}` or `{:error, Supabase.Error}`.
+  """
+  @type on_response :: (on_response_input -> :ok | {:ok, term} | {:error, Supabase.Error.t()})
+
   @callback request(Request.t()) :: Supabase.result(Response.t())
   @callback request_async(Request.t()) :: Supabase.result(Response.t())
   @callback upload(Request.t(), filepath :: Path.t()) :: Supabase.result(Response.t())
   @callback stream(Request.t()) :: Supabase.result(Response.t())
   @callback stream(Request.t(), on_response) :: Supabase.result(Response.t())
-            when on_response: ({Supabase.Fetcher.status(), Supabase.Fetcher.headers(),
-                                body :: Enumerable.t()} ->
-                                 Supabase.result(Response.t()))
 end

mix.exs

@@ -1,7 +1,7 @@
 defmodule Supabase.MixProject do
   use Mix.Project
 
-  @version "0.6.0"
+  @version "0.6.1"
   @source_url "https://github.com/supabase-community/supabase-ex"
 
   def project do

supabase/config.toml

@@ -242,7 +242,7 @@ enabled = false
 # user_pool_region = "us-east-1"
 
 [edge_runtime]
-enabled = false
+enabled = true
 # Configure one of the supported request policies: `oneshot`, `per_worker`.
 # Use `oneshot` for hot reload, or `per_worker` for load testing.
 policy = "oneshot"
@@ -258,6 +258,36 @@ inspector_port = 8083
 # Supported file extensions are: .ts, .js, .mjs, .jsx, .tsx
 # entrypoint = "./functions/MY_FUNCTION_NAME/index.ts"
 
+[functions.binary-data]
+enabled = true
+verify_jwt = false
+entrypoint = "./functions/binary-data/index.ts"
+
+[functions.error-cases]
+enabled = true
+verify_jwt = false
+entrypoint = "./functions/error-cases/index.ts"
+
+[functions.headers-demo]
+enabled = true
+verify_jwt = false
+entrypoint = "./functions/headers-demo/index.ts"
+
+[functions.json-echo]
+enabled = true
+verify_jwt = false
+entrypoint = "./functions/json-echo/index.ts"
+
+[functions.simple-text]
+enabled = true
+verify_jwt = false
+entrypoint = "./functions/simple-text/index.ts"
+
+[functions.stream-data]
+enabled = true
+verify_jwt = false
+entrypoint = "./functions/stream-data/index.ts"
+
 [analytics]
 enabled = false
 port = 54327

supabase/functions/binary-data/index.ts

@@ -0,0 +1,9 @@
+Deno.serve(async (req) => {
+  const data = await req.arrayBuffer();
+  // Reverse the binary data as an example transformation
+  const reversed = new Uint8Array(data).reverse();
+  
+  return new Response(reversed, {
+    headers: { "Content-Type": "application/octet-stream" }
+  });
+});

supabase/functions/error-cases/index.ts

@@ -0,0 +1,30 @@
+Deno.serve(async (req) => {
+  const { error_type } = await req.json();
+  
+  switch (error_type) {
+    case "validation":
+      return new Response(JSON.stringify({
+        error: "Validation failed",
+        details: ["Field 'name' is required"]
+      }), {
+        status: 400,
+        headers: { "Content-Type": "application/json" }
+      });
+      
+    case "unauthorized":
+      return new Response(JSON.stringify({
+        error: "Unauthorized access"
+      }), {
+        status: 401,
+        headers: { "Content-Type": "application/json" }
+      });
+      
+    default:
+      return new Response(JSON.stringify({
+        error: "Internal server error"
+      }), {
+        status: 500,
+        headers: { "Content-Type": "application/json" }
+      });
+  }
+});

supabase/functions/headers-demo/index.ts

@@ -0,0 +1,14 @@
+Deno.serve(async (req) => {
+  const headers = Object.fromEntries(req.headers);
+  const customHeader = req.headers.get("x-custom-header") || "no custom header";
+  
+  return new Response(JSON.stringify({ 
+    headers,
+    customHeader 
+  }), {
+    headers: { 
+      "Content-Type": "application/json",
+      "X-Response-Header": "test-value"
+    }
+  });
+});

supabase/functions/json-echo/index.ts

@@ -0,0 +1,6 @@
+Deno.serve(async (req) => {
+  const body = await req.json();
+  return new Response(JSON.stringify(body), {
+    headers: { "Content-Type": "application/json" }
+  });
+});

supabase/functions/simple-text/index.ts

@@ -0,0 +1,5 @@
+Deno.serve(async (req) => {
+  return new Response("Hello from Deno!", {
+    headers: { "Content-Type": "text/plain" }
+  });
+});

supabase/functions/stream-data/index.ts

@@ -0,0 +1,25 @@
+Deno.serve(async (req) => {
+  const stream = new ReadableStream({
+    start(controller) {
+      let count = 0;
+      const interval = setInterval(() => {
+        if (count >= 5) {
+          clearInterval(interval);
+          controller.close();
+          return;
+        }
+        const message = `Event ${count}\n`;
+        controller.enqueue(new TextEncoder().encode(message));
+        count++;
+      }, 1000);
+    }
+  });
+
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache",
+      "Connection": "keep-alive"
+    }
+  });
+});

test/supabase/fetcher/adapter/finch_test.exs

@@ -60,31 +60,94 @@ defmodule Supabase.Fetcher.Adapter.FinchTest do
   describe "dealing with streams" do
     test "streams a response successfully", %{builder: builder} do
       @mock
-      |> expect(:stream, fn %Request{}, on_response, _opts ->
+      |> expect(:stream, fn %Request{}, _opts ->
         status = 200
         headers = [{"content-length", 80_543}]
         stream = Stream.cycle(["chunk1", "chunk2"])
-        on_response.({status, headers, stream})
+        body = Enum.take(stream, 2) |> Enum.to_list() |> Enum.join(",")
+        {:ok, %Finch.Response{status: status, headers: headers, body: body}}
       end)
 
-      on_response = fn {status, headers, body} ->
-        body = Enum.take(body, 2) |> Enum.to_list() |> Enum.join(",")
-
-        {:ok,
-         [status: status, headers: headers, body: body]
-         |> then(&struct(Finch.Response, &1))}
-      end
-
       builder =
         Request.with_body_decoder(builder, fn %{body: body}, _opts ->
           {:ok, String.split(body, ",", trim: true)}
         end)
 
-      assert {:ok, %Response{} = resp} = Fetcher.stream(builder, on_response)
+      assert {:ok, %Response{} = resp} = Fetcher.stream(builder)
       assert resp.status == 200
       assert Response.get_header(resp, "content-length") == 80_543
       assert resp.body == ["chunk1", "chunk2"]
     end
+
+    test "streams a response successfully with custom on_response handler with no body", %{
+      builder: builder
+    } do
+      @mock
+      |> expect(:stream, fn %Request{}, on_response, _opts ->
+        status = 200
+        headers = [{"content-length", 80_543}]
+        stream = Stream.cycle(["chunk1", "chunk2"])
+        on_response.({status, headers, stream})
+      end)
+
+      on_response = fn {status, headers, body} ->
+        assert status == 200
+        assert List.keyfind(headers, "content-length", 0) == {"content-length", 80_543}
+
+        # very expensing body consuming
+        _ = Enum.take(body, 2) |> Enum.to_list()
+
+        :ok
+      end
+
+      assert :ok = Fetcher.stream(builder, on_response)
+    end
+
+    test "streams a response successfully with custom on_response handler returning body", %{
+      builder: builder
+    } do
+      @mock
+      |> expect(:stream, fn %Request{}, on_response, _opts ->
+        status = 200
+        headers = [{"content-length", 80_543}]
+        stream = Stream.cycle(["chunk1", "chunk2"])
+        on_response.({status, headers, stream})
+      end)
+
+      on_response = fn {status, headers, body} ->
+        assert status == 200
+        assert List.keyfind(headers, "content-length", 0) == {"content-length", 80_543}
+
+        {:ok,
+         body
+         |> Enum.take(2)
+         |> Enum.to_list()}
+      end
+
+      assert {:ok, body} = Fetcher.stream(builder, on_response)
+      assert body == ["chunk1", "chunk2"]
+    end
+
+    test "returns an error from response streaming with custom on_response handler", %{
+      builder: builder
+    } do
+      @mock
+      |> expect(:stream, fn %Request{}, on_response, _opts ->
+        status = 404
+        headers = [{"content-length", 80_543}]
+        stream = Stream.cycle([])
+        on_response.({status, headers, stream})
+      end)
+
+      on_response = fn {status, headers, _body} ->
+        assert status == 404
+        assert List.keyfind(headers, "content-length", 0) == {"content-length", 80_543}
+
+        {:error, Error.new(code: :not_found, metadata: %{status: status})}
+      end
+
+      assert {:error, %Error{}} = Fetcher.stream(builder, on_response)
+    end
   end
 
   describe "dealing with errors" do