feat: source schema api (#2225)

* feat: source schema api

* chore: compilation warnings

lib/logflare/source_schemas.ex

@@ -5,6 +5,7 @@ defmodule Logflare.SourceSchemas do
 
   alias Logflare.Repo
   alias Logflare.SourceSchemas.SourceSchema
+  alias Logflare.Google.BigQuery.SchemaUtils
 
   require Logger
 
@@ -101,4 +102,61 @@ defmodule Logflare.SourceSchemas do
   def change_source_schema(%SourceSchema{} = source_schema, attrs \\ %{}) do
     SourceSchema.changeset(source_schema, attrs)
   end
+
+  def format_schema(bq_schema, variant, to_merge \\ %{})
+
+  def format_schema(%SourceSchema{bigquery_schema: bq_schema}, :dot, to_merge) do
+    bq_schema
+    |> SchemaUtils.bq_schema_to_flat_typemap()
+    |> Enum.filter(fn
+      {_k, :map} -> false
+      _ -> true
+    end)
+    |> Enum.map(fn
+      {k, {:list, type}} -> {k, "#{type}[]"}
+      {k, v} -> {k, Atom.to_string(v)}
+    end)
+    |> Map.new()
+    |> Map.merge(to_merge)
+  end
+
+  def format_schema(%SourceSchema{bigquery_schema: bq_schema}, :json_schema, to_merge) do
+    bq_schema
+    |> SchemaUtils.to_typemap()
+    |> typemap_to_json_schema()
+    |> Map.merge(to_merge)
+    |> Map.put(
+      "$schema",
+      "https://json-schema.org/draft/2020-12/schema"
+    )
+  end
+
+  defp typemap_to_json_schema(map) when is_map(map) do
+    properties = Enum.map(map, &typemap_to_json_schema/1) |> Map.new()
+
+    %{
+      "properties" => properties,
+      "type" => "object"
+    }
+  end
+
+  defp typemap_to_json_schema({key, %{fields: fields, t: :map}}) do
+    {Atom.to_string(key), typemap_to_json_schema(fields)}
+  end
+
+  defp typemap_to_json_schema({key, %{t: {:list, type}}}) do
+    {Atom.to_string(key), %{"type" => "array", "items" => %{"type" => Atom.to_string(type)}}}
+  end
+
+  defp typemap_to_json_schema({key, %{t: :datetime}}),
+    do: {Atom.to_string(key), %{"type" => "number"}}
+
+  defp typemap_to_json_schema({key, %{t: :integer}}),
+    do: {Atom.to_string(key), %{"type" => "number"}}
+
+  defp typemap_to_json_schema({key, %{t: :float}}),
+    do: {Atom.to_string(key), %{"type" => "number"}}
+
+  defp typemap_to_json_schema({key, %{t: type}}),
+    do: {Atom.to_string(key), %{"type" => Atom.to_string(type)}}
 end

lib/logflare_web/controllers/api/source_controller.ex

@@ -3,6 +3,7 @@ defmodule LogflareWeb.Api.SourceController do
   use OpenApiSpex.ControllerSpecs
 
   alias Logflare.Sources
+  alias Logflare.SourceSchemas
   alias Logflare.Backends
   alias LogflareWeb.OpenApi.Accepted
   alias LogflareWeb.OpenApi.Created
@@ -11,6 +12,7 @@ defmodule LogflareWeb.Api.SourceController do
   alias LogflareWeb.OpenApiSchemas.Event
 
   alias LogflareWeb.OpenApiSchemas.Source
+  alias LogflareWeb.OpenApiSchemas
 
   action_fallback(LogflareWeb.Api.FallbackController)
 
@@ -151,7 +153,7 @@ defmodule LogflareWeb.Api.SourceController do
     end
   end
 
-  operation(:removebackend,
+  operation(:remove_backend,
     summary: "Remove source backend",
     parameters: [
       source_token: [in: :path, description: "Source Token", type: :string],
@@ -173,4 +175,30 @@ defmodule LogflareWeb.Api.SourceController do
       |> json(source)
     end
   end
+
+  operation(:show_schema,
+    summary: "Show source schema",
+    parameters: [token: [in: :path, description: "Source Token", type: :string]],
+    responses: %{
+      200 => OpenApiSchemas.SourceSchema.response(),
+      404 => NotFound.response()
+    }
+  )
+
+  def show_schema(%{assigns: %{user: user}} = conn, %{"source_token" => token} = params) do
+    with source when not is_nil(source) <- Sources.get_by(token: token, user_id: user.id),
+         schema = SourceSchemas.get_source_schema_by(source_id: source.id) do
+      data =
+        if Map.get(params, "variant") == "dot" do
+          SourceSchemas.format_schema(schema, :dot)
+        else
+          SourceSchemas.format_schema(schema, :json_schema, %{
+            :title => source.name,
+            :"$id" => ~p"/sources/#{source.token}/schema"
+          })
+        end
+
+      json(conn, data)
+    end
+  end
 end

lib/logflare_web/open_api_schemas.ex

@@ -123,6 +123,12 @@ defmodule LogflareWeb.OpenApiSchemas do
     use LogflareWeb.OpenApi, properties: @properties, required: [:name]
   end
 
+  defmodule SourceSchema do
+    @properties %{}
+
+    use LogflareWeb.OpenApi, properties: @properties, required: []
+  end
+
   defmodule RuleApiSchema do
     @properties %{
       id: %Schema{type: :integer},

lib/logflare_web/router.ex

@@ -391,6 +391,7 @@ defmodule LogflareWeb.Router do
       param: "token",
       only: [:index, :show, :create, :update, :delete]
     ) do
+      get "/schema", Api.SourceController, :show_schema
       get "/recent", Api.SourceController, :recent
       post "/backends/:backend_token", Api.SourceController, :add_backend
       delete "/backends/:backend_token", Api.SourceController, :remove_backend

test/logflare/sources_schemas_test.exs

@@ -0,0 +1,66 @@
+defmodule Logflare.SourceSchemasTest do
+  @moduledoc false
+  use Logflare.DataCase
+
+  alias Logflare.SourceSchemas
+
+  describe "format_schema/3" do
+    setup do
+      insert(:plan, name: "Free")
+      user = insert(:user)
+      source = insert(:source, user: user)
+      %{user: user, source: source}
+    end
+
+    test "dot notation with nested values", %{
+      source: source
+    } do
+      schema =
+        insert(:source_schema,
+          source: source,
+          bigquery_schema:
+            TestUtils.build_bq_schema(%{
+              "test" => %{"nested" => 123, "listical" => ["testing", "123"]}
+            })
+        )
+
+      assert %{
+               "test.nested" => "integer",
+               "timestamp" => "datetime",
+               "test.listical" => "string[]"
+             } = params = SourceSchemas.format_schema(schema, :dot)
+
+      refute Map.get(params, "test")
+    end
+
+    test "json schema ", %{
+      source: source
+    } do
+      schema =
+        insert(:source_schema,
+          source: source,
+          bigquery_schema:
+            TestUtils.build_bq_schema(%{
+              "test" => %{"nested" => 123, "listical" => ["testing", "123"]}
+            })
+        )
+
+      assert %{
+               "properties" => %{
+                 "test" => %{
+                   "type" => "object",
+                   "properties" => %{
+                     "nested" => %{
+                       "type" => "number"
+                     },
+                     "listical" => %{
+                       "type" => "array",
+                       "items" => %{"type" => "string"}
+                     }
+                   }
+                 }
+               }
+             } = SourceSchemas.format_schema(schema, :json_schema)
+    end
+  end
+end

test/logflare_web/controllers/api/source_controller_test.exs

@@ -197,6 +197,61 @@ defmodule LogflareWeb.Api.SourceControllerTest do
     end
   end
 
+  describe "show_schema/2" do
+    test "GET schema with dot syntax", %{conn: conn, user: user, sources: [source | _]} do
+      insert(:source_schema,
+        source: source,
+        bigquery_schema:
+          TestUtils.build_bq_schema(%{
+            "test" => %{"nested" => 123}
+          })
+      )
+
+      conn =
+        conn
+        |> add_access_token(user, "private")
+        |> get("/api/sources/#{source.token}/schema?variant=dot")
+
+      # returns the source
+      assert %{
+               "id" => "string",
+               "event_message" => "string",
+               "timestamp" => "datetime",
+               "test.nested" => "integer"
+             } = json_response(conn, 200)
+    end
+
+    test "GET schema with json schema", %{conn: conn, user: user, sources: [source | _]} do
+      insert(:source_schema,
+        source: source,
+        bigquery_schema:
+          TestUtils.build_bq_schema(%{
+            "test" => %{"nested" => 123, "listical" => ["testing", "123"]}
+          })
+      )
+
+      %{name: source_name} = source
+
+      conn =
+        conn
+        |> add_access_token(user, "private")
+        |> get("/api/sources/#{source.token}/schema")
+
+      # returns the source
+      assert %{
+               "$schema" => _,
+               "$id" => _,
+               "title" => ^source_name,
+               "type" => "object",
+               "properties" => %{
+                 "id" => %{"type" => "string"},
+                 "event_message" => %{"type" => "string"},
+                 "timestamp" => %{"type" => "number"}
+               }
+             } = json_response(conn, 200)
+    end
+  end
+
   describe "add_backend/2" do
     test "attaches a backend", %{conn: conn, user: user, sources: [source | _]} do
       backend = insert(:backend, user: user)