feat: expose ingest sources through source index

docs/docs.logflare.com/docs/concepts/access-tokens/index.md

@@ -52,6 +52,12 @@ Access tokens can be exposed for client-side usage. Consider restricting tokens
 
 Private access tokens should not be exposed to the client side, as private access tokens have complete access to all management APIs used for account control.
 
+## Discovering ingest sources
+
+`GET /api/sources` can be used to discover sources available to an ingest credential. An account-wide `ingest` token, legacy API key, deprecated empty/public token, or source-scoped `ingest:source:<id>` (including deprecated `ingest:collection:<id>`) token returns an ordered JSON list containing only `token` and `name`. Source-scoped tokens return only owned, authorized sources.
+
+Private tokens retain the management response, which contains the complete source representation. Any credential authorized to list sources can request the minimal RFC 4180 CSV list with `GET /api/sources?format=csv`; its columns are `token,name`. Source lists are returned with `Cache-Control: no-store`.
+
 ## Rotation
 
 When rotating access tokens for client-side access tokens, we recommend creating another client-side token and performing a gradual shift to the new token before revoking the old token.

lib/logflare/sources.ex

@@ -47,6 +47,22 @@ defmodule Logflare.Sources do
     |> Enum.map(&put_retention_days/1)
   end
 
+  @spec list_source_tokens_by_user(pos_integer(), [pos_integer()] | nil) :: [map()]
+  def list_source_tokens_by_user(user_id, source_ids \\ nil) when is_integer(user_id) do
+    query =
+      from(s in Source,
+        where: s.user_id == ^user_id,
+        order_by: [asc: s.name, asc: s.id],
+        select: %{token: s.token, name: s.name}
+      )
+
+    case source_ids do
+      nil -> Repo.all(query)
+      [] -> []
+      source_ids -> Repo.all(where(query, [s], s.id in ^source_ids))
+    end
+  end
+
   @spec list_system_sources_by_user(User.t()) :: [Source.t()]
   def list_system_sources_by_user(%User{id: user_id}), do: list_system_sources_by_user(user_id)
 

lib/logflare_web/api/source_csv.ex

@@ -0,0 +1,22 @@
+defmodule LogflareWeb.Api.SourceCsv do
+  @moduledoc false
+
+  @spec encode([%{token: String.t() | atom(), name: String.t()}]) :: String.t()
+  def encode(sources) do
+    (["token,name"] ++ Enum.map(sources, &row/1))
+    |> Enum.join("\r\n")
+    |> Kernel.<>("\r\n")
+  end
+
+  defp row(%{token: token, name: name}), do: escape(token) <> "," <> escape(name)
+
+  defp escape(value) do
+    value = to_string(value)
+
+    if String.contains?(value, [",", "\"", "\r", "\n"]) do
+      "\"" <> String.replace(value, "\"", "\"\"") <> "\""
+    else
+      value
+    end
+  end
+end

lib/logflare_web/controllers/api/source_controller.ex

@@ -2,32 +2,139 @@ defmodule LogflareWeb.Api.SourceController do
   use LogflareWeb, :controller
   use OpenApiSpex.ControllerSpecs
 
+  alias Logflare.Backends
+  alias Logflare.Partners.Partner
   alias Logflare.Sources
   alias Logflare.SourceSchemas
-  alias Logflare.Backends
+  alias LogflareWeb.Api.FallbackController
+  alias LogflareWeb.Api.SourceCsv
   alias LogflareWeb.OpenApi.Accepted
+  alias LogflareWeb.OpenApi.BadRequest
   alias LogflareWeb.OpenApi.Created
   alias LogflareWeb.OpenApi.List
   alias LogflareWeb.OpenApi.NotFound
+  alias LogflareWeb.OpenApi.Unauthorized
   alias LogflareWeb.OpenApi.UnprocessableEntity
+  alias LogflareWeb.OpenApiSchemas
   alias LogflareWeb.OpenApiSchemas.Event
-
   alias LogflareWeb.OpenApiSchemas.Source
+  alias LogflareWeb.OpenApiSchemas.SourceIndexResponse
   alias LogflareWeb.OpenApiSchemas.SourceParams
-  alias LogflareWeb.OpenApiSchemas
+  alias OpenApiSpex.MediaType
+  alias OpenApiSpex.Response
+  alias OpenApiSpex.Schema
 
   action_fallback(LogflareWeb.Api.FallbackController)
 
+  @max_source_id 9_223_372_036_854_775_807
+
   tags(["management"])
 
   operation(:index,
     summary: "List sources",
-    responses: %{200 => List.response(Source)}
+    description:
+      "Private tokens receive the full management source representation by default. Ingest-compatible credentials receive only token and name. Set format=csv for a token,name CSV list.",
+    parameters: [
+      format: [
+        in: :query,
+        description: "Response format (csv)",
+        schema: %Schema{type: :string, enum: ["csv"]}
+      ]
+    ],
+    responses: %{
+      200 => %Response{
+        description: "Source list",
+        content: %{
+          "application/json" => %MediaType{schema: SourceIndexResponse},
+          "text/csv" => %MediaType{schema: %Schema{type: :string}}
+        }
+      },
+      400 => BadRequest.response(),
+      401 => Unauthorized.response()
+    }
   )
 
-  def index(%{assigns: %{user: user}} = conn, _) do
-    sources = Sources.list_sources_by_user(user.id) |> Sources.preload_for_dashboard()
-    json(conn, sources)
+  def index(%{assigns: %{user: user}} = conn, params) do
+    case {params["format"], source_access_for_conn(conn)} do
+      {_format, :unauthorized} ->
+        FallbackController.call(conn, {:error, :unauthorized})
+
+      {format, _access} when format not in [nil, "csv"] ->
+        FallbackController.call(conn, {:error, "Unsupported format"})
+
+      {"csv", access} ->
+        conn
+        |> put_resp_header("cache-control", "no-store")
+        |> put_resp_content_type("text/csv")
+        |> send_resp(
+          200,
+          SourceCsv.encode(Sources.list_source_tokens_by_user(user.id, source_ids(access)))
+        )
+
+      {nil, :private} ->
+        render_private_json(conn, user.id)
+
+      {nil, :partner} ->
+        render_private_json(conn, user.id)
+
+      {nil, {:ingest, source_ids}} ->
+        render_minimal_json(conn, user.id, source_ids)
+    end
+  end
+
+  defp render_private_json(conn, user_id) do
+    conn
+    |> put_resp_header("cache-control", "no-store")
+    |> json(Sources.list_sources_by_user(user_id) |> Sources.preload_for_dashboard())
+  end
+
+  defp render_minimal_json(conn, user_id, source_ids) do
+    conn
+    |> put_resp_header("cache-control", "no-store")
+    |> json(Sources.list_source_tokens_by_user(user_id, source_ids))
+  end
+
+  defp source_access_for_conn(%{assigns: %{partner: %Partner{}}}), do: :partner
+
+  defp source_access_for_conn(%{assigns: %{access_token: access_token}}),
+    do: source_access_for_token(access_token)
+
+  defp source_access_for_conn(_conn), do: {:ingest, nil}
+
+  defp source_ids(:private), do: nil
+  defp source_ids(:partner), do: nil
+  defp source_ids({:ingest, source_ids}), do: source_ids
+
+  defp source_access_for_token(nil), do: {:ingest, nil}
+
+  defp source_access_for_token(%{scopes: scopes}) do
+    scopes = String.split(scopes || "")
+
+    cond do
+      "private" in scopes -> :private
+      scopes == [] or "public" in scopes or "ingest" in scopes -> {:ingest, nil}
+      true -> scoped_source_access(scopes)
+    end
+  end
+
+  defp scoped_source_access(scopes) do
+    source_ids =
+      scopes
+      |> Enum.flat_map(fn
+        "ingest:source:" <> id -> parse_source_id(id)
+        "ingest:collection:" <> id -> parse_source_id(id)
+        _ -> []
+      end)
+      |> Enum.uniq()
+
+    if source_ids == [], do: :unauthorized, else: {:ingest, source_ids}
+  end
+
+  defp parse_source_id(id) do
+    case Integer.parse(id) do
+      {source_id, ""} when source_id > 0 and source_id <= @max_source_id -> [source_id]
+      _ -> []
+    end
   end
 
   operation(:show,

lib/logflare_web/open_api_schemas.ex

@@ -154,6 +154,24 @@ defmodule LogflareWeb.OpenApiSchemas do
     use LogflareWeb.OpenApi, properties: @properties, required: [:name]
   end
 
+  defmodule SourceListItem do
+    @properties %{
+      token: %Schema{type: :string},
+      name: %Schema{type: :string}
+    }
+
+    use LogflareWeb.OpenApi, properties: @properties, required: [:token, :name]
+  end
+
+  defmodule SourceIndexResponse do
+    require OpenApiSpex
+
+    OpenApiSpex.schema(%{
+      type: :array,
+      items: %Schema{anyOf: [Source, SourceListItem]}
+    })
+  end
+
   defmodule SourceParams do
     @properties %{
       name: %Schema{type: :string},

lib/logflare_web/router.ex

@@ -147,6 +147,14 @@ defmodule LogflareWeb.Router do
     plug(LogflareWeb.Plugs.VerifyApiAccess, scopes: ~w(private))
   end
 
+  pipeline :source_list_api do
+    plug(Plug.RequestId)
+    plug(LogflareWeb.Plugs.SetHeaders)
+    plug(OpenApiSpex.Plug.PutApiSpec, module: LogflareWeb.ApiSpec)
+    plug(LogflareWeb.Plugs.VerifyApiAccess, require_token: true)
+    plug(:accepts, ["json", "csv"])
+  end
+
   pipeline :require_auth do
     plug(LogflareWeb.Plugs.RequireAuth)
   end
@@ -447,6 +455,12 @@ defmodule LogflareWeb.Router do
     get("/", HealthCheckController, :ready)
   end
 
+  scope "/api", LogflareWeb do
+    pipe_through(:source_list_api)
+
+    get("/sources", Api.SourceController, :index)
+  end
+
   # Account management API.
   scope "/api", LogflareWeb do
     pipe_through([:api, :require_mgmt_api_auth])
@@ -462,7 +476,7 @@ defmodule LogflareWeb.Router do
 
     resources("/sources", Api.SourceController,
       param: "token",
-      only: [:index, :show, :create, :update, :delete]
+      only: [:show, :create, :update, :delete]
     ) do
       get "/schema", Api.SourceController, :show_schema
       get "/recent", Api.SourceController, :recent