# SPDX-FileCopyrightText: 2025 ash_neo4j contributors # # SPDX-License-Identifier: MIT defmodule AshNeo4j.DataLayer do @moduledoc "Ash DataLayer for Neo4j" @behaviour Ash.DataLayer require Logger alias AshNeo4j.Resource.Info, as: ResourceInfo alias AshNeo4j.ResourceMapping alias AshNeo4j.EdgeDescriptor alias AshNeo4j.Neo4jHelper alias AshNeo4j.QueryHelper alias AshNeo4j.Cypher alias AshNeo4j.Cypher.Query, as: CypherQuery alias AshNeo4j.DataLayer.Cast alias AshNeo4j.DataLayer.Dump alias AshNeo4j.DataLayer.TypeClassifier alias AshNeo4j.Util @impl true def can?(_, :read), do: true def can?(_, :create), do: true def can?(_, :composite_primary_key), do: true def can?(_, :update), do: true def can?(_, :upsert), do: true def can?(_, :destroy), do: true def can?(_, :sort), do: true def can?(_, :filter), do: true def can?(_, :limit), do: true def can?(_, :offset), do: true def can?(_, :boolean_filter), do: true def can?(_, {:sort, _}), do: true def can?(_, {:join, _}), do: true def can?(_, {:lateral_join, _}), do: true def can?(_, {:filter_relationship, _}), do: true def can?(_, :nested_expressions), do: true # Operators with actual Cypher equivalents in convert_operator/1 def can?(_, {:filter_expr, %Ash.Query.Operator.Eq{}}), do: true def can?(_, {:filter_expr, %Ash.Query.Operator.NotEq{}}), do: true def can?(_, {:filter_expr, %Ash.Query.Operator.In{}}), do: true def can?(_, {:filter_expr, %Ash.Query.Operator.LessThanOrEqual{}}), do: true def can?(_, {:filter_expr, %Ash.Query.Operator.LessThan{}}), do: true def can?(_, {:filter_expr, %Ash.Query.Operator.GreaterThan{}}), do: true def can?(_, {:filter_expr, %Ash.Query.Operator.GreaterThanOrEqual{}}), do: true def can?(_, {:filter_expr, %Ash.Query.Operator.IsNil{}}), do: true # contains — handled in predicates/2 via the %{name: :contains} branch def can?(_, {:filter_expr, %Ash.Query.Function.Contains{}}), do: true # spatial — st_contains(box, point) → point.withinBBox pushdown def can?(_, {:filter_expr, %AshNeo4j.Functions.StContains{}}), do: true # spatial — st_distance(point, point) → point.distance pushdown (in comparisons) def can?(_, {:filter_expr, %AshNeo4j.Functions.StDistance{}}), do: true # spatial — st_dwithin(point, point, ^km) → point.distance <= ^km pushdown def can?(_, {:filter_expr, %AshNeo4j.Functions.StDwithin{}}), do: true # spatial parity (in-memory eval) — st_within, st_intersects def can?(_, {:filter_expr, %AshNeo4j.Functions.StWithin{}}), do: true def can?(_, {:filter_expr, %AshNeo4j.Functions.StIntersects{}}), do: true # spatial — st_distance_in_meters: alias for st_distance, same pushdown def can?(_, {:filter_expr, %AshNeo4j.Functions.StDistanceInMeters{}}), do: true # All other filter expressions are accepted so Ash can hydrate and then evaluate # them in-memory via filter_stream / RuntimeExpression. Cypher builder falls # back to TRUE for unrecognised predicates; filter_stream corrects the results. def can?(_, {:filter_expr, _}), do: true def can?(_, :transact), do: true def can?(_, :expression_calculation), do: true def can?(_, {:aggregate, kind}) when kind in [:count, :exists, :sum, :avg, :min, :max, :list, :first], do: true def can?(_, {:query_aggregate, kind}) when kind in [:count, :exists, :sum, :avg, :min, :max, :list, :first], do: true def can?(_, {:aggregate_relationship, _}), do: true # combination queries (#10) def can?(_, :combine), do: true def can?(_, {:combine, :base}), do: true def can?(_, {:combine, :union}), do: true def can?(_, {:combine, :union_all}), do: true def can?(_, {:combine, :intersect}), do: true def can?(_, {:combine, :except}), do: true def can?(_, _), do: false @impl true def functions(_resource) do [ AshNeo4j.Functions.StContains, AshNeo4j.Functions.StDistance, AshNeo4j.Functions.StDistanceInMeters, AshNeo4j.Functions.StDwithin, AshNeo4j.Functions.StIntersects, AshNeo4j.Functions.StWithin ] end @neo4j %Spark.Dsl.Section{ name: :neo4j, examples: [ """ neo4j do label :Comment relate [{:post, :BELONGS_TO, :outgoing, :Post}] end """ ], schema: [ label: [ type: :atom, doc: "Optional node label", required: false ], relate: [ type: {:list, {:tuple, [:atom, :atom, :atom, :atom]}}, doc: "Optional list of relationships, as tuples of {relationship_name, edge_label, edge_direction, destination_label}", required: false, default: [] ], guard: [ type: {:list, {:tuple, [:atom, :atom, :atom]}}, doc: "Optional list of node relationships, as tuples of {edge_label, edge_direction, destination_label}", required: false, default: [] ], skip: [ type: {:list, :atom}, doc: "Optional list of attributes not to be stored directly as node properties", required: false, default: [] ] ] } @impl true def limit(query, offset, _), do: {:ok, %{query | limit: offset}} @impl true def offset(query, offset, _), do: {:ok, %{query | offset: offset}} @impl true def filter(query, filter, _resource) do {:ok, %{query | filter: filter}} end @impl true def sort(query, sort, _resource) do deduplicated = Enum.uniq_by(sort, fn {field, _order} -> field end) {:ok, %{query | sort: deduplicated}} end @doc false def store_opt(attributes) do if Enum.all?(attributes, &is_atom/1) do {:ok, attributes} else {:error, "Expected all attribute names to be atoms"} end end @sections [@neo4j] use Spark.Dsl.Extension, sections: @sections, persisters: [ AshNeo4j.Persisters.PersistLabels, AshNeo4j.Persisters.PersistTranslations, AshNeo4j.Persisters.PersistRelationshipAttributes, AshNeo4j.Persisters.PersistRelate, AshNeo4j.Persisters.PersistMapping ], verifiers: [ AshNeo4j.Verifiers.VerifyLabelsPascalCase, AshNeo4j.Verifiers.VerifyRelate, AshNeo4j.Verifiers.VerifyGuard, AshNeo4j.Verifiers.VerifyPropertiesCamelCase, AshNeo4j.Verifiers.VerifyEnrichable, AshNeo4j.Verifiers.VerifyAttributeType ] defmodule Query do @moduledoc false defstruct [ :resource, :sort, :filter, :limit, :offset, :domain, aggregates: %{}, calculations: %{}, combination_of: [] ] end @impl true @spec run_query(any(), atom()) :: {:error, any()} | {:ok, any()} def run_query(query, resource) do Logger.debug(""" AshNeo4j.DataLayer: run_query(#{inspect(query)}, #{inspect(resource)}) """) result = case QueryHelper.query_nodes(query) do {:error, error} -> {:error, error} {:ok, []} -> {:ok, []} {:ok, groups} -> all_results = convert_groups_to_resources(query, groups) |> Enum.to_list() case Enum.find(all_results, &match?({:error, _}, &1)) do nil -> records = Enum.map(all_results, fn {:ok, r} -> r r -> r end) aggregates = Map.values(Map.get(query, :aggregates) || %{}) calculations = Map.values(Map.get(query, :calculations) || %{}) with {:ok, records} <- apply_calculations_to_records(records, calculations, resource), records <- filter_matches(records, query.filter, query.domain), {:ok, records} <- apply_aggregates_to_records(records, aggregates, resource) do {:ok, apply_calculation_sort(records, query.sort, query.domain)} end {:error, reason} -> {:error, reason} end end Logger.debug(""" AshNeo4j.DataLayer: run_query result #{inspect(result)} """) result end @impl true def add_aggregates(query, aggregates, _resource) do {:ok, %{query | aggregates: Map.merge(query.aggregates || %{}, Map.new(aggregates, &{&1.name, &1}))}} end @impl true def add_calculation(query, calculation, expression, _resource) do calcs = Map.get(query, :calculations) || %{} {:ok, %{query | calculations: Map.put(calcs, calculation.name, {calculation, expression})}} end @impl true def run_aggregate_query(data_layer_query, aggregates, resource) do mapping = ResourceInfo.mapping(resource) pk_field = hd(Ash.Resource.Info.primary_key(resource)) neo4j_pk = Keyword.get(mapping.properties, pk_field, pk_field) case run_query(data_layer_query, resource) do {:ok, records} -> ids = Enum.map(records, &Map.get(&1, pk_field)) Enum.reduce_while(aggregates, {:ok, %{}}, fn aggregate, {:ok, acc} -> case run_aggregate_for_ids(mapping, neo4j_pk, ids, aggregate, :total) do {:ok, value} -> {:cont, {:ok, Map.put(acc, aggregate.name, value)}} {:error, e} -> {:halt, {:error, e}} end end) {:error, e} -> {:error, e} end end @impl true @spec create(atom() | map(), any()) :: {:error, <<_::64, _::_*8>> | %{:__exception__ => true, :__struct__ => atom(), optional(atom()) => any()}} | {:ok, any()} def create(resource, changeset) do Logger.debug(""" AshNeo4j.DataLayer: create(#{inspect(resource)}, #{inspect(changeset)}) """) mapping = ResourceInfo.mapping(resource) primary_keys = Ash.Resource.Info.primary_key(mapping.module) id_attributes = Map.take(changeset.attributes, primary_keys) result = if Enum.empty?(id_attributes) do {:error, "no values supplied for primary keys #{primary_keys}"} else create_from_attributes(mapping, changeset.attributes) end Logger.debug(""" AshNeo4j.DataLayer: create result #{inspect(result)} """) result end @impl true def upsert(resource, changeset, keys) do Logger.debug(""" AshNeo4j.DataLayer: upsert(#{inspect(resource)}, #{inspect(changeset)}, #{inspect(keys)}) """) mapping = ResourceInfo.mapping(resource) id_properties = id_properties(mapping, changeset.attributes) result = if Enum.any?(Map.values(id_properties), &is_nil(&1)) do create(resource, changeset) else key_filters = Enum.map(keys, fn key -> {key, Ash.Changeset.get_attribute(changeset, key) || Map.get(changeset.params, key) || Map.get(changeset.params, to_string(key))} end) query = Ash.Query.do_filter(resource, and: [key_filters]) resource |> resource_to_query(changeset.domain) |> Map.put(:filter, query.filter) |> Map.put(:tenant, changeset.tenant) |> run_query(resource) |> case do {:ok, []} -> create(resource, changeset) {:ok, [result]} -> to_set = Ash.Changeset.set_on_upsert(changeset, keys) changeset = changeset |> Map.put(:attributes, %{}) |> Map.put(:data, result) |> Ash.Changeset.force_change_attributes(to_set) update(resource, changeset) {:ok, _} -> {:error, "Multiple records matching keys"} end end Logger.debug(""" AshNeo4j.DataLayer: upsert result #{inspect(result)} """) result end @impl true def update(resource, changeset) do Logger.debug(""" AshNeo4j.DataLayer: update(#{inspect(resource)}, #{inspect(changeset)}}) """) mapping = ResourceInfo.mapping(resource) subject_id = id_properties(mapping, changeset.data) subject_label = mapping.label_pair update_properties = dump_properties(mapping, changeset.attributes) remove_property_names = remove_property_names(mapping, changeset.attributes) property_update_result = if !Enum.empty?(update_properties) or !Enum.empty?(remove_property_names) do case subject_label |> Neo4jHelper.update_node(subject_id, update_properties, remove_property_names) do {:ok, %Bolty.Response{results: []}} -> {:error, "no result to update node"} {:ok, %Bolty.Response{results: [node_map | _]}} -> node = Map.get(node_map, "n") convert_node_to_resource(resource, node) {:error, error} -> {:error, error} end end relationship_update_result = if accessing_from = Map.get(changeset.context, :accessing_from) do object_resource = Map.get(accessing_from, :source) object_label = ResourceInfo.label(object_resource) object_relationship_name = Map.get(accessing_from, :name) object_node_relationship = ResourceInfo.node_relationship(object_resource, object_relationship_name) if Map.get(accessing_from, :unrelating?) do object_id = relationship_properties(resource, object_resource, changeset.data, object_relationship_name) case map_size(object_id) do 0 -> {:error, "couldn't unrelate nodes"} _ -> {_relationship_name, edge_label, object_to_subject_direction, _destination_label} = object_node_relationship case Neo4jHelper.unrelate_nodes( subject_label, subject_id, object_label, object_id, edge_label, Util.reverse(object_to_subject_direction) ) do {:ok, %Bolty.Response{results: []}} -> {:error, "no result to unrelate nodes"} {:ok, %Bolty.Response{results: [node_map | _]}} -> node = Map.get(node_map, "s") convert_node_to_resource(resource, node) {:error, error} -> {:error, error} end end else object_id = relationship_properties(resource, object_resource, changeset.attributes, object_relationship_name) case map_size(object_id) do 0 -> {:error, "couldn't relate nodes"} _ -> {_relationship_name, edge_label, object_to_subject_direction, _object_label} = object_node_relationship case Neo4jHelper.relate_nodes( subject_label, subject_id, object_label, object_id, edge_label, Util.reverse(object_to_subject_direction) ) do {:ok, %Bolty.Response{results: []}} -> {:error, "no result to relate nodes"} {:ok, %Bolty.Response{results: [node_map | _]}} -> node = Map.get(node_map, "s") convert_node_to_resource(resource, node) {:error, error} -> {:error, error} end end end else if changeset.relationships do Enum.reduce_while(changeset.relationships, nil, fn {relationship_name, relationship_change}, _acc -> subject_edge = Enum.find(mapping.edges, &(&1.relationship == relationship_name)) subject_relationship = Ash.Resource.Info.relationship(resource, relationship_name) object_resource = subject_relationship.destination object_label = ResourceInfo.label(object_resource) {arguments, options} = hd(relationship_change) type = Keyword.get(options, :type) cond do arguments == [] or type == :remove -> subject_source_attribute = subject_relationship.source_attribute subject_destination_attribute = subject_relationship.destination_attribute object_property_name = ResourceInfo.convert_to_property_name(object_resource, subject_destination_attribute) object_property_value = Map.get(changeset.data, subject_source_attribute) object_id = %{object_property_name => object_property_value} case map_size(object_id) do 0 -> {:halt, {:error, "couldn't unrelate nodes"}} _ -> case Neo4jHelper.unrelate_nodes( subject_label, subject_id, object_label, object_id, subject_edge.label, subject_edge.direction ) do {:ok, %Bolty.Response{results: []}} -> {:halt, {:error, "no result to unrelate nodes"}} {:ok, %Bolty.Response{results: [node_map | _]}} -> node = Map.get(node_map, "s") case convert_node_to_resource(resource, node) do {:ok, resource} -> {:cont, {:ok, resource}} {:error, reason} -> {:halt, {:error, reason}} end {:error, error} -> {:halt, {:error, error}} end end true -> arg_relate_result = Enum.reduce_while(arguments, nil, fn argument, _acc -> object_id = ResourceInfo.convert_to_properties(object_resource, argument) case map_size(object_id) do 0 -> {:halt, {:error, "couldn't relate nodes using argument"}} _ -> subject_exclusive? = ResourceInfo.source_exclusive?(resource, relationship_name) object_exclusive? = ResourceInfo.destination_exclusive?(resource, relationship_name) case Neo4jHelper.relate_nodes( subject_label, subject_id, object_label, object_id, subject_edge.label, subject_edge.direction, {subject_exclusive?, object_exclusive?} ) do {:ok, %Bolty.Response{results: []}} -> {:halt, {:error, "no result to relate nodes"}} {:ok, %Bolty.Response{results: [node_map | _]}} -> node = Map.get(node_map, "s") case convert_node_to_resource(resource, node) do {:ok, resource} -> {:cont, {:ok, resource}} {:error, reason} -> {:halt, {:error, reason}} end {:error, error} -> {:halt, {:error, error}} end end end) case arg_relate_result do {:error, _} -> {:halt, arg_relate_result} {:ok, _} -> {:cont, arg_relate_result} end end end) else {:error, "changeset not handled"} end end result = relationship_update_result || property_update_result Logger.debug(""" AshNeo4j.DataLayer: update result #{inspect(result)} """) result end @impl true def destroy(resource, changeset) do Logger.debug(""" AshNeo4j.DataLayer: destroy(#{inspect(resource)}, #{inspect(changeset)}}) """) mapping = ResourceInfo.mapping(resource) label = mapping.label_pair id_properties = id_properties(mapping, changeset.data) result = case Neo4jHelper.safe_delete_nodes(label, id_properties, ResourceInfo.preserve_node_relationships(resource)) do {:ok, _} -> :ok {:error, "nothing deleted"} -> {:error, Ash.Error.Invalid.Unavailable.exception( resource: resource, source: AshNeo4j.DataLayer, reason: "guarded relationships prevent deletion" )} {:error, error} -> {:error, error} end Logger.debug("AshNeo4j.DataLayer: delete result #{inspect(result)}") result end @impl true def resource_to_query(resource, domain) do %Query{resource: resource, domain: domain} end @impl true def combination_of(combinations, resource, domain) do {:ok, %Query{resource: resource, domain: domain, combination_of: combinations}} end @impl true def transaction(resource, fun, _timeout, _) do label = ResourceInfo.label(resource) if AshNeo4j.Sandbox.active?() do prev = Process.get(:ash_neo4j_in_sandbox_tx, false) Process.put(:ash_neo4j_in_sandbox_tx, true) Process.put({:neo4j_in_transaction, label}, true) try do case fun.() do {:error, error} -> {:error, error} result -> {:ok, result} end catch {{:neo4j_rollback, _}, value} -> {:error, value} after Process.put(:ash_neo4j_in_sandbox_tx, prev) Process.delete({:neo4j_in_transaction, label}) end else Bolty.transaction(Bolt, fn conn -> stack = Process.get(:ash_neo4j_tx_stack, []) Process.put(:ash_neo4j_tx_stack, [conn | stack]) Process.put({:neo4j_in_transaction, label}, true) try do case fun.() do {:error, error} -> Bolty.rollback(conn, error) result -> result end catch {{:neo4j_rollback, _}, value} -> Bolty.rollback(conn, value) after Process.put(:ash_neo4j_tx_stack, stack) Process.delete({:neo4j_in_transaction, label}) end end) end end @impl true def rollback(resource, error) do throw({{:neo4j_rollback, ResourceInfo.label(resource)}, error}) end @impl true def in_transaction?(_resource) do Process.get(:ash_neo4j_tx_stack, []) != [] or Process.get(:ash_neo4j_in_sandbox_tx, false) end defp filter_matches(records, nil, _domain), do: records defp filter_matches(records, filter, domain) do {:ok, records} = Ash.Filter.Runtime.filter_matches(domain, records, filter) records end defp consolidate_groups(groups) when is_list(groups) do Enum.reduce(groups, [], fn group, acc -> s = Map.get(group, "s") tuple_count = Integer.floor_div(Enum.count(group), 2) tuples = Enum.reduce(0..(tuple_count - 1)//1, [], fn tuple, acc -> r = Map.get(group, "r#{tuple}") || Map.get(group, "r") d = Map.get(group, "d#{tuple}") || Map.get(group, "d") cond do r != nil && d != nil -> [{r, d} | acc] true -> acc end end) cond do [] == acc -> [{s, tuples}] [previous | tail] = acc -> cond do Map.get(s, :id) == elem(previous, 0).id -> cond do tuples == [] -> acc true -> [{s, tuples ++ elem(previous, 1)} | tail] end true -> [{s, tuples} | acc] end end end) |> Enum.into( [], fn group -> {source, related} = group {source, Enum.reverse(related)} end ) |> Enum.reverse() end defp convert_groups_to_resources(query, groups) when is_struct(query, Query) and is_list(groups) do mapping = ResourceInfo.mapping(query.resource) consolidate_groups(groups) |> Stream.map(&convert_to_resource(query, mapping, &1)) end defp convert_to_resource(query, %ResourceMapping{} = mapping, consolidated_group) when is_struct(query, Query) and is_tuple(consolidated_group) do source_node = elem(consolidated_group, 0) related = elem(consolidated_group, 1) enrichments = Enum.reduce(related, [], &enrichments(mapping, &2, &1)) |> consolidate_enrichments() convert_node_to_resource(mapping, source_node, enrichments) end defp consolidate_enrichments(enrichments) when is_list(enrichments) do Enum.reduce(enrichments, [], fn enrichment, acc -> case enrichment do {name, value} -> cond do [] == acc -> [enrichment] [head | tail] = acc -> cond do name == elem(head, 0) and is_list(value) and is_list(elem(head, 1)) -> merged_value = [hd(value) | elem(head, 1)] [{name, merged_value} | tail] true -> [enrichment | acc] end end nil -> acc end end) end defp enrichments(%ResourceMapping{} = mapping, acc, {edge, dest_node}) when is_list(acc) and is_map(edge) and is_map(dest_node) do dest_labels = Enum.into(dest_node.labels, [], &String.to_atom(&1)) edge_label = String.to_atom(edge.type) edge_direction = edge_direction(edge, dest_node) dest_labels_filtered = List.delete(dest_labels, mapping.domain_label) relationship = Enum.find_value(dest_labels_filtered, fn dest_label -> case Enum.find(mapping.edges, fn ed -> ed.label == edge_label and ed.direction == edge_direction and ed.destination_label == dest_label end) do nil -> nil ed -> Ash.Resource.Info.relationship(mapping.module, ed.relationship) end end) if relationship != nil do reverse_node_relationship = ResourceInfo.reverse_node_relationship(mapping.module, relationship.name) reverse_relationship = if reverse_node_relationship != nil do Ash.Resource.Info.relationship(relationship.destination, elem(reverse_node_relationship, 0)) end cond do relationship.cardinality == :one && relationship.type == :belongs_to -> destination_property = ResourceInfo.convert_to_property_name(relationship.destination, relationship.destination_attribute) [ {relationship.source_attribute, Map.get(dest_node.properties, destination_property)} | acc ] reverse_relationship != nil && (reverse_relationship.cardinality == :one && reverse_relationship.type == :has_one) -> source_property = Keyword.get(mapping.properties, relationship.source_attribute, relationship.source_attribute) |> to_string() [ {relationship.destination_attribute, Map.get(dest_node.properties, source_property)} | acc ] relationship.cardinality == :many && reverse_relationship != nil && reverse_relationship.cardinality == :many -> case convert_node_to_resource(relationship.destination, dest_node, []) do {:ok, dest_resource} -> [{relationship.name, [dest_resource]} | acc] {:error, reason} -> Logger.debug("AshNeo4j.DataLayer: unable to convert enrichment node: #{inspect(reason)}") acc end true -> acc end else acc end end defp edge_direction(edge, dest_node) when is_map(edge) and is_map(dest_node) do cond do dest_node.id == edge.start -> :incoming dest_node.id == edge.end -> :outgoing true -> nil end end defp convert_node_to_resource(subject, node, enrichments \\ []) defp convert_node_to_resource(%ResourceMapping{} = mapping, node, enrichments) when is_map(node) and is_list(enrichments) do convert_node_to_resource_impl(mapping.module, mapping.properties, node, enrichments) end defp convert_node_to_resource(resource, node, enrichments) when is_atom(resource) and is_map(node) and is_list(enrichments) do convert_node_to_resource_impl(resource, ResourceInfo.translations(resource), node, enrichments) end defp convert_node_to_resource_impl(resource, translations, node, enrichments) when is_atom(resource) and is_list(translations) and is_map(node) and is_list(enrichments) do enriched = Enum.into(enrichments, %{}, fn {field, value} -> {field, value} end) fields_result = Enum.reduce_while(translations, {:ok, enriched}, fn {resource_field, node_field}, {:ok, acc} -> property_value = Map.get(node.properties, to_string(node_field)) case cast_attribute(resource, resource_field, property_value) do {:ok, value} -> {:cont, {:ok, Map.put(acc, resource_field, value)}} {:error, reason} -> {:halt, {:error, reason}} end end) case fields_result do {:error, reason} -> {:error, reason} {:ok, fields} -> belongs_to = Ash.Resource.Info.relationships(resource) |> Enum.filter(fn relationship -> relationship.type == :belongs_to and relationship.allow_nil? end) nilled_fields = Enum.reduce(belongs_to, fields, fn relationship, acc -> if Map.get(fields, relationship.source_attribute) == nil do acc |> Map.put(relationship.name, nil) else acc end end) {:ok, struct!(resource, nilled_fields) |> Ash.Resource.set_metadata(%{node_id: node.id, data_layer: __MODULE__, labels: node.labels}) |> Ash.Resource.set_meta(struct(Ecto.Schema.Metadata, state: :loaded))} end end defp create_from_attributes(%ResourceMapping{} = mapping, attributes) when is_map(attributes) do properties = dump_properties(mapping, attributes) case create_node(mapping, properties) do {:ok, source_resource} -> relate_nodes(source_resource, mapping.module, attributes, mapping) {:error, error} -> {:error, error} end end defp relate_nodes(source_resource, resource, attributes, %ResourceMapping{} = mapping) when is_struct(source_resource) and is_atom(resource) and is_map(attributes) do relationship_attributes = mapping.relationship_attributes |> Keyword.delete(:id) relationship_source_attributes = Map.take(attributes, Keyword.keys(relationship_attributes)) case Enum.count(relationship_source_attributes) do 0 -> {:ok, source_resource} _ -> relationships = relationship_attributes |> Enum.reduce( [], fn {source_attribute, name}, acc -> relationship = Ash.Resource.Info.relationship(resource, name) dest_resource = relationship.destination case Enum.find(mapping.edges, &(&1.relationship == name)) do nil -> acc %EdgeDescriptor{label: edge_label, direction: edge_direction, destination_label: destination_label} -> dest_node_property_name = Keyword.get(ResourceInfo.translations(dest_resource), relationship.destination_attribute) dest_id_value = Map.get(relationship_source_attributes, source_attribute) if dest_id_value == nil do acc else dest_id = %{dest_node_property_name => dest_id_value} exclusive = ResourceInfo.destination_exclusive?(resource, name) [{destination_label, dest_id, edge_label, edge_direction, exclusive} | acc] end end end ) label = mapping.label_pair id_properties = id_properties(mapping, attributes) case Neo4jHelper.relate_nodes(label, id_properties, relationships) do :ok -> case Neo4jHelper.read_nodes_related(label, id_properties) do {:ok, %Bolty.Response{results: groups}} -> consolidated_groups = consolidate_groups(groups) cond do length(consolidated_groups) == 1 -> query = resource_to_query(resource, Ash.Resource.Info.domain(resource)) convert_to_resource(query, mapping, hd(consolidated_groups)) true -> {:error, "expected groups to consolidate to a single group (resource)"} end {:error, error} -> {:error, error} end {:error, error} -> {:error, error} end end end defp create_node(%ResourceMapping{} = mapping, properties) when is_map(properties) do case mapping.all_labels |> Neo4jHelper.create_node(properties) do {:ok, %Bolty.Response{results: [node_map | _]}} -> node = Map.get(node_map, "n") convert_node_to_resource(mapping.module, node) {:error, error} -> {:error, error} end end defp id_properties(%ResourceMapping{} = mapping, map) when is_map(map) do primary_keys = Ash.Resource.Info.primary_key(mapping.module) Enum.into(primary_keys, %{}, fn key -> {Keyword.get(mapping.properties, key, key), Map.get(map, key)} end) end defp relationship_properties(source_resource, dest_resource, source_map, dest_relationship_name) when is_atom(source_resource) and is_atom(dest_resource) and is_map(source_map) and is_atom(dest_relationship_name) do source_node_relationship = ResourceInfo.reverse_node_relationship(dest_resource, dest_relationship_name) if source_node_relationship != nil do source_relationship_name = elem(source_node_relationship, 0) source_relationship = Ash.Resource.Info.relationship(source_resource, source_relationship_name) value = Map.get(source_map, source_relationship.source_attribute) if value != nil do dest_property_name = ResourceInfo.convert_to_property_name(dest_resource, source_relationship.destination_attribute) |> String.to_atom() %{dest_property_name => value} else %{} end else %{} end end defp remove_property_names(%ResourceMapping{} = mapping, map) when is_map(map) do map |> Map.reject(fn {_k, v} -> v != nil end) |> Enum.into([], fn {field, _} -> Keyword.get(mapping.properties, field, nil) end) |> Enum.reject(fn field -> field == nil end) end def cast_attribute(_resource, _name, nil) do {:ok, nil} end def cast_attribute(resource, name, value) when is_atom(resource) and is_atom(name) do attribute = Ash.Resource.Info.attribute(resource, name) if attribute == nil do Logger.debug( "AshNeo4j.DataLayer: no attribute found for resource #{inspect(resource)} and name #{inspect(name)}, returning original value" ) {:ok, value} else Cast.cast(attribute.type, value, attribute.constraints) end end defp dump_properties(%ResourceMapping{} = mapping, attributes) when is_map(attributes) do mapping.properties |> Enum.reduce(%{}, fn {key, translated_key}, acc -> value = Map.get(attributes, key) if value != nil do attribute = Ash.Resource.Info.attribute(mapping.module, key) attribute_type = Ash.Type.get_type!(attribute.type) dumped = Dump.dump(attribute_type, value, attribute.constraints) acc = Map.put(acc, translated_key, dumped) # Types that expose `companions/1` (e.g. Box, Polygon) also write # dotted scalar companion properties for indexed predicates. if is_atom(attribute_type) and Code.ensure_loaded?(attribute_type) and function_exported?(attribute_type, :companions, 1) do attribute_type.companions(dumped) |> Enum.reduce(acc, fn {suffix, val}, inner_acc -> Map.put(inner_acc, "#{translated_key}.#{suffix}", val) end) else acc end else acc end end) end defp apply_aggregates_to_records(records, [], _resource), do: {:ok, records} defp apply_aggregates_to_records(records, aggregates, resource) do mapping = ResourceInfo.mapping(resource) pk_field = hd(Ash.Resource.Info.primary_key(resource)) neo4j_pk = Keyword.get(mapping.properties, pk_field, pk_field) ids = Enum.map(records, &Map.get(&1, pk_field)) Enum.reduce_while(aggregates, {:ok, records}, fn aggregate, {:ok, acc_records} -> case run_aggregate_for_ids(mapping, neo4j_pk, ids, aggregate, :per_record) do {:ok, agg_map} -> updated = Enum.map(acc_records, fn record -> id = Map.get(record, pk_field) value = Map.get(agg_map, id, aggregate.default_value) Map.put(record, aggregate.name, value) end) {:cont, {:ok, updated}} {:error, e} -> {:halt, {:error, e}} end end) end defp apply_calculations_to_records(records, [], _resource), do: {:ok, records} defp apply_calculations_to_records(records, calculations, resource) do Enum.reduce_while(calculations, {:ok, records}, fn {calculation, expression}, {:ok, acc} -> case Ash.Filter.hydrate_refs(expression, %{resource: resource, public?: false, eval?: true}) do {:ok, hydrated} -> updated = Enum.map(acc, fn record -> case Ash.Expr.eval_hydrated(hydrated, record: record, resource: resource, unknown_on_unknown_refs?: true) do {:ok, value} -> Map.put(record, calculation.name, value) _ -> record end end) {:cont, {:ok, updated}} {:error, e} -> {:halt, {:error, e}} end end) end defp apply_calculation_sort(records, sort, _domain) when sort in [nil, []], do: records defp apply_calculation_sort(records, sort, domain) do if Enum.any?(sort, fn {term, _} -> is_struct(term, Ash.Query.Calculation) end) do Ash.Actions.Sort.runtime_sort(records, sort, domain: domain, rekey?: false) else records end end defp run_aggregate_for_ids(_mapping, _neo4j_pk, [], aggregate, _mode) do {:ok, aggregate.default_value} end defp run_aggregate_for_ids(%ResourceMapping{} = mapping, neo4j_pk, ids, aggregate, mode) do case resolve_aggregate_path(mapping, aggregate.relationship_path) do {:ok, path_segments, dest_mapping} -> neo4j_field = if aggregate.field && is_atom(aggregate.field), do: Keyword.get(dest_mapping.properties, aggregate.field, aggregate.field), else: nil embedded = embedded_field_type(dest_mapping.module, aggregate.field) cond do # Expression-based aggregates always load full records in Elixir; # run_expr_agg handles aggregate.filter internally via apply_record_filter. is_struct(aggregate.field, Ash.Query.Calculation) -> run_expr_agg(mapping, neo4j_pk, ids, aggregate, mode, path_segments, dest_mapping) # When a filter is present, try to push scalar == conditions into Cypher. # Falls back to Elixir-side filtering for complex or embedded-field filters. aggregate_has_filter?(aggregate) -> case {simple_agg_filter(aggregate, dest_mapping), embedded} do {{:ok, dest_conditions}, nil} -> run_simple_filtered_aggregate( mapping, neo4j_pk, ids, aggregate, mode, path_segments, neo4j_field, dest_conditions ) _ -> run_filtered_aggregate(mapping, neo4j_pk, ids, aggregate, mode, path_segments, dest_mapping) end embedded -> {field_type, field_constraints} = embedded run_embedded_agg( mapping, neo4j_pk, ids, aggregate, mode, path_segments, neo4j_field, field_type, field_constraints ) true -> query = case mode do :per_record -> CypherQuery.aggregate_per_record( mapping.label_pair, neo4j_pk, ids, path_segments, aggregate.kind, neo4j_field, aggregate.name, aggregate.uniq? ) :total -> CypherQuery.aggregate_total( mapping.label_pair, neo4j_pk, ids, path_segments, aggregate.kind, neo4j_field, aggregate.name, aggregate.uniq? ) end case Cypher.run(query) do {:ok, %Bolty.Response{results: rows}} -> case mode do :per_record -> {:ok, Map.new(rows, fn row -> {Map.get(row, "source_id"), Map.get(row, to_string(aggregate.name))} end)} :total -> value = rows |> List.first(%{}) |> Map.get(to_string(aggregate.name), aggregate.default_value) {:ok, value} end {:error, e} -> {:error, e} end end {:error, e} -> {:error, e} end end # Handles any aggregate that carries a filter expression. Loads all destination # records for the given source IDs via Elixir, applies the Ash runtime filter, # then computes the aggregate in Elixir. # # This path is also used for expression-based aggregates (Ash.Query.Calculation # field) when a filter is present, because we already load full records there. defp run_filtered_aggregate(mapping, neo4j_pk, ids, aggregate, mode, path_segments, dest_mapping) do query = CypherQuery.related_nodes(mapping.label_pair, neo4j_pk, ids, path_segments) dest_resource = dest_mapping.module domain = Ash.Resource.Info.domain(dest_resource) with {:ok, %Bolty.Response{results: rows}} <- Cypher.run(query) do pairs = Enum.flat_map(rows, fn row -> source_id = Map.get(row, "source_id") dest_node = Map.get(row, "dest_node") if dest_node do case convert_node_to_resource(dest_resource, dest_node) do {:ok, record} -> [{source_id, record}] _ -> [] end else [] end end) case mode do :per_record -> grouped = Enum.group_by(pairs, &elem(&1, 0), &elem(&1, 1)) result = Map.new(grouped, fn {source_id, records} -> {:ok, filtered} = Ash.Filter.Runtime.filter_matches(domain, records, aggregate.query.filter) values = extract_aggregate_field_values(filtered, aggregate) {source_id, apply_elixir_aggregate(aggregate.kind, values, aggregate.default_value)} end) {:ok, result} :total -> all_records = Enum.map(pairs, &elem(&1, 1)) {:ok, filtered} = Ash.Filter.Runtime.filter_matches(domain, all_records, aggregate.query.filter) values = extract_aggregate_field_values(filtered, aggregate) {:ok, apply_elixir_aggregate(aggregate.kind, values, aggregate.default_value)} end end end # Handles aggregates whose filter is a set of simple scalar == conditions that can be # expressed as WHERE clauses in Cypher, avoiding full record loading in Elixir. defp run_simple_filtered_aggregate(mapping, neo4j_pk, ids, aggregate, mode, path_segments, neo4j_field, dest_conditions) do query = case mode do :per_record -> CypherQuery.aggregate_per_record( mapping.label_pair, neo4j_pk, ids, path_segments, aggregate.kind, neo4j_field, aggregate.name, aggregate.uniq?, dest_conditions ) :total -> CypherQuery.aggregate_total( mapping.label_pair, neo4j_pk, ids, path_segments, aggregate.kind, neo4j_field, aggregate.name, aggregate.uniq?, dest_conditions ) end case Cypher.run(query) do {:ok, %Bolty.Response{results: rows}} -> case mode do :per_record -> {:ok, Map.new(rows, fn row -> {Map.get(row, "source_id"), Map.get(row, to_string(aggregate.name))} end)} :total -> value = rows |> List.first(%{}) |> Map.get(to_string(aggregate.name), aggregate.default_value) {:ok, value} end {:error, e} -> {:error, e} end end # Returns {:ok, [{prop_string, value}]} when the aggregate filter consists entirely of # scalar == equality predicates on non-embedded destination attributes, enabling # WHERE pushdown into Cypher. Returns :complex otherwise and falls back to Elixir-side filtering. defp simple_agg_filter(aggregate, dest_mapping) do filter = aggregate_query_filter(aggregate) try do simple = Ash.Filter.to_simple_filter(filter, skip_invalid?: false) predicates = Map.get(simple, :predicates, []) if Enum.empty?(predicates) do :complex else Enum.reduce_while(predicates, {:ok, []}, fn predicate, {:ok, acc} -> cond do Map.get(predicate, :operator) != :== -> {:halt, :complex} not match?(%Ash.Query.Ref{}, Map.get(predicate, :left)) -> {:halt, :complex} match?(%Ash.Query.Calculation{}, Map.get(predicate.left, :attribute)) -> {:halt, :complex} true -> attr_name = Ash.Query.Ref.name(predicate.left) case embedded_field_type(dest_mapping.module, attr_name) do nil -> prop = Keyword.get(dest_mapping.properties, attr_name, attr_name) |> to_string() {:cont, {:ok, acc ++ [{prop, predicate.right}]}} _ -> {:halt, :complex} end end end) end rescue _ -> :complex end end # Extracts the aggregate's target field value from each record, respecting uniq?. defp extract_aggregate_field_values(records, aggregate) do values = Enum.map(records, fn record -> case aggregate.field do nil -> record field when is_atom(field) -> Map.get(record, field) _ -> record end end) if aggregate.uniq?, do: Enum.uniq(values), else: values end defp embedded_field_type(resource_module, field_name) when is_atom(field_name) do case Ash.Resource.Info.attribute(resource_module, field_name) do nil -> nil attr -> type = Ash.Type.get_type(attr.type) case TypeClassifier.classify(type) do {:ok, :ash_json, _} -> {type, attr.constraints} _ -> nil end end end defp embedded_field_type(_, _), do: nil defp run_embedded_agg( mapping, neo4j_pk, ids, aggregate, mode, path_segments, neo4j_field, field_type, field_constraints ) do query = case mode do :per_record -> CypherQuery.aggregate_per_record( mapping.label_pair, neo4j_pk, ids, path_segments, :list, neo4j_field, aggregate.name, aggregate.uniq? ) :total -> CypherQuery.aggregate_total( mapping.label_pair, neo4j_pk, ids, path_segments, :list, neo4j_field, aggregate.name, aggregate.uniq? ) end case Cypher.run(query) do {:ok, %Bolty.Response{results: rows}} -> agg_key = to_string(aggregate.name) case mode do :per_record -> {:ok, Map.new(rows, fn row -> source_id = Map.get(row, "source_id") raw_list = Map.get(row, agg_key, []) cast_list = cast_raw_list(raw_list, field_type, field_constraints) {source_id, apply_elixir_aggregate(aggregate.kind, cast_list, aggregate.default_value)} end)} :total -> raw_list = rows |> List.first(%{}) |> Map.get(agg_key, []) cast_list = cast_raw_list(raw_list, field_type, field_constraints) {:ok, apply_elixir_aggregate(aggregate.kind, cast_list, aggregate.default_value)} end {:error, e} -> {:error, e} end end defp run_expr_agg(mapping, neo4j_pk, ids, aggregate, mode, path_segments, dest_mapping) do query = CypherQuery.related_nodes(mapping.label_pair, neo4j_pk, ids, path_segments) dest_resource = dest_mapping.module domain = Ash.Resource.Info.domain(dest_resource) calc = aggregate.field expr = calc.opts[:expr] case Ash.Filter.hydrate_refs(expr, %{resource: dest_resource, public?: false, eval?: true}) do {:ok, hydrated} -> case Cypher.run(query) do {:ok, %Bolty.Response{results: rows}} -> record_pairs = Enum.flat_map(rows, fn row -> source_id = Map.get(row, "source_id") dest_node = Map.get(row, "dest_node") if dest_node do case convert_node_to_resource(dest_resource, dest_node) do {:ok, record} -> [{source_id, record}] _ -> [] end else [] end end) # Apply aggregate filter if present, then evaluate the expression. pairs = apply_record_filter(record_pairs, aggregate_query_filter(aggregate), domain) |> Enum.flat_map(fn {source_id, record} -> case Ash.Expr.eval_hydrated(hydrated, record: record, resource: dest_resource, unknown_on_unknown_refs?: true ) do {:ok, value} when not is_nil(value) -> [{source_id, value}] _ -> [] end end) case mode do :per_record -> grouped = Enum.group_by(pairs, &elem(&1, 0), &elem(&1, 1)) {:ok, Map.new(grouped, fn {source_id, values} -> {source_id, apply_elixir_aggregate(aggregate.kind, values, aggregate.default_value)} end)} :total -> values = Enum.map(pairs, &elem(&1, 1)) {:ok, apply_elixir_aggregate(aggregate.kind, values, aggregate.default_value)} end {:error, e} -> {:error, e} end {:error, e} -> {:error, e} end end # Returns true when the aggregate carries a real (non-trivial) filter in its # query. Ash always provides an Ash.Query on the aggregate; unfiltered aggregates # have %Ash.Filter{expression: true}. We only route through the Elixir-side # path when there is an actual user-defined filter to honour. defp aggregate_has_filter?(aggregate) do case aggregate_query_filter(aggregate) do %Ash.Filter{expression: true} -> false %Ash.Filter{} -> true _ -> false end end # Extracts the filter from aggregate.query, returning nil if absent. defp aggregate_query_filter(aggregate) do case Map.get(aggregate, :query) do %Ash.Query{filter: filter} -> filter _ -> nil end end # Applies an Ash filter (if any) to a list of {source_id, record} pairs, # keeping per-source grouping so filter predicates referencing destination # attributes are evaluated correctly. defp apply_record_filter(pairs, nil, _domain), do: pairs defp apply_record_filter(pairs, filter, domain) do grouped = Enum.group_by(pairs, &elem(&1, 0), &elem(&1, 1)) Enum.flat_map(grouped, fn {source_id, records} -> {:ok, filtered} = Ash.Filter.Runtime.filter_matches(domain, records, filter) Enum.map(filtered, &{source_id, &1}) end) end defp cast_raw_list(raw_list, field_type, field_constraints) when is_list(raw_list) do case Cast.cast({:array, field_type}, raw_list, field_constraints) do {:ok, values} -> values {:error, _} -> [] end end defp cast_raw_list(_, _, _), do: [] defp apply_elixir_aggregate(:list, values, _default), do: values defp apply_elixir_aggregate(:first, values, default), do: List.first(values, default) defp apply_elixir_aggregate(:count, values, _default), do: length(values) defp apply_elixir_aggregate(:exists, values, _default), do: values != [] defp apply_elixir_aggregate(:sum, [], default), do: default defp apply_elixir_aggregate(:sum, values, _default), do: Enum.sum(values) defp apply_elixir_aggregate(:avg, [], default), do: default defp apply_elixir_aggregate(:avg, values, _default), do: Enum.sum(values) / length(values) defp apply_elixir_aggregate(:min, [], default), do: default defp apply_elixir_aggregate(:min, values, _default), do: Enum.min(values) defp apply_elixir_aggregate(:max, [], default), do: default defp apply_elixir_aggregate(:max, values, _default), do: Enum.max(values) defp resolve_aggregate_path(%ResourceMapping{} = mapping, relationship_path) do Enum.reduce_while(relationship_path, {mapping, []}, fn name, {current_mapping, segments} -> case Enum.find(current_mapping.edges, &(&1.relationship == name)) do nil -> {:halt, {:error, "relationship #{name} not found on #{current_mapping.module}"}} %EdgeDescriptor{label: edge_label, direction: direction, destination_label: dest_label} -> relationship = Ash.Resource.Info.relationship(current_mapping.module, name) next_mapping = ResourceInfo.mapping(relationship.destination) {:cont, {next_mapping, [{edge_label, direction, dest_label} | segments]}} end end) |> case do {:error, _} = error -> error {dest_mapping, segments} -> {:ok, Enum.reverse(segments), dest_mapping} end end end