Packages
ash
3.29.3
3.29.3
3.29.2
3.29.1
3.29.0
3.28.0
3.27.8
3.27.7
3.27.6
3.27.5
3.27.4
3.27.3
3.27.2
3.27.1
3.27.0
3.26.0
3.25.2
3.25.1
3.25.0
retired
3.24.7
3.24.6
3.24.5
3.24.4
3.24.3
3.24.2
3.24.1
3.24.0
3.23.1
3.23.0
3.22.2
3.22.1
3.22.0
3.21.3
3.21.2
3.21.1
3.21.0
3.20.0
3.19.3
3.19.2
3.19.1
3.19.0
3.18.0
3.17.1
3.17.0
3.16.0
3.15.0
3.14.1
3.14.0
retired
3.13.2
3.13.1
3.13.0
3.12.0
3.11.3
3.11.2
3.11.1
3.11.0
3.10.1
3.10.0
3.9.0
3.8.0
3.7.6
3.7.5
3.7.4
3.7.3
3.7.2
3.7.1
3.7.0
retired
3.6.3
retired
3.6.2
3.6.1
3.6.0
3.5.43
3.5.42
3.5.41
3.5.40
3.5.39
3.5.38
3.5.37
3.5.36
3.5.35
3.5.34
3.5.33
3.5.32
3.5.31
3.5.30
3.5.29
3.5.28
3.5.27
3.5.26
3.5.25
3.5.24
3.5.23
3.5.22
3.5.21
3.5.20
3.5.19
3.5.18
3.5.17
3.5.16
3.5.15
3.5.14
3.5.13
3.5.12
3.5.11
3.5.10
3.5.9
3.5.8
3.5.7
3.5.6
3.5.5
3.5.4
3.5.3
3.5.2
3.5.1
3.5.0
3.4.74
retired
3.4.73
3.4.72
3.4.71
3.4.70
3.4.69
3.4.68
3.4.67
3.4.66
3.4.65
3.4.64
3.4.63
3.4.62
3.4.61
3.4.60
3.4.59
3.4.58
3.4.57
3.4.56
3.4.55
3.4.54
3.4.53
3.4.52
3.4.51
3.4.50
3.4.49
3.4.48
3.4.47
3.4.46
3.4.45
3.4.44
3.4.43
3.4.42
3.4.41
3.4.40
3.4.39
3.4.38
3.4.37
3.4.36
3.4.35
3.4.34
3.4.33
3.4.32
3.4.31
3.4.30
3.4.29
3.4.28
3.4.27
3.4.26
3.4.25
3.4.24
3.4.23
3.4.22
3.4.21
3.4.20
3.4.19
3.4.18
3.4.17
3.4.16
3.4.15
3.4.14
3.4.13
3.4.12
3.4.11
3.4.10
3.4.9
3.4.8
3.4.7
3.4.6
3.4.5
3.4.4
3.4.3
3.4.2
3.4.1
3.4.0
3.3.3
3.3.2
3.3.1
3.3.0
3.2.6
3.2.5
3.2.4
3.2.3
3.2.2
3.2.1
3.2.0
3.1.8
3.1.7
3.1.6
3.1.5
3.1.4
3.1.3
3.1.2
3.1.1
3.1.0
3.0.16
3.0.15
3.0.14
3.0.13
3.0.12
3.0.11
3.0.10
3.0.9
3.0.8
3.0.7
3.0.6
3.0.5
3.0.4
3.0.3
3.0.2
3.0.1
3.0.0
3.0.0-rc.46
3.0.0-rc.45
3.0.0-rc.44
3.0.0-rc.43
3.0.0-rc.42
3.0.0-rc.41
3.0.0-rc.40
3.0.0-rc.39
3.0.0-rc.38
3.0.0-rc.37
3.0.0-rc.36
3.0.0-rc.35
3.0.0-rc.34
3.0.0-rc.33
3.0.0-rc.32
3.0.0-rc.31
3.0.0-rc.29
3.0.0-rc.28
3.0.0-rc.27
3.0.0-rc.26
3.0.0-rc.25
3.0.0-rc.24
3.0.0-rc.23
3.0.0-rc.22
3.0.0-rc.21
3.0.0-rc.20
3.0.0-rc.19
3.0.0-rc.18
3.0.0-rc.17
3.0.0-rc.16
3.0.0-rc.15
3.0.0-rc.14
3.0.0-rc.13
3.0.0-rc.12
3.0.0-rc.11
3.0.0-rc.10
3.0.0-rc.9
3.0.0-rc.8
3.0.0-rc.7
3.0.0-rc.6
3.0.0-rc.5
3.0.0-rc.4
3.0.0-rc.3
3.0.0-rc.1
3.0.0-rc.0
2.21.15
2.21.14
2.21.13
2.21.12
2.21.11
2.21.10
2.21.9
2.21.8
2.21.7
2.21.6
2.21.5
2.21.4
2.21.3
2.21.2
2.21.1
2.21.0
2.20.3
2.20.2
2.20.1
2.20.0
2.19.14
2.19.13
2.19.12
2.19.11
2.19.10
2.19.9
2.19.8
2.19.7
2.19.6
2.19.5
2.19.4
2.19.3
retired
2.19.2
retired
2.19.1
retired
2.19.0
retired
2.18.2
2.18.1
2.18.0
2.17.24
2.17.23
2.17.22
2.17.21
2.17.20
2.17.19
2.17.18
2.17.17
2.17.16
2.17.15
2.17.14
2.17.13
2.17.12
2.17.11
2.17.10
2.17.9
2.17.8
2.17.7
2.17.6
2.17.5
2.17.4
2.17.3
2.17.2
2.17.1
2.17.0
2.16.1
2.16.0
2.15.20
2.15.19
2.15.18
2.15.17
2.15.16
2.15.15
2.15.14
2.15.13
2.15.12
2.15.11
2.15.10
2.15.9
2.15.8
2.15.7
2.15.6
2.15.5
2.15.4
2.15.2
2.15.1
2.15.0
2.14.21
2.14.20
2.14.19
2.14.18
2.14.17
2.14.16
2.14.15
2.14.14
2.14.13
2.14.12
2.14.11
2.14.10
2.14.9
2.14.8
2.14.7
2.14.6
2.14.5
2.14.4
2.14.3
2.14.2
2.14.1
2.14.0
2.13.4
retired
2.13.3
2.13.2
2.13.1
2.13.0
2.12.1
2.12.0
2.11.11
2.11.10
2.11.9
2.11.8
2.11.7
2.11.6
2.11.5
2.11.4
2.11.3
2.11.2
2.11.1
2.11.0
2.11.0-rc.3
2.11.0-rc.2
2.11.0-rc.1
2.11.0-rc.0
2.10.2
2.10.1
2.10.0
2.9.29
2.9.28
2.9.27
2.9.26
2.9.25
2.9.24
2.9.23
2.9.22
2.9.21
2.9.20
2.9.19
2.9.18
2.9.17
2.9.16
2.9.15
2.9.14
2.9.13
2.9.12
2.9.11
2.9.10
2.9.9
2.9.8
2.9.7
2.9.6
2.9.5
2.9.4
2.9.3
2.9.2
2.9.1
2.9.0
2.8.1
2.8.0
2.7.1
2.7.0
2.6.31
2.6.30
2.6.29
2.6.28
2.6.27
2.6.26
2.6.25
2.6.24
2.6.23
2.6.22
2.6.21
2.6.20
2.6.19
2.6.18
2.6.17
2.6.16
2.6.15
2.6.14
2.6.13
2.6.11
2.6.10
2.6.9
2.6.8
2.6.7
2.6.6
2.6.5
2.6.4
2.6.3
2.6.2
2.6.1
2.6.0
2.5.16
2.5.15
2.5.14
2.5.13
2.5.12
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.5.0-rc.6
2.5.0-rc.5
2.5.0-rc.4
2.5.0-rc.3
2.5.0-rc.2
2.5.0-rc.1
2.5.0-rc.0
2.4.30
2.4.29
2.4.28
2.4.27
2.4.26
2.4.25
2.4.24
2.4.23
2.4.22
2.4.21
2.4.20
2.4.19
2.4.18
2.4.17
2.4.16
2.4.15
2.4.14
2.4.13
2.4.12
2.4.11
2.4.10
2.4.9
2.4.8
2.4.7
2.4.6
2.4.5
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.0
2.2.0
2.1.0
2.0.0
2.0.0-rc.15
2.0.0-rc.14
2.0.0-rc.13
2.0.0-rc.12
2.0.0-rc.11
2.0.0-rc.10
2.0.0-rc.9
2.0.0-rc.8
2.0.0-rc.7
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
2.0.0-pre.8
2.0.0-pre.7
2.0.0-pre.6
2.0.0-pre.5
2.0.0-pre.4
2.0.0-pre.3
2.0.0-pre.2
2.0.0-pre.1
2.0.0-pre.0
1.53.3
1.53.2
1.53.0
1.52.0-rc.22
1.52.0-rc.21
1.52.0-rc.20
1.52.0-rc.19
1.52.0-rc.18
1.52.0-rc.17
1.52.0-rc.16
1.52.0-rc.15
1.52.0-rc.14
1.52.0-rc.13
1.52.0-rc.12
1.52.0-rc.11
1.52.0-rc.10
1.52.0-rc.9
1.52.0-rc.8
1.52.0-rc.7
1.52.0-rc.6
1.52.0-rc.5
1.52.0-rc.4
1.52.0-rc.3
1.52.0-rc.2
1.52.0-rc.1
1.52.0-rc.0
1.51.2
1.51.1
retired
1.51.0
1.50.21
1.50.20
1.50.19
1.50.18
1.50.17
1.50.16
1.50.15
1.50.14
1.50.13
1.50.12
1.50.11
1.50.10
1.50.9
1.50.8
1.50.7
1.50.6
1.50.5
1.50.4
1.50.3
1.50.2
1.50.1
1.50.0
1.49.0
1.48.0-rc.30
1.48.0-rc.29
1.48.0-rc.28
1.48.0-rc.27
1.48.0-rc.26
1.48.0-rc.25
1.48.0-rc.24
1.48.0-rc.23
1.48.0-rc.22
1.48.0-rc.21
1.48.0-rc.20
1.48.0-rc.19
1.48.0-rc.18
1.48.0-rc.17
1.48.0-rc.16
1.48.0-rc.15
1.48.0-rc.14
1.48.0-rc.13
1.48.0-rc.12
1.48.0-rc.11
1.48.0-rc.10
1.48.0-rc.9
1.48.0-rc.8
1.48.0-rc.7
1.48.0-rc.6
1.48.0-rc.5
1.48.0-rc.4
1.48.0-rc.3
1.48.0-rc.2
1.48.0-rc.1
1.48.0-rc.0
1.47.12
1.47.11
1.47.10
1.47.9
1.47.8
1.47.7
1.47.6
1.47.5
1.47.4
1.47.3
1.47.2
1.47.1
1.47.0
1.46.13
1.46.12
1.46.11
1.46.10
1.46.9
1.46.8
1.46.7
1.46.6
1.46.5
1.46.4
1.46.3
1.46.2
1.46.1
1.46.0
1.45.0-rc9
1.45.0-rc8
1.45.0-rc7
1.45.0-rc6
1.45.0-rc5
1.45.0-rc4
1.45.0-rc3
1.45.0-rc20
1.45.0-rc2
1.45.0-rc19
1.45.0-rc18
1.45.0-rc17
1.45.0-rc16
1.45.0-rc15
1.45.0-rc14
1.45.0-rc13
1.45.0-rc12
1.45.0-rc11
1.45.0-rc10
1.45.0-rc1
1.45.0-rc0
1.44.13
1.44.12
1.44.11
1.44.10
1.44.9
1.44.8
1.44.7
1.44.6
1.44.5
1.44.4
1.44.3
1.44.2
1.44.1
1.44.0
1.43.12
1.43.11
1.43.10
1.43.9
1.43.8
1.43.7
1.43.6
1.43.5
1.43.4
1.43.3
1.43.2
1.43.1
1.43.0
1.42.0
1.41.12
1.41.11
1.41.10
1.41.9
1.41.8
1.41.7
1.41.6
1.41.5
1.41.4
1.41.3
1.41.2
1.41.1
1.41.0
1.40.0
1.39.7
1.39.6
1.39.5
1.39.4
1.39.3
1.39.2
1.39.1
1.39.0
1.38.0
1.37.2
1.37.1
1.37.0
1.36.22
1.36.21
1.36.19
1.36.18
1.36.17
1.36.16
1.36.15
1.36.14
1.36.13
1.36.12
1.36.11
1.36.10
1.36.9
1.36.8
1.36.7
1.36.6
1.36.5
1.36.4
1.36.3
1.36.2
1.36.0
1.35.1
1.35.0
1.34.9
1.34.8
1.34.7
1.34.6
1.34.5
1.34.4
1.34.3
1.34.2
1.34.1
1.34.0
1.33.0
1.32.2
1.32.1
1.32.0
1.31.1
1.31.0
1.30.2
1.30.1
1.29.0-rc1
1.29.0-rc0
1.28.1
1.28.0
1.27.1
1.27.0
1.26.13
1.26.12
1.26.11
1.26.10
1.26.9
1.26.8
1.26.7
1.26.6
1.26.5
1.26.4
1.26.2
1.26.1
1.26.0
1.25.8
1.25.7
1.25.6
1.25.5
1.25.4
1.25.3
1.25.2
1.25.1
1.25.0
1.24.2
1.24.1
1.24.0
1.23.3
1.23.2
1.23.1
1.23.0
1.22.1
1.22.0
1.20.1
1.20.0
1.19.1
1.19.0
1.18.1
1.18.0
1.17.1
1.17.0
1.16.2
1.15.1
1.15.0
1.14.0
1.13.4
1.13.3
1.13.2
1.13.1
1.13.0
1.12.0
1.11.1
1.11.0
1.10.0
1.9.0
1.8.0
1.7.0
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.6.2
1.6.1
1.6.0
1.5.1
1.5.0
1.4.1
1.4.0
1.3.1
1.3.0
1.2.1
1.2.0
1.1.3
1.1.2
1.1.0
1.0.3
1.0.2
1.0.1
1.0.0
0.13.1
0.13.0
0.12.0
0.10.0
0.9.1
0.9.0
0.8.0
0.7.0
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.9
0.1.8
0.1.3
0.1.1
0.1.0
A declarative, extensible framework for building Elixir applications.
Current section
Files
Jump to
Current section
Files
lib/ash/data_layer/mnesia/mnesia.ex
# SPDX-FileCopyrightText: 2019 ash contributors <https://github.com/ash-project/ash/graphs/contributors>
#
# SPDX-License-Identifier: MIT
defmodule Ash.DataLayer.Mnesia do
@behaviour Ash.DataLayer
@mnesia %Spark.Dsl.Section{
name: :mnesia,
describe: """
A section for configuring the mnesia data layer
""",
examples: [
"""
mnesia do
table :custom_table
end
"""
],
schema: [
table: [
type: :atom,
doc: "The table name to use, defaults to the name of the resource"
]
]
}
@moduledoc """
An Mnesia backed Ash Datalayer.
In your application initialization, you will need to call `Mnesia.create_schema([node()])`.
Additionally, you will want to create your mnesia tables there.
This data layer is *unoptimized*, fetching all records from a table and filtering them
in memory. For that reason, it is not recommended to use it with large amounts of data. It can be
great for prototyping or light usage, though.
"""
use Spark.Dsl.Extension,
sections: [@mnesia],
verifiers: [Ash.DataLayer.Verifiers.RequirePreCheckWith]
alias Ash.Actions.Sort
alias :mnesia, as: Mnesia
require Logger
@doc """
Creates the table for each mnesia resource in a domain
"""
def start(domain, resources \\ []) do
Mnesia.create_schema([node()])
Mnesia.start()
Code.ensure_compiled(domain)
domain
|> Ash.Domain.Info.resources()
|> Enum.concat(resources)
|> Enum.filter(&(__MODULE__ in Spark.extensions(&1)))
|> Enum.flat_map(fn resource ->
resource
|> Ash.DataLayer.Mnesia.Info.table()
|> List.wrap()
end)
|> Enum.each(&Mnesia.create_table(&1, attributes: [:_pkey, :val]))
end
defmodule Query do
@moduledoc false
defstruct [
:domain,
:resource,
:filter,
:limit,
:tenant,
sort: [],
context: %{},
relationships: %{},
offset: 0,
aggregates: [],
calculations: []
]
end
@doc false
@impl true
def can?(_, :async_engine), do: true
def can?(_, :bulk_create), do: true
def can?(_, :multitenancy), do: true
def can?(_, :composite_primary_key), do: true
def can?(_, :upsert), do: true
def can?(_, :create), do: true
def can?(_, :read), do: true
def can?(_, :calculate), do: true
def can?(_, :update), do: true
def can?(_, :destroy), do: true
def can?(_, :sort), do: true
def can?(_, :filter), do: true
def can?(_, {:filter_relationship, _}), do: true
def can?(_, :expression_calculation), do: true
def can?(_, :expression_calculation_sort), do: true
def can?(_, :limit), do: true
def can?(_, :offset), do: true
def can?(_, :boolean_filter), do: true
def can?(_, :transact), do: true
def can?(_, :aggregate_filter), do: true
def can?(_, :aggregate_sort), do: true
def can?(_, :expr_error), do: true
def can?(_, {:aggregate_relationship, _}), do: true
def can?(_, {:aggregate, :count}), do: true
def can?(_, {:aggregate, :first}), do: true
def can?(_, {:aggregate, :sum}), do: true
def can?(_, {:aggregate, :list}), do: true
def can?(_, {:aggregate, :max}), do: true
def can?(_, {:aggregate, :min}), do: true
def can?(_, {:aggregate, :avg}), do: true
def can?(_, {:aggregate, :exists}), do: true
def can?(_, {:aggregate, :unrelated}), do: true
def can?(_, {:exists, :unrelated}), do: true
def can?(resource, {:query_aggregate, kind}), do: can?(resource, {:aggregate, kind})
case Application.compile_env(:ash, :no_join_mnesia_ets) || false do
false ->
def can?(_, {:join, _resource}) do
# we synthesize all filters under the hood using `Ash.Filter.Runtime`
true
end
true ->
def can?(_, {:join, _resource}) do
# we synthesize all filters under the hood using `Ash.Filter.Runtime`
false
end
:dynamic ->
def can?(_, {:join, resource}) do
Ash.Resource.Info.data_layer(resource) == __MODULE__ ||
Application.get_env(:ash, :mnesia_ets_join?, true)
end
end
def can?(_, {:filter_expr, _}), do: true
def can?(_, :nested_expressions), do: true
def can?(_, {:sort, _}), do: true
def can?(_, {:atomic, :update}), do: true
def can?(_, {:atomic, :upsert}), do: true
def can?(_, {:atomic, :create}), do: true
def can?(_, :through_relationship), do: true
def can?(_, _), do: false
@doc false
@impl true
def resource_to_query(resource, domain) do
%Query{
resource: resource,
domain: domain
}
end
@doc false
@impl true
def in_transaction?(_), do: Mnesia.is_transaction()
@doc false
@impl true
def limit(query, offset, _), do: {:ok, %{query | limit: offset}}
@doc false
@impl true
def offset(query, offset, _), do: {:ok, %{query | offset: offset}}
@doc false
@impl true
def filter(query, filter, _resource) do
if query.filter do
{:ok, %{query | filter: Ash.Filter.add_to_filter!(query.filter, filter)}}
else
{:ok, %{query | filter: filter}}
end
end
@doc false
@impl true
def sort(query, sort, _resource) do
{:ok, %{query | sort: sort}}
end
@doc false
@impl true
def add_aggregate(query, aggregate, _),
do: {:ok, %{query | aggregates: [aggregate | query.aggregates]}}
@doc false
@impl true
def add_calculations(query, calculations, _),
do: {:ok, %{query | calculations: query.calculations ++ calculations}}
@doc false
@impl true
def run_aggregate_query(%{domain: domain} = query, aggregates, resource) do
case run_query(query, resource) do
{:ok, results} ->
Enum.reduce_while(aggregates, {:ok, %{}}, fn
%{
kind: kind,
name: name,
query: query,
field: field,
resource: resource,
uniq?: uniq?,
include_nil?: include_nil?,
default_value: default_value
},
{:ok, acc} ->
results
|> filter_matches(
Map.get(query || %{}, :filter),
domain,
query.tenant,
query.context[:private][:actor]
)
|> case do
{:ok, matches} ->
field = field || Enum.at(Ash.Resource.Info.primary_key(resource), 0)
value =
Ash.DataLayer.Ets.aggregate_value(
matches,
kind,
field,
uniq?,
include_nil?,
default_value
)
{:cont, {:ok, Map.put(acc, name, value)}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
{:error, error} ->
{:error, error}
end
|> case do
{:error, error} ->
{:error, Ash.Error.to_ash_error(error)}
{:aborted, error} ->
{:error, error}
end
end
@doc false
@impl true
def set_tenant(_resource, query, tenant) do
{:ok, %{query | tenant: tenant}}
end
@doc false
@impl true
def set_context(_resource, query, context) do
{:ok, %{query | context: context}}
end
@doc false
@impl true
def run_query(
%Query{
domain: domain,
resource: resource,
filter: filter,
offset: offset,
calculations: calculations,
limit: limit,
sort: sort,
aggregates: aggregates,
tenant: tenant,
context: context
},
_resource
) do
with {:atomic, records} <-
Mnesia.transaction(fn ->
Mnesia.select(Ash.DataLayer.Mnesia.Info.table(resource), [{:_, [], [:"$_"]}])
end),
{:ok, records} <-
records |> Enum.map(&elem(&1, 2)) |> Ash.DataLayer.Ets.cast_records(resource),
{:ok, filtered} <-
filter_matches(records, filter, domain, tenant, context[:private][:actor]),
offset_records <-
filtered |> Sort.runtime_sort(sort, domain: domain) |> Enum.drop(offset || 0),
limited_records <- do_limit(offset_records, limit),
{:ok, records} <-
Ash.DataLayer.Ets.do_add_aggregates(
limited_records,
domain,
resource,
aggregates
),
{:ok, records} <-
Ash.DataLayer.Ets.do_add_calculations(
records,
resource,
calculations,
domain
) do
{:ok, records}
else
{:error, error} ->
{:error, error}
{:aborted, reason} ->
{:error, reason}
end
end
defp do_limit(records, nil), do: records
defp do_limit(records, limit), do: Enum.take(records, limit)
defp filter_matches(
records,
filter,
domain,
tenant,
actor,
parent \\ nil,
conflicting_upsert_values \\ nil
)
defp filter_matches([], _, _domain, _tenant, _actor, _parent, _conflicting_upsert_values),
do: {:ok, []}
defp filter_matches(
records,
nil,
_domain,
_tenant,
_actor,
_parent,
_conflicting_upsert_values
),
do: {:ok, records}
defp filter_matches(records, filter, domain, tenant, actor, parent, conflicting_upsert_values) do
Ash.Filter.Runtime.filter_matches(domain, records, filter,
parent: parent,
tenant: tenant,
actor: actor,
conflicting_upsert_values: conflicting_upsert_values
)
end
@doc """
Bulk create records in the database.
This function is used to create multiple records in a single transaction.
If you are NOT setting the `upsert? = true` option, this will be optimized by
creating a single transaction and bulk creating all of the entries. The way
:mnesia.write works, will effectively do an upsert, but you cannot control
which fields are updated and the only identity you are matching on is the
primary key.
If you are using an `upsert?` it will be unoptimized and will load all records
into memory before performing the upsert operation.
"""
@impl true
def bulk_create(resource, stream, options) do
stream = Enum.to_list(stream)
log_bulk_create(resource, stream, options)
if options[:upsert?] do
# This uses a lot of the ETS datalayer as a reference point. It is
# recommended to NOT use this and instead upsert on the pkey which should
# use the optmized approach in the `else` logic.
stream
|> Enum.reduce_while({:ok, []}, fn changeset, {:ok, results} ->
changeset =
Ash.Changeset.set_context(changeset, %{
private:
Map.merge(changeset.context[:private] || %{}, %{
upsert_fields: options[:upsert_fields] || [],
touch_update_defaults?: Map.get(options, :touch_update_defaults?, true)
})
})
case do_upsert(resource, changeset, options.upsert_keys, nil, true) do
{:ok, result} ->
result =
if options[:return_records?] do
Ash.Resource.put_metadata(
result,
:bulk_create_index,
changeset.context.bulk_create.index
)
else
result
end
{:cont, {:ok, [result | results]}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
|> case do
{:ok, results} ->
if options[:return_records?] do
{:ok, Enum.reverse(results)}
else
:ok
end
{:error, error} ->
{:error, error}
end
else
do_bulk_write(resource, stream, options)
|> case do
{:atomic, {:ok, results}} ->
if options[:return_records?] do
{:ok, Enum.reverse(results)}
else
:ok
end
{:aborted, {:error, error}} ->
{:error, error}
end
end
end
# Does a bulk write using a single transaction
defp do_bulk_write(resource, stream, options) do
Mnesia.transaction(fn ->
Enum.reduce_while(stream, {:ok, []}, fn changeset, {:ok, results} ->
# Sending in `false` prevents a transaction for every write
case create(resource, changeset, with_transaction: false, from_bulk_create?: true) do
{:ok, result} ->
result =
if options[:return_records?] do
Ash.Resource.put_metadata(
result,
:bulk_create_index,
changeset.context.bulk_create.index
)
else
result
end
{:cont, {:ok, [result | results]}}
{:error, error} ->
{:halt, {:error, error}}
end
end)
end)
end
@doc false
@impl true
def create(resource, changeset, opts \\ []) do
if !opts[:from_bulk_create?], do: log_create(resource, changeset)
with {:ok, record} <- Ash.Changeset.apply_attributes(changeset),
{:ok, record} <- apply_create_atomics(changeset, resource, record) do
pkey =
resource
|> Ash.Resource.Info.primary_key()
|> Enum.map(fn attr ->
Map.get(record, attr)
end)
resource
|> Ash.Resource.Info.attributes()
|> Map.new(&{&1.name, Map.get(record, &1.name)})
|> Ash.DataLayer.Ets.dump_to_native(Ash.Resource.Info.attributes(resource))
|> case do
{:ok, values} ->
case do_write(
Ash.DataLayer.Mnesia.Info.table(resource),
pkey,
values,
Keyword.get(opts, :with_transaction, true)
) do
# If with_transaction is false, we are in a transaction and will only return :ok
:ok ->
{:ok, %{record | __meta__: %Ecto.Schema.Metadata{state: :loaded, schema: resource}}}
{:atomic, _} ->
{:ok, %{record | __meta__: %Ecto.Schema.Metadata{state: :loaded, schema: resource}}}
{:aborted, error} ->
{:error, error}
end
{:error, error} ->
{:error, error}
end
end
end
defp apply_create_atomics(%{create_atomics: create_atomics} = changeset, resource, record)
when create_atomics != [] do
Enum.reduce_while(create_atomics, {:ok, record}, fn {key, expr}, {:ok, acc} ->
case Ash.Expr.eval(expr,
resource: resource,
record: acc,
domain: changeset.domain,
unknown_on_unknown_refs?: true
) do
{:ok, value} ->
{:cont, {:ok, Map.put(acc, key, value)}}
{:error, error} ->
{:halt, {:error, error}}
:unknown ->
{:halt, {:error, "Could not evaluate expression #{inspect(expr)}"}}
end
end)
end
defp apply_create_atomics(_changeset, _resource, record), do: {:ok, record}
# This allows for writing to Mnesia without a transaction in case one was
# started elsewhere. This was explicitly created for `bulk_create/3`.
defp do_write(table, pkey, values, with_transaction) do
if with_transaction && !Mnesia.is_transaction() do
Mnesia.transaction(fn ->
Mnesia.write({table, pkey, values})
end)
else
Mnesia.write({table, pkey, values})
end
end
@doc false
@impl true
def destroy(resource, %{data: record}) do
pkey =
resource
|> Ash.Resource.Info.primary_key()
|> Enum.map(&Map.get(record, &1))
result =
Mnesia.transaction(fn ->
Mnesia.delete({Ash.DataLayer.Mnesia.Info.table(resource), pkey})
end)
case result do
{:atomic, _} -> :ok
{:aborted, error} -> {:error, error}
end
end
@doc false
@impl true
def update(resource, changeset, from_bulk_create? \\ false) do
pkey = pkey_list(resource, changeset.data)
if !from_bulk_create? do
log_update(
resource,
Map.take(changeset.data, Ash.Resource.Info.primary_key(resource)),
changeset
)
end
result =
Mnesia.transaction(fn ->
with {:ok, record} <- Ash.Changeset.apply_attributes(%{changeset | action_type: :update}),
{:ok, record} <-
do_update(Ash.DataLayer.Mnesia.Info.table(resource), {pkey, record}, resource),
{:ok, record} <- Ash.DataLayer.Ets.cast_record(record, resource) do
new_pkey = pkey_list(resource, record)
if new_pkey != pkey do
case destroy(resource, changeset) do
:ok ->
{:ok,
%{record | __meta__: %Ecto.Schema.Metadata{state: :loaded, schema: resource}}}
{:error, error} ->
{:error, error}
end
else
{:ok, %{record | __meta__: %Ecto.Schema.Metadata{state: :loaded, schema: resource}}}
end
else
{:error, error} ->
{:error, error}
end
end)
case result do
{:atomic, {:error, error}} ->
{:error, error}
{:atomic, {:ok, result}} ->
{:ok, result}
{:aborted, {reason, stacktrace}} when is_exception(reason) ->
{:error, Ash.Error.to_ash_error(reason, stacktrace)}
{:aborted, reason} ->
{:error, reason}
end
end
@impl true
def calculate(resource, expressions, context) do
Enum.reduce_while(expressions, {:ok, []}, fn expression, {:ok, results} ->
case Ash.Expr.eval(expression, resource: resource, context: context) do
{:ok, result} -> {:cont, {:ok, [result | results]}}
{:error, error} -> {:halt, {:error, error}}
end
end)
|> case do
{:ok, results} -> {:ok, Enum.reverse(results)}
{:error, error} -> {:error, error}
end
end
defp pkey_list(resource, data) do
resource
|> Ash.Resource.Info.primary_key()
|> Enum.map(&Map.get(data, &1))
end
defp do_update(table, {pkey, record}, resource) do
attributes = Ash.Resource.Info.attributes(resource)
case Ash.DataLayer.Ets.dump_to_native(record, attributes) do
{:ok, casted} ->
case Mnesia.read({Ash.DataLayer.Mnesia.Info.table(resource), pkey}) do
[] ->
{:error, "Record not found matching: #{inspect(pkey)}"}
[{_, _, record}] ->
Mnesia.write({table, pkey, Map.merge(record, casted)})
[{_, _, record}] = Mnesia.read({table, pkey})
{:ok, record}
end
{:error, error} ->
{:error, error}
end
end
@doc false
@impl true
def upsert(resource, changeset, keys, identity \\ nil) do
do_upsert(resource, changeset, keys, identity)
end
defp do_upsert(resource, changeset, keys, identity, from_bulk_create? \\ false) do
keys = keys || Ash.Resource.Info.primary_key(resource)
if (is_nil(identity) || !identity.nils_distinct?) &&
Enum.any?(keys, &is_nil(Ash.Changeset.get_attribute(changeset, &1))) do
create(resource, changeset, from_bulk_create?: from_bulk_create?)
else
key_filters =
Enum.map(keys, fn key ->
value =
Ash.Changeset.get_attribute(changeset, key) || Map.get(changeset.params, key) ||
Map.get(changeset.params, to_string(key))
{key,
if is_nil(value) do
[is_nil: true]
else
value
end}
end)
query =
resource
|> Ash.Query.do_filter(and: [key_filters])
|> then(fn query ->
if is_nil(identity) || is_nil(identity.where) do
query
else
Ash.Query.do_filter(query, identity.where)
end
end)
resource
|> resource_to_query(changeset.domain)
|> Map.put(:filter, query.filter)
|> Map.put(:tenant, changeset.tenant)
|> run_query(resource)
|> case do
{:ok, []} ->
resource
|> create(changeset, from_bulk_create?: from_bulk_create?)
|> set_upsert_action(:create)
{:ok, [result]} ->
with {:ok, conflicting_upsert_values} <- Ash.Changeset.apply_attributes(changeset),
{:ok, [^result]} <-
upsert_conflict_check(
changeset,
result,
conflicting_upsert_values
) do
to_set =
changeset
|> Ash.Changeset.set_on_upsert(keys)
|> apply_upsert_update_defaults(resource, result, changeset)
changeset =
changeset
|> Map.put(:attributes, %{})
|> Map.put(:data, result)
|> Ash.Changeset.force_change_attributes(to_set)
resource
|> update(%{changeset | action_type: :update, filter: nil}, from_bulk_create?)
|> set_upsert_action(:update)
else
{:ok, []} ->
{:ok, Ash.Resource.put_metadata(result, :upsert_skipped, true)}
{:error, reason} ->
{:error, reason}
end
{:ok, _} ->
{:error, "Multiple records matching keys"}
end
end
end
defp set_upsert_action({:ok, record}, action) do
{:ok, Ash.Resource.put_metadata(record, :upsert_action, action)}
end
defp set_upsert_action(result, _), do: result
@spec upsert_conflict_check(
changeset :: Ash.Changeset.t(),
subject :: record,
conflicting_upsert_values :: record
) :: {:ok, [record]} | {:error, reason}
when record: Ash.Resource.Record.t(), reason: term()
defp upsert_conflict_check(changeset, subject, conflicting_upsert_values)
defp upsert_conflict_check(
%Ash.Changeset{filter: nil},
result,
_conflicting_upsert_values
),
do: {:ok, [result]}
defp upsert_conflict_check(
%Ash.Changeset{filter: filter, domain: domain, context: context},
result,
conflicting_upsert_values
) do
filter_matches(
[result],
filter,
domain,
context.private[:tenant],
context.private[:actor],
nil,
conflicting_upsert_values
)
end
# Mnesia's update/2 calls apply_attributes which re-applies update_defaults
# via set_defaults/3. To prevent unwanted updates, we preserve existing
# values from the record for update_default fields so set_defaults sees
# them as already set and skips them.
defp apply_upsert_update_defaults(to_set, resource, existing_record, changeset) do
touch_update_defaults? =
changeset.context[:private][:touch_update_defaults?]
if touch_update_defaults? == false || to_set == [] do
upsert_fields = changeset.context[:private][:upsert_fields]
update_default_attrs =
resource
|> Ash.Resource.Info.attributes()
|> Enum.filter(& &1.update_default)
Enum.reduce(update_default_attrs, to_set, fn attr, acc ->
if explicitly_set?(attr.name, upsert_fields, changeset) do
acc
else
Keyword.put(acc, attr.name, Map.get(existing_record, attr.name))
end
end)
else
to_set
end
end
defp explicitly_set?(key, upsert_fields, _changeset) when is_list(upsert_fields),
do: key in upsert_fields
defp explicitly_set?(key, _, changeset),
do: Map.has_key?(changeset.attributes, key) && key not in Map.get(changeset, :defaults, [])
defp log_bulk_create(resource, stream, options) do
Logger.debug(
"Mnesia: #{bulk_create_operation(options, stream)} #{inspect(resource)}:\n\n#{Enum.map_join(stream, "\n", &format_changes(&1, from_bulk_create?: true))}"
)
end
defp bulk_create_operation(
%{
upsert?: true,
upsert_keys: upsert_keys,
upsert_fields: upsert_fields,
upsert_where: expr
},
stream
) do
where_expr =
if is_nil(expr) do
""
else
"where #{inspect(expr)}"
end
"Upserting #{Enum.count(stream)} on #{inspect(upsert_keys)} #{where_expr}, setting #{inspect(List.wrap(upsert_fields))}"
end
defp bulk_create_operation(_options, stream) do
"Creating #{Enum.count(stream)}"
end
defp log_create(resource, changeset) do
Logger.debug("""
Mnesia: Creating #{inspect(resource)}:
#{format_changes(changeset)}
""")
end
defp log_update(resource, pkey, changeset) do
pkey =
if Enum.count_until(pkey, 2) == 2 do
inspect(pkey)
else
inspect(pkey |> Enum.at(0) |> elem(1))
end
Logger.debug("""
Mnesia: Updating #{inspect(resource)} #{pkey}:
#{format_changes(changeset)}
""")
end
defp format_changes(changeset, opts \\ []) do
prefix =
if opts[:from_bulk_create?] do
""
else
"Setting "
end
inspect_opts = %Inspect.Opts{
limit: 10,
printable_limit: 36,
pretty: true
}
doc =
Inspect.Algebra.container_doc(
"%{",
Enum.uniq_by(Enum.to_list(changeset.attributes) ++ changeset.atomics, &elem(&1, 0)),
"}",
inspect_opts,
fn {k, v}, _opts ->
v =
if Ash.Expr.expr?(v) do
Ash.Filter.map(v, fn nested ->
truncate_unless_expr(nested, inspect_opts)
end)
|> inspect(
limit: inspect_opts.limit,
printable_limit: inspect_opts.printable_limit,
pretty: true
)
else
truncate_inspect(v, inspect_opts)
end
# Create the value document from the truncated string
value_doc = Inspect.Algebra.string(v)
Inspect.Algebra.concat([to_string(k), ": ", value_doc])
end,
separator: ",",
break: :flex
)
# Format the document to a string with proper line breaks
result = Inspect.Algebra.format(doc, inspect_opts.width) |> IO.iodata_to_binary()
prefix <> result
rescue
e ->
"Failed to format changes: #{Exception.message(e)}"
end
defp truncate_unless_expr(nested, inspect_opts) do
if Ash.Expr.expr?(nested) do
nested
else
cond do
is_atom(nested) ->
nested
is_binary(nested) ->
truncate(nested, inspect_opts)
is_map(nested) ->
Map.new(nested, fn {k, v} -> {k, truncate_unless_expr(v, inspect_opts)} end)
is_list(nested) ->
Enum.map(nested, fn v -> truncate_unless_expr(v, inspect_opts) end)
is_tuple(nested) ->
nested
|> Tuple.to_list()
|> Enum.map(&truncate_unless_expr(&1, inspect_opts))
|> List.to_tuple()
true ->
truncate_inspect(nested, inspect_opts)
end
end
end
defp truncate_inspect(v, inspect_opts) do
value_str =
inspect(v,
limit: inspect_opts.limit,
printable_limit: inspect_opts.printable_limit,
pretty: true
)
truncate(value_str, inspect_opts)
end
defp truncate(value_str, inspect_opts) do
if String.length(value_str) > inspect_opts.printable_limit + 3 do
String.slice(value_str, 0, inspect_opts.printable_limit) <>
"..."
else
value_str
end
end
@doc false
@impl true
def transaction(_, func, _timeout, _reason) do
case Mnesia.transaction(func, 1) do
{:atomic, result} ->
{:ok, result}
{:aborted, {reason, stacktrace}} when is_exception(reason) ->
{:error, Ash.Error.to_ash_error(reason, stacktrace)}
{:aborted, %Ash.Changeset{} = changeset} ->
{:error, changeset}
{:aborted, %Ash.Query{} = query} ->
{:error, query}
{:aborted, reason} ->
if Ash.Error.ash_error?(reason) do
{:error, reason}
else
{:error, Ash.Error.to_ash_error(Exception.format_exit(reason))}
end
end
end
@doc false
@impl true
@spec rollback(term, term) :: no_return
def rollback(_, value) do
Mnesia.abort(value)
end
end