diff --git a/docs/docs.json b/docs/docs.json
index 2a3ae212c..0b4f359ed 100644
--- a/docs/docs.json
+++ b/docs/docs.json
@@ -190,6 +190,16 @@
"management-api/http-endpoints/delete"
]
},
+ {
+ "group": "Functions",
+ "pages": [
+ "management-api/functions/list",
+ "management-api/functions/get",
+ "management-api/functions/create",
+ "management-api/functions/update",
+ "management-api/functions/delete"
+ ]
+ },
{
"group": "Sink consumers",
"pages": [
diff --git a/docs/management-api/functions/create.mdx b/docs/management-api/functions/create.mdx
new file mode 100644
index 000000000..8b21e327a
--- /dev/null
+++ b/docs/management-api/functions/create.mdx
@@ -0,0 +1,165 @@
+---
+title: "Create Function"
+description: "Create a new function"
+---
+
+## Request
+
+
+```bash cURL
+curl -X POST "https://api.sequinstream.com/api/functions" \
+ -H "Authorization: Bearer YOUR_API_TOKEN" \
+ -H "Content-Type: application/json" \
+ -d '{
+ "name": "my-filter",
+ "description": "Filter records with value greater than 40",
+ "type": "filter",
+ "code": "def filter(action, record, changes, metadata) do\n record[\"value\"] > 40\nend"
+ }'
+```
+
+```javascript JavaScript
+const response = await fetch('https://api.sequinstream.com/api/functions', {
+ method: 'POST',
+ headers: {
+ 'Authorization': 'Bearer YOUR_API_TOKEN',
+ 'Content-Type': 'application/json'
+ },
+ body: JSON.stringify({
+ name: 'my-filter',
+ description: 'Filter records with value greater than 40',
+ type: 'filter',
+ code: `def filter(action, record, changes, metadata) do
+ record["value"] > 40
+end`
+ })
+});
+const function = await response.json();
+```
+
+
+## Parameters
+
+
+ Unique name for the function
+
+
+
+ Optional description of what the function does
+
+
+
+ Type of function: `filter`, `transform`, `enrichment`, `path`, or `routing`
+
+
+
+ Function code (required for `filter`, `transform`, `enrichment`, and `routing` types)
+
+
+
+ Path to extract (required for `path` type)
+
+
+
+ Sink type (required for `routing` type): `http_push`, `sqs`, `kafka`, etc.
+
+
+## Function Types
+
+### Filter Function
+
+```json
+{
+ "name": "my-filter",
+ "description": "Filter VIP customers",
+ "type": "filter",
+ "code": "def filter(action, record, changes, metadata) do\n record[\"customer_type\"] == \"VIP\"\nend"
+}
+```
+
+### Transform Function
+
+```json
+{
+ "name": "my-transform",
+ "description": "Extract ID and action",
+ "type": "transform",
+ "code": "def transform(action, record, changes, metadata) do\n %{id: record[\"id\"], action: action}\nend"
+}
+```
+
+### Path Function
+
+```json
+{
+ "name": "my-path",
+ "description": "Extract record",
+ "type": "path",
+ "path": "record"
+}
+```
+
+### Routing Function
+
+```json
+{
+ "name": "my-routing",
+ "description": "Route to REST API",
+ "type": "routing",
+ "sink_type": "http_push",
+ "code": "def route(action, record, changes, metadata) do\n %{\n method: \"POST\",\n endpoint_path: \"/api/users/#{record[\"id\"]}\"\n }\nend"
+}
+```
+
+### Enrichment Function
+
+```json
+{
+ "name": "my-enrichment",
+ "description": "Enrich with customer data",
+ "type": "enrichment",
+ "code": "SELECT\n u.id,\n a.name as account_name\nFROM\n users u\nJOIN\n accounts a on u.account_id = a.id\nWHERE\n u.id = ANY($1)"
+}
+```
+
+## Response
+
+
+ Unique identifier for the function
+
+
+
+ Name of the function
+
+
+
+ Description of the function
+
+
+
+ Type of the function
+
+
+
+ Function code (for applicable types)
+
+
+
+ Path (for path type functions)
+
+
+
+ Sink type (for routing type functions)
+
+
+## Example Response
+
+```json
+{
+ "id": "550e8400-e29b-41d4-a716-446655440000",
+ "name": "my-filter",
+ "description": "Filter records with value greater than 40",
+ "type": "filter",
+ "code": "def filter(action, record, changes, metadata) do\n record[\"value\"] > 40\nend"
+}
+```
diff --git a/docs/management-api/functions/delete.mdx b/docs/management-api/functions/delete.mdx
new file mode 100644
index 000000000..509ae992e
--- /dev/null
+++ b/docs/management-api/functions/delete.mdx
@@ -0,0 +1,63 @@
+---
+title: "Delete Function"
+description: "Delete a function by ID or name"
+---
+
+## Request
+
+You can delete a function by either its ID or name.
+
+
+```bash cURL (by ID)
+curl -X DELETE "https://api.sequinstream.com/api/functions/550e8400-e29b-41d4-a716-446655440000" \
+ -H "Authorization: Bearer YOUR_API_TOKEN"
+```
+
+```bash cURL (by name)
+curl -X DELETE "https://api.sequinstream.com/api/functions/my-filter" \
+ -H "Authorization: Bearer YOUR_API_TOKEN"
+```
+
+```javascript JavaScript
+const response = await fetch('https://api.sequinstream.com/api/functions/my-filter', {
+ method: 'DELETE',
+ headers: {
+ 'Authorization': 'Bearer YOUR_API_TOKEN'
+ }
+});
+const result = await response.json();
+```
+
+
+## Path Parameters
+
+
+ Function ID (UUID) or name
+
+
+## Response
+
+Returns the deleted function's ID and a confirmation flag.
+
+
+ ID of the deleted function
+
+
+
+ Always `true` for successful deletions
+
+
+## Example Response
+
+```json
+{
+ "id": "550e8400-e29b-41d4-a716-446655440000",
+ "deleted": true
+}
+```
+
+## Notes
+
+
+ Deleting a function that is referenced by a sink will fail with a foreign key constraint error. Remove the function from any sinks before deleting it.
+
diff --git a/docs/management-api/functions/get.mdx b/docs/management-api/functions/get.mdx
new file mode 100644
index 000000000..78f931725
--- /dev/null
+++ b/docs/management-api/functions/get.mdx
@@ -0,0 +1,51 @@
+---
+title: "Get Function"
+description: "Get a function by ID or name"
+---
+
+## Request
+
+You can retrieve a function by either its ID or name.
+
+
+```bash cURL (by ID)
+curl -X GET "https://api.sequinstream.com/api/functions/550e8400-e29b-41d4-a716-446655440000" \
+ -H "Authorization: Bearer YOUR_API_TOKEN"
+```
+
+```bash cURL (by name)
+curl -X GET "https://api.sequinstream.com/api/functions/my-filter" \
+ -H "Authorization: Bearer YOUR_API_TOKEN"
+```
+
+```javascript JavaScript
+const response = await fetch('https://api.sequinstream.com/api/functions/my-filter', {
+ headers: {
+ 'Authorization': 'Bearer YOUR_API_TOKEN'
+ }
+});
+const function = await response.json();
+```
+
+
+## Path Parameters
+
+
+ Function ID (UUID) or name
+
+
+## Response
+
+Returns a function object.
+
+## Example Response
+
+```json
+{
+ "id": "550e8400-e29b-41d4-a716-446655440000",
+ "name": "my-filter",
+ "description": "Filter records with value greater than 40",
+ "type": "filter",
+ "code": "def filter(action, record, changes, metadata) do\n record[\"value\"] > 40\nend"
+}
+```
diff --git a/docs/management-api/functions/list.mdx b/docs/management-api/functions/list.mdx
new file mode 100644
index 000000000..afd9b6219
--- /dev/null
+++ b/docs/management-api/functions/list.mdx
@@ -0,0 +1,53 @@
+---
+title: "List Functions"
+description: "List all functions for your account"
+---
+
+## Request
+
+
+```bash cURL
+curl -X GET "https://api.sequinstream.com/api/functions" \
+ -H "Authorization: Bearer YOUR_API_TOKEN"
+```
+
+```javascript JavaScript
+const response = await fetch('https://api.sequinstream.com/api/functions', {
+ headers: {
+ 'Authorization': 'Bearer YOUR_API_TOKEN'
+ }
+});
+const { data } = await response.json();
+```
+
+
+## Response
+
+Returns an array of function objects.
+
+
+ Array of function objects
+
+
+## Example Response
+
+```json
+{
+ "data": [
+ {
+ "id": "550e8400-e29b-41d4-a716-446655440000",
+ "name": "my-filter",
+ "description": "Filter records with value greater than 40",
+ "type": "filter",
+ "code": "def filter(action, record, changes, metadata) do\n record[\"value\"] > 40\nend"
+ },
+ {
+ "id": "660e8400-e29b-41d4-a716-446655440001",
+ "name": "my-transform",
+ "description": "Extract ID and action",
+ "type": "transform",
+ "code": "def transform(action, record, changes, metadata) do\n %{id: record[\"id\"], action: action}\nend"
+ }
+ ]
+}
+```
diff --git a/docs/management-api/functions/update.mdx b/docs/management-api/functions/update.mdx
new file mode 100644
index 000000000..ea3de19ec
--- /dev/null
+++ b/docs/management-api/functions/update.mdx
@@ -0,0 +1,92 @@
+---
+title: "Update Function"
+description: "Update an existing function"
+---
+
+## Request
+
+You can update a function by either its ID or name.
+
+
+```bash cURL (by ID)
+curl -X PUT "https://api.sequinstream.com/api/functions/550e8400-e29b-41d4-a716-446655440000" \
+ -H "Authorization: Bearer YOUR_API_TOKEN" \
+ -H "Content-Type: application/json" \
+ -d '{
+ "description": "Updated description",
+ "code": "def filter(action, record, changes, metadata) do\n record[\"value\"] > 50\nend"
+ }'
+```
+
+```bash cURL (by name)
+curl -X PUT "https://api.sequinstream.com/api/functions/my-filter" \
+ -H "Authorization: Bearer YOUR_API_TOKEN" \
+ -H "Content-Type: application/json" \
+ -d '{
+ "description": "Updated description"
+ }'
+```
+
+```javascript JavaScript
+const response = await fetch('https://api.sequinstream.com/api/functions/my-filter', {
+ method: 'PUT',
+ headers: {
+ 'Authorization': 'Bearer YOUR_API_TOKEN',
+ 'Content-Type': 'application/json'
+ },
+ body: JSON.stringify({
+ description: 'Updated description',
+ code: `def filter(action, record, changes, metadata) do
+ record["value"] > 50
+end`
+ })
+});
+const function = await response.json();
+```
+
+
+## Path Parameters
+
+
+ Function ID (UUID) or name
+
+
+## Body Parameters
+
+All parameters are optional. Only include the fields you want to update.
+
+
+ Update the function name
+
+
+
+ Update the description
+
+
+
+ Update the function code (for applicable types)
+
+
+
+ Update the path (for path type functions)
+
+
+
+ Update the sink type (for routing type functions)
+
+
+## Response
+
+Returns the updated function object.
+
+## Example Response
+
+```json
+{
+ "id": "550e8400-e29b-41d4-a716-446655440000",
+ "name": "my-filter",
+ "description": "Updated description",
+ "type": "filter",
+ "code": "def filter(action, record, changes, metadata) do\n record[\"value\"] > 50\nend"
+}
+```
diff --git a/lib/sequin/consumers/consumers.ex b/lib/sequin/consumers/consumers.ex
index 58d78522d..dd98a71e8 100644
--- a/lib/sequin/consumers/consumers.ex
+++ b/lib/sequin/consumers/consumers.ex
@@ -220,22 +220,31 @@ defmodule Sequin.Consumers do
end)
end
- def update_function(account_id, id, params) do
+ def update_function(function_or_account_id, params_or_id, opts_or_params \\ [])
+
+ def update_function(%Function{} = function, params, opts) when is_map(params) do
Repo.transact(fn ->
- %Function{id: id, account_id: account_id}
- |> Function.update_changeset(params)
- |> Repo.update()
- |> case do
- {:ok, function} ->
- ConsumerLifecycleEventWorker.enqueue(:update, :function, function.id)
- {:ok, function}
+ res =
+ function
+ |> Function.update_changeset(params)
+ |> Repo.update()
- {:error, error} ->
- {:error, error}
+ with {:ok, updated_function} <- res do
+ if !opts[:skip_lifecycle] do
+ ConsumerLifecycleEventWorker.enqueue(:update, :function, updated_function.id)
+ end
+
+ {:ok, updated_function}
end
end)
end
+ def update_function(account_id, id, params) when is_binary(account_id) and is_binary(id) do
+ with {:ok, function} <- get_function_for_account(account_id, id) do
+ update_function(function, params, [])
+ end
+ end
+
def delete_function(account_id, id) do
with {:ok, function} <- get_function_for_account(account_id, id) do
function
diff --git a/lib/sequin/consumers/function.ex b/lib/sequin/consumers/function.ex
index c5f87fd39..a9c9072e4 100644
--- a/lib/sequin/consumers/function.ex
+++ b/lib/sequin/consumers/function.ex
@@ -41,7 +41,7 @@ defmodule Sequin.Consumers.Function do
function
|> cast(attrs, [:name, :description])
|> changeset(attrs)
- |> unique_constraint([:account_id, :name], error_key: :name)
+ |> unique_constraint([:account_id, :name], name: :transforms_account_id_name_index, error_key: :name)
end
def update_changeset(function, attrs) do
diff --git a/lib/sequin/transforms/transforms.ex b/lib/sequin/transforms/transforms.ex
index e2a1f7cfa..9c4a4bcbe 100644
--- a/lib/sequin/transforms/transforms.ex
+++ b/lib/sequin/transforms/transforms.ex
@@ -423,6 +423,7 @@ defmodule Sequin.Transforms do
def to_external(%Function{function: %PathFunction{}} = function, _show_sensitive) do
%{
+ id: function.id,
name: function.name,
description: function.description,
type: function.type,
@@ -432,6 +433,7 @@ defmodule Sequin.Transforms do
def to_external(%Function{function: %TransformFunction{}} = function, _show_sensitive) do
%{
+ id: function.id,
name: function.name,
description: function.description,
type: function.type,
@@ -441,6 +443,7 @@ defmodule Sequin.Transforms do
def to_external(%Function{function: %RoutingFunction{}} = function, _show_sensitive) do
%{
+ id: function.id,
name: function.name,
description: function.description,
type: function.type,
@@ -451,6 +454,7 @@ defmodule Sequin.Transforms do
def to_external(%Function{function: %EnrichmentFunction{}} = function, _show_sensitive) do
%{
+ id: function.id,
name: function.name,
description: function.description,
type: function.type,
@@ -460,6 +464,7 @@ defmodule Sequin.Transforms do
def to_external(%Function{function: %FilterFunction{}} = function, _show_sensitive) do
%{
+ id: function.id,
name: function.name,
description: function.description,
type: function.type,
@@ -785,6 +790,68 @@ defmodule Sequin.Transforms do
end)
end
+ def from_external_function(attrs) do
+ # Support both flat structure and nested structure
+ coerce_function_attrs(attrs)
+ end
+
+ # Helper function to coerce function attributes from external format to internal format
+ # Supports both flat and nested structures
+ defp coerce_function_attrs(%{"function" => _, "transform" => _}) do
+ {:error, Error.validation(summary: "Cannot specify both `function` and `transform`")}
+ end
+
+ defp coerce_function_attrs(%{"transform" => function} = raw_attrs) do
+ attrs =
+ raw_attrs
+ |> Map.delete("transform")
+ |> Map.put("function", coerce_function_sink_type(function))
+ |> update_in(["function", "type"], &coerce_type_to_transform/1)
+
+ {:ok, attrs}
+ end
+
+ defp coerce_function_attrs(%{"function" => _} = attrs) do
+ {:ok, Map.update!(attrs, "function", &coerce_function_sink_type/1)}
+ end
+
+ # Assume that if you don't have "function" or "transform" that you used flat structure
+ defp coerce_function_attrs(flat) when is_map(flat) do
+ # Check if this is a flat function definition or just top-level updates
+ has_function_fields = Enum.any?(["type", "sink_type", "code", "path"], &Map.has_key?(flat, &1))
+
+ if has_function_fields do
+ # This is a flat function definition, create nested structure
+ inner =
+ flat
+ |> Map.take(["type", "sink_type", "code", "description", "path"])
+ |> coerce_function_sink_type()
+ |> Map.update("type", nil, &coerce_type_to_transform/1)
+
+ nested_attrs =
+ flat
+ |> Map.take(["id", "name"])
+ |> Map.put("function", inner)
+
+ {:ok, nested_attrs}
+ else
+ # This is just top-level updates (name, description), pass through
+ {:ok, Map.take(flat, ["id", "name", "description"])}
+ end
+ end
+
+ defp coerce_function_attrs(_), do: {:error, Error.validation(summary: "Invalid function attributes")}
+
+ # Helper function to coerce "function" type to "transform" for backwards compatibility
+ defp coerce_type_to_transform("function"), do: "transform"
+ defp coerce_type_to_transform(type), do: type
+
+ defp coerce_function_sink_type(%{"sink_type" => "webhook"} = attrs) do
+ Map.put(attrs, "sink_type", "http_push")
+ end
+
+ defp coerce_function_sink_type(attrs), do: attrs
+
# Helper functions
defp parse_headers(nil), do: {:ok, %{}}
diff --git a/lib/sequin_web/controllers/function_controller.ex b/lib/sequin_web/controllers/function_controller.ex
new file mode 100644
index 000000000..34f48dfa7
--- /dev/null
+++ b/lib/sequin_web/controllers/function_controller.ex
@@ -0,0 +1,60 @@
+defmodule SequinWeb.FunctionController do
+ use SequinWeb, :controller
+
+ alias Sequin.Consumers
+ alias Sequin.Transforms
+ alias SequinWeb.ApiFallbackPlug
+
+ action_fallback ApiFallbackPlug
+
+ def index(conn, _params) do
+ account_id = conn.assigns.account_id
+
+ render(conn, "index.json", functions: Consumers.list_functions_for_account(account_id))
+ end
+
+ def show(conn, %{"id_or_name" => id_or_name}) do
+ account_id = conn.assigns.account_id
+
+ with {:ok, function} <- find_function_for_account(account_id, id_or_name) do
+ render(conn, "show.json", function: function)
+ end
+ end
+
+ def create(conn, params) do
+ account_id = conn.assigns.account_id
+
+ with {:ok, cleaned_params} <- Transforms.from_external_function(params),
+ {:ok, function} <- Consumers.create_function(account_id, cleaned_params) do
+ render(conn, "show.json", function: function)
+ end
+ end
+
+ def update(conn, %{"id_or_name" => id_or_name} = params) do
+ params = Map.delete(params, "id_or_name")
+ account_id = conn.assigns.account_id
+
+ with {:ok, existing_function} <- find_function_for_account(account_id, id_or_name),
+ {:ok, cleaned_params} <- Transforms.from_external_function(params),
+ {:ok, updated_function} <- Consumers.update_function(existing_function, cleaned_params) do
+ render(conn, "show.json", function: updated_function)
+ end
+ end
+
+ def delete(conn, %{"id_or_name" => id_or_name}) do
+ account_id = conn.assigns.account_id
+
+ with {:ok, function} <- find_function_for_account(account_id, id_or_name),
+ {:ok, _function} <- Consumers.delete_function(account_id, function.id) do
+ render(conn, "delete.json", function: function)
+ end
+ end
+
+ defp find_function_for_account(account_id, id_or_name) do
+ if Sequin.String.uuid?(id_or_name) do
+ Consumers.get_function_for_account(account_id, id_or_name)
+ else
+ Consumers.find_function(account_id, name: id_or_name)
+ end
+ end
+end
diff --git a/lib/sequin_web/controllers/function_json.ex b/lib/sequin_web/controllers/function_json.ex
new file mode 100644
index 000000000..2c9f7c821
--- /dev/null
+++ b/lib/sequin_web/controllers/function_json.ex
@@ -0,0 +1,22 @@
+defmodule SequinWeb.FunctionJSON do
+ @doc """
+ Renders a list of functions.
+ """
+ def index(%{functions: functions}) do
+ %{data: for(function <- functions, do: Sequin.Transforms.to_external(function))}
+ end
+
+ @doc """
+ Renders a single function.
+ """
+ def show(%{function: function}) do
+ Sequin.Transforms.to_external(function)
+ end
+
+ @doc """
+ Renders a deleted function.
+ """
+ def delete(%{function: function}) do
+ %{id: function.id, deleted: true}
+ end
+end
diff --git a/lib/sequin_web/router.ex b/lib/sequin_web/router.ex
index a3e0330e9..f5c41e946 100644
--- a/lib/sequin_web/router.ex
+++ b/lib/sequin_web/router.ex
@@ -166,6 +166,9 @@ defmodule SequinWeb.Router do
# HTTP Endpoints routes
resources("/destinations/http_endpoints", HttpEndpointController, except: [:new, :edit], param: "id_or_name")
+ # Function routes
+ resources("/functions", FunctionController, except: [:new, :edit], param: "id_or_name")
+
# Sink Consumer routes
resources("/sinks", SinkConsumerController, except: [:new, :edit], param: "id_or_name")
# Backfill routes
diff --git a/test/sequin_web/controllers/function_controller_test.exs b/test/sequin_web/controllers/function_controller_test.exs
new file mode 100644
index 000000000..e94e8c4b0
--- /dev/null
+++ b/test/sequin_web/controllers/function_controller_test.exs
@@ -0,0 +1,269 @@
+defmodule SequinWeb.FunctionControllerTest do
+ use SequinWeb.ConnCase, async: true
+
+ alias Sequin.Consumers
+ alias Sequin.Factory.AccountsFactory
+ alias Sequin.Factory.FunctionsFactory
+
+ setup :authenticated_conn
+
+ setup %{account: account} do
+ other_account = AccountsFactory.insert_account!()
+ function = FunctionsFactory.insert_function!(account_id: account.id)
+
+ other_function =
+ FunctionsFactory.insert_function!(account_id: other_account.id)
+
+ %{function: function, other_function: other_function, other_account: other_account}
+ end
+
+ describe "index" do
+ test "lists functions in the given account", %{
+ conn: conn,
+ account: account,
+ function: function
+ } do
+ another_function = FunctionsFactory.insert_function!(account_id: account.id)
+
+ conn = get(conn, ~p"/api/functions")
+ assert %{"data" => functions} = json_response(conn, 200)
+ assert length(functions) == 2
+ atomized_functions = Enum.map(functions, &Sequin.Map.atomize_keys/1)
+ assert_lists_equal([function, another_function], atomized_functions, &(&1.id == &2.id))
+ end
+
+ test "does not list functions from another account", %{
+ conn: conn,
+ other_function: other_function
+ } do
+ conn = get(conn, ~p"/api/functions")
+ assert %{"data" => functions} = json_response(conn, 200)
+ refute Enum.any?(functions, &(&1["id"] == other_function.id))
+ end
+ end
+
+ describe "show" do
+ test "shows function details by id", %{conn: conn, function: function} do
+ conn = get(conn, ~p"/api/functions/#{function.id}")
+ assert function_json = json_response(conn, 200)
+
+ assert function.name == function_json["name"]
+ assert function.id == function_json["id"]
+ end
+
+ test "shows function details by name", %{conn: conn, function: function} do
+ conn = get(conn, ~p"/api/functions/#{function.name}")
+ assert function_json = json_response(conn, 200)
+
+ assert function.name == function_json["name"]
+ assert function.id == function_json["id"]
+ end
+
+ test "returns 404 if function belongs to another account", %{
+ conn: conn,
+ other_function: other_function
+ } do
+ conn = get(conn, ~p"/api/functions/#{other_function.id}")
+ assert json_response(conn, 404)
+ end
+ end
+
+ describe "create" do
+ test "creates a filter function under the authenticated account", %{
+ conn: conn,
+ account: account
+ } do
+ function_attrs = %{
+ name: "my-filter",
+ description: "Filter records with value > 40",
+ type: "filter",
+ code: """
+ def filter(action, record, changes, metadata) do
+ record["value"] > 40
+ end
+ """
+ }
+
+ conn = post(conn, ~p"/api/functions", function_attrs)
+ assert %{"name" => name, "id" => id} = json_response(conn, 200)
+
+ {:ok, function} = Consumers.get_function_for_account(account.id, id)
+ assert function.account_id == account.id
+ assert function.name == name
+ assert function.type == "filter"
+ end
+
+ test "creates a transform function with nested structure", %{
+ conn: conn,
+ account: account
+ } do
+ function_attrs = %{
+ name: "my-transform",
+ description: "Extract ID and action",
+ function: %{
+ type: "transform",
+ code: """
+ def transform(action, record, changes, metadata) do
+ %{id: record["id"], action: action}
+ end
+ """
+ }
+ }
+
+ conn = post(conn, ~p"/api/functions", function_attrs)
+ assert %{"name" => name, "id" => id} = json_response(conn, 200)
+
+ {:ok, function} = Consumers.get_function_for_account(account.id, id)
+ assert function.account_id == account.id
+ assert function.name == name
+ assert function.type == "transform"
+ end
+
+ test "creates a path function", %{
+ conn: conn,
+ account: account
+ } do
+ function_attrs = %{
+ name: "my-path",
+ description: "Extract record",
+ type: "path",
+ path: "record"
+ }
+
+ conn = post(conn, ~p"/api/functions", function_attrs)
+ assert %{"name" => name, "id" => id} = json_response(conn, 200)
+
+ {:ok, function} = Consumers.get_function_for_account(account.id, id)
+ assert function.account_id == account.id
+ assert function.name == name
+ assert function.type == "path"
+ end
+
+ test "creates a routing function", %{
+ conn: conn,
+ account: account
+ } do
+ function_attrs = %{
+ name: "my-routing",
+ description: "Route to REST API",
+ type: "routing",
+ sink_type: "http_push",
+ code: """
+ def route(action, record, changes, metadata) do
+ %{
+ method: "POST",
+ endpoint_path: "/api/users/\#{record["id"]}"
+ }
+ end
+ """
+ }
+
+ conn = post(conn, ~p"/api/functions", function_attrs)
+ assert %{"name" => name, "id" => id} = json_response(conn, 200)
+
+ {:ok, function} = Consumers.get_function_for_account(account.id, id)
+ assert function.account_id == account.id
+ assert function.name == name
+ assert function.type == "routing"
+ end
+
+ test "creating a function with duplicate name fails", %{conn: conn, account: account} do
+ FunctionsFactory.insert_function!(account_id: account.id, name: "duplicate-name")
+
+ conn =
+ post(conn, ~p"/api/functions", %{
+ name: "duplicate-name",
+ type: "filter",
+ code: "def filter(action, record, changes, metadata), do: true"
+ })
+
+ assert json_response(conn, 422)["errors"] != %{}
+ end
+
+ test "returns validation error for invalid attributes", %{conn: conn} do
+ invalid_attrs = %{name: nil}
+ conn = post(conn, ~p"/api/functions", invalid_attrs)
+ assert json_response(conn, 422)["errors"] != %{}
+ end
+
+ test "ignores provided account_id and uses authenticated account", %{
+ conn: conn,
+ account: account,
+ other_account: other_account
+ } do
+ function_attrs = %{
+ name: "my-function",
+ type: "filter",
+ code: "def filter(action, record, changes, metadata), do: true",
+ account_id: other_account.id
+ }
+
+ conn = post(conn, ~p"/api/functions", function_attrs)
+
+ assert %{"id" => id} = json_response(conn, 200)
+
+ {:ok, function} = Consumers.get_function_for_account(account.id, id)
+ assert function.account_id == account.id
+ assert function.account_id != other_account.id
+ end
+ end
+
+ describe "update" do
+ test "updates the function with valid attributes by id", %{conn: conn, function: function} do
+ update_attrs = %{description: "Updated description"}
+ conn = put(conn, ~p"/api/functions/#{function.id}", update_attrs)
+ assert %{"id" => id} = json_response(conn, 200)
+
+ {:ok, updated_function} = Consumers.get_function_for_account(function.account_id, id)
+ assert updated_function.description == "Updated description"
+ assert updated_function.name == function.name
+ end
+
+ test "updates the function with valid attributes by name", %{conn: conn, function: function} do
+ update_attrs = %{description: "Updated description via name"}
+ conn = put(conn, ~p"/api/functions/#{function.name}", update_attrs)
+ assert %{"id" => id} = json_response(conn, 200)
+
+ {:ok, updated_function} = Consumers.get_function_for_account(function.account_id, id)
+ assert updated_function.description == "Updated description via name"
+ end
+
+ test "returns validation error for invalid attributes", %{conn: conn, function: function} do
+ invalid_attrs = %{name: ""}
+ conn = put(conn, ~p"/api/functions/#{function.id}", invalid_attrs)
+ assert json_response(conn, 422)["errors"] != %{}
+ end
+
+ test "returns 404 if function belongs to another account", %{
+ conn: conn,
+ other_function: other_function
+ } do
+ conn = put(conn, ~p"/api/functions/#{other_function.id}", %{description: "New description"})
+ assert json_response(conn, 404)
+ end
+ end
+
+ describe "delete" do
+ test "deletes the function by id", %{conn: conn, function: function} do
+ conn = delete(conn, ~p"/api/functions/#{function.id}")
+ assert %{"id" => id, "deleted" => true} = json_response(conn, 200)
+
+ assert {:error, _} = Consumers.get_function_for_account(function.account_id, id)
+ end
+
+ test "deletes the function by name", %{conn: conn, function: function} do
+ conn = delete(conn, ~p"/api/functions/#{function.name}")
+ assert %{"id" => _id, "deleted" => true} = json_response(conn, 200)
+
+ assert {:error, _} = Consumers.find_function(function.account_id, name: function.name)
+ end
+
+ test "returns 404 if function belongs to another account", %{
+ conn: conn,
+ other_function: other_function
+ } do
+ conn = delete(conn, ~p"/api/functions/#{other_function.id}")
+ assert json_response(conn, 404)
+ end
+ end
+end