diff --git a/lib/data_layer.ex b/lib/data_layer.ex index ec44534..7b6cf27 100644 --- a/lib/data_layer.ex +++ b/lib/data_layer.ex @@ -307,6 +307,9 @@ defmodule AshSqlite.DataLayer do AshSqlite.Transformers.ValidateReferences, AshSqlite.Transformers.VerifyRepo, AshSqlite.Transformers.EnsureTableOrPolymorphic + ], + verifiers: [ + AshSqlite.Verifiers.VerifyTemporal ] def migrate(args) do @@ -440,6 +443,15 @@ defmodule AshSqlite.DataLayer do import Ecto.Query, only: [from: 2] + # SQLite has no range type, so a range attribute is stored as JSON text. Ash + # builds its Ecto schema from this callback when the data layer answers it, so + # this is what makes the *write* path dump a range -- the expression seam in + # `AshSqlite.SqlImplementation` only covers reads. + @impl true + def attribute_ecto_type(_resource, %{type: Ash.Type.Range}), do: AshSqlite.Type.Range + + def attribute_ecto_type(_resource, _attribute), do: nil + @impl true def can?(_, :async_engine), do: false def can?(_, :bulk_create), do: true @@ -449,6 +461,10 @@ defmodule AshSqlite.DataLayer do def can?(resource, :transact), do: AshSqlite.DataLayer.Info.write_transactions?(resource) def can?(_, :composite_primary_key), do: true + # SQLite has no `FOR PORTION OF` and no exclusion constraint, so a period split is a + # read-modify-write in `AshSqlite.Temporal` rather than one statement. The semantics + # are the same; the enforcement underneath differs. See `AshSqlite.Temporal`. + def can?(_, :temporal), do: AshSqlite.Temporal.supported?() def can?(_, {:atomic, :update}), do: true def can?(_, {:atomic, :upsert}), do: true def can?(_, {:atomic, :create}), do: true @@ -628,18 +644,103 @@ defmodule AshSqlite.DataLayer do ] end + # A point-in-time read is a containment test on the period. The bounds stay in the + # JSON and the planner matches the index built on the same `json_extract` expression, + # so this is an index seek for a keyed read rather than a scan. + # + # `set_as_of/3` is not a callback of a released ash, so defining it unconditionally + # would put `@impl true` on a behaviour that does not declare it. + if AshSqlite.Temporal.supported?() do + @impl true + def set_as_of(resource, query, as_of) do + as_of = Ash.Temporal.resolve_as_of(as_of) + + if Ash.Resource.Info.temporal_strategy(resource) == :context && as_of do + attribute = AshSqlite.Temporal.attribute(resource) + encoded = AshSqlite.Temporal.encode(as_of) + + {:ok, + from(row in query, + where: + fragment("json_extract(?, '$.lower') <= ?", field(row, ^attribute), ^encoded) and + (fragment("json_extract(?, '$.upper') IS NULL", field(row, ^attribute)) or + fragment("json_extract(?, '$.upper') > ?", field(row, ^attribute), ^encoded)) + )} + else + {:ok, query} + end + end + else + defp set_as_of(_resource, query, _as_of), do: {:ok, query} + end + @impl true def resource_to_query(resource, _) do from(row in {AshSqlite.DataLayer.Info.table(resource) || "", resource}, []) end @impl true - def bulk_create(resource, stream, options) do + # The temporal arm is compiled out entirely against a released ash, where + # `temporal?/1` is a constant false and the branch is dead code. + if AshSqlite.Temporal.supported?() do + def bulk_create(resource, stream, options) do + if options[:upsert?] && AshSqlite.Temporal.temporal?(resource) do + temporal_bulk_upsert(resource, stream, options) + else + do_bulk_create(resource, stream, options) + end + end + else + def bulk_create(resource, stream, options), do: do_bulk_create(resource, stream, options) + end + + if AshSqlite.Temporal.supported?() do + # SQLite cannot `ON CONFLICT` its way to a temporal upsert (see `temporal_upsert/3`), + # and `bulk_create/3` builds one whenever `upsert?` is set. Each changeset is resolved + # on its own instead, inside one transaction. + defp temporal_bulk_upsert(resource, stream, options) do + changesets = Enum.to_list(stream) + repo = AshSql.dynamic_repo(resource, AshSqlite.SqlImplementation, Enum.at(changesets, 0)) + keys = options[:upsert_keys] || Ash.Resource.Info.primary_key(resource) + + AshSqlite.Temporal.transactionally(repo, fn -> + changesets + |> Enum.reduce_while({:ok, []}, fn changeset, {:ok, acc} -> + case do_temporal_upsert(repo, resource, changeset, keys, options[:upsert_fields]) do + # Ash correlates a bulk result to its changeset by this metadata, and drops a + # record that does not carry it. + {:ok, record} -> + {:cont, {:ok, [tag_bulk_record(record, changeset) | acc]}} + + {:error, error} -> + {:halt, {:error, error}} + end + end) + |> case do + {:ok, records} -> + if options[:return_records?], do: {:ok, Enum.reverse(records)}, else: {:ok, []} + + other -> + other + end + end) + end + + defp tag_bulk_record(record, changeset) do + case changeset.context do + %{bulk_create: %{ref: ref}} -> Ash.Resource.put_metadata(record, :bulk_action_ref, ref) + _other -> record + end + end + end + + defp do_bulk_create(resource, stream, options) do # Cell-wise default values are not supported on INSERT statements by SQLite # This requires that we group changesets by what attributes are changing # And *omit* any defaults instead of using something like `(1, 2, DEFAULT)` # like we could with postgres stream + |> Enum.map(&stamp_period(resource, &1)) |> Enum.group_by(&Map.keys(&1.attributes)) |> Enum.reduce_while({:ok, []}, fn {_, changesets}, {:ok, acc} -> repo = AshSql.dynamic_repo(resource, AshSqlite.SqlImplementation, Enum.at(changesets, 0)) @@ -905,6 +1006,22 @@ defmodule AshSqlite.DataLayer do end end + # A temporal create establishes the period `as_of` names. The period attribute is + # never accepted as action input, so this is the only place it is set on a create. + defp stamp_period(resource, changeset) do + if AshSqlite.Temporal.temporal?(resource) do + attribute = AshSqlite.Temporal.attribute(resource) + as_of = AshSqlite.Temporal.write_instant(resource, changeset) + period = AshSqlite.Temporal.period(resource, as_of) + + # `put_new`: a temporal upsert that misses has already bounded the period above + # by the version that follows it, and must not have that replaced by `[as_of, oo)`. + %{changeset | attributes: Map.put_new(changeset.attributes, attribute, period)} + else + changeset + end + end + defp ecto_changeset(record, changeset, type, table_error?) do filters = if changeset.action_type == :create do @@ -1345,6 +1462,95 @@ defmodule AshSqlite.DataLayer do @impl true def upsert(resource, changeset, keys \\ nil) do + if AshSqlite.Temporal.temporal?(resource) do + temporal_upsert(resource, changeset, keys || Ash.Resource.Info.primary_key(resource)) + else + do_upsert(resource, changeset, keys) + end + end + + # SQLite cannot `ON CONFLICT` its way to a temporal upsert. The only unique index on a + # temporal table is partial -- one current version per key -- and SQLite will not use a + # partial index as a conflict target unless the statement repeats its WHERE clause. So + # the match is resolved first and the two halves are issued separately, which is also + # what Postgres ends up doing for a different reason: it cannot `ON CONFLICT` against + # the `WITHOUT OVERLAPS` exclusion constraint either. + defp temporal_upsert(resource, changeset, keys) do + repo = AshSql.dynamic_repo(resource, AshSqlite.SqlImplementation, changeset) + + AshSqlite.Temporal.transactionally(repo, fn -> + do_temporal_upsert(repo, resource, changeset, keys) + end) + end + + defp do_temporal_upsert(repo, resource, changeset, keys) do + do_temporal_upsert( + repo, + resource, + changeset, + keys, + changeset.context[:private][:upsert_fields] + ) + end + + defp do_temporal_upsert(repo, resource, changeset, keys, upsert_fields) do + attribute = AshSqlite.Temporal.attribute(resource) + as_of = AshSqlite.Temporal.write_instant(resource, changeset) + pkey = Map.new(keys, fn key -> {key, Ash.Changeset.get_attribute(changeset, key)} end) + + case AshSqlite.Temporal.current_version(repo, resource, pkey, as_of) do + nil -> + upper = AshSqlite.Temporal.next_lower_bound(repo, resource, pkey, as_of) + period = AshSqlite.Temporal.period(resource, as_of, upper) + + create(resource, %{ + changeset + | attributes: Map.put(changeset.attributes, attribute, period) + }) + + {rowid, prior} -> + # The rowid of the version the upsert now applies to: the prior one when the + # write landed on its own lower bound, otherwise the copy opened at `as_of`. + rowid = + case AshSqlite.Temporal.close_version(repo, resource, rowid, prior, as_of) do + :replace -> + rowid + + {:closed, prior} -> + new_period = AshSqlite.Temporal.period(resource, as_of, prior.upper) + AshSqlite.Temporal.copy_forward(repo, resource, rowid, new_period) + end + + attributes = + case upsert_fields do + nil -> changeset.attributes + fields -> Map.take(changeset.attributes, fields) + end + + # The record is read back rather than built from the changeset, so the + # attributes the upsert did not name come from the row instead of the struct's + # defaults. + do_update(resource, %{ + changeset + | action_type: :update, + action_select: action_select(resource, changeset), + attributes: Map.drop(attributes, [attribute] ++ keys), + data: AshSqlite.Temporal.version_at(repo, resource, rowid) + }) + end + end + + # A bulk create carries no action select, and an update that selects nothing is an + # Ecto error rather than an update returning nothing. + defp action_select(resource, changeset) do + case changeset.action_select do + nil -> Enum.map(Ash.Resource.Info.attributes(resource), & &1.name) + [] -> Enum.map(Ash.Resource.Info.attributes(resource), & &1.name) + selected -> selected + end + end + + defp do_upsert(resource, changeset, keys) do keys = keys || Ash.Resource.Info.primary_key(keys) touch_update_defaults? = @@ -1493,6 +1699,18 @@ defmodule AshSqlite.DataLayer do @impl true def update(resource, changeset) do + if AshSqlite.Temporal.temporal?(resource) do + repo = AshSql.dynamic_repo(resource, AshSqlite.SqlImplementation, changeset) + + AshSqlite.Temporal.transactionally(repo, fn -> + AshSqlite.Temporal.update(repo, resource, changeset, &do_update(resource, &1)) + end) + else + do_update(resource, changeset) + end + end + + defp do_update(resource, changeset) do source = resolve_source(resource, changeset) query = @@ -1538,17 +1756,32 @@ defmodule AshSqlite.DataLayer do end end + # For a temporal resource the row identity is the primary key *plus* the period -- + # the key alone names every version of the record, so filtering by it would aim a + # single-row update at the whole history. defp pkey_filter(query, %resource{} = record) do pkey = record - |> Map.take(Ash.Resource.Info.primary_key(resource)) + |> Map.take(AshSqlite.Temporal.identity_fields(resource)) |> Map.to_list() Ecto.Query.where(query, ^pkey) end @impl true - def destroy(resource, %{data: record} = changeset) do + def destroy(resource, changeset) do + if AshSqlite.Temporal.temporal?(resource) do + repo = AshSql.dynamic_repo(resource, AshSqlite.SqlImplementation, changeset) + + AshSqlite.Temporal.transactionally(repo, fn -> + AshSqlite.Temporal.destroy(repo, resource, changeset, &do_destroy_changeset(resource, &1)) + end) + else + do_destroy_changeset(resource, changeset) + end + end + + defp do_destroy_changeset(resource, %{data: record} = changeset) do source = resolve_source(resource, changeset) query = @@ -1625,6 +1858,61 @@ defmodule AshSqlite.DataLayer do def update_query(query, changeset, resource, options) do repo = AshSql.dynamic_repo(resource, AshSqlite.SqlImplementation, changeset) + if AshSqlite.Temporal.temporal?(resource) do + AshSqlite.Temporal.transactionally(repo, fn -> + {:ok, query} = temporal_split(repo, query, changeset, resource) + do_update_query(query, changeset, resource, options, repo) + end) + else + do_update_query(query, changeset, resource, options, repo) + end + end + + # The instant a write against a query takes effect. The query's own pin wins, because + # Ash has already applied it as a containment filter and a split at any other instant + # produces versions that filter matches neither of. `AshPostgres.DataLayer` resolves + # the bound the same way, for the same reason. + # Both arms go through `write_instant/2`, which casts to the resource's own period + # type. The query's pin arrives as a `DateTime` whatever the resource stores, so a + # `:date` period would otherwise be split at a timestamp and stop comparing against + # its own bounds. + defp write_instant_for(query, resource, changeset) do + case get_in(query.__ash_bindings__, [:context, :private, :as_of]) do + nil -> AshSqlite.Temporal.write_instant(resource, changeset) + as_of -> AshSqlite.Temporal.write_instant(resource, as_of) + end + end + + # Splits every version the query matches, then runs the caller's statement against a + # query pinned to `as_of`. After the split that query matches exactly the new + # versions, so the statement needs no knowledge that a split happened. + defp temporal_split(repo, query, changeset, resource) do + as_of = write_instant_for(query, resource, changeset) + {:ok, query} = set_as_of(resource, query, as_of) + + repo + |> matching_rowids(query) + |> then(&AshSqlite.Temporal.split_all(repo, resource, &1, as_of)) + + {:ok, query} + end + + # The rowids a data layer query matches. `rowid` is SQLite's own row identity and is + # stable within a statement, which is what the split needs -- the primary key is not + # unique on a temporal table, and the period is the thing being rewritten. + # + # Only `:select` is dropped, and only because it is being replaced. A `:limit` must + # survive: Ash puts one on a bulk update's query, and splitting past it would rewrite + # the periods of rows the update then leaves alone. `:order_by` survives with it, + # because which rows a limit selects depends on it. + defp matching_rowids(repo, query) do + query + |> Ecto.Query.exclude(:select) + |> Ecto.Query.select([row], fragment("rowid")) + |> repo.all() + end + + defp do_update_query(query, changeset, resource, options, repo) do ecto_changeset = case changeset.data do %Ash.Changeset.OriginalDataNotAvailable{} -> @@ -1750,6 +2038,36 @@ defmodule AshSqlite.DataLayer do def destroy_query(query, changeset, resource, options) do repo = AshSql.dynamic_repo(resource, AshSqlite.SqlImplementation, changeset) + if AshSqlite.Temporal.temporal?(resource) do + AshSqlite.Temporal.transactionally(repo, fn -> + temporal_destroy_query(repo, query, changeset, resource, options) + end) + else + do_destroy_query(query, changeset, resource, options, repo) + end + end + + # A temporal destroy ends validity rather than removing rows. Only a version whose + # period begins exactly at `as_of` is deleted, because truncating it would leave a + # period of zero width rather than a shortened history. + defp temporal_destroy_query(repo, query, changeset, resource, options) do + as_of = write_instant_for(query, resource, changeset) + {:ok, pinned} = set_as_of(resource, query, as_of) + + case repo + |> matching_rowids(pinned) + |> then(&AshSqlite.Temporal.truncate_all(repo, resource, &1, as_of)) do + [] -> + if options[:return_records?], do: {:ok, []}, else: :ok + + rowids -> + pinned + |> Ecto.Query.where([row], fragment("rowid") in ^rowids) + |> do_destroy_query(changeset, resource, options, repo) + end + end + + defp do_destroy_query(query, changeset, resource, options, repo) do ecto_changeset = case changeset.data do %Ash.Changeset.OriginalDataNotAvailable{} -> @@ -1930,8 +2248,13 @@ defmodule AshSqlite.DataLayer do case root_query_result do {:ok, root_query, acc, selected_atomics?} -> + # On a temporal resource the primary key names every version of a + # record, so joining on it alone makes the outer UPDATE hit the closed + # ones too and falsifies history. The period completes the identity, as + # it does in `pkey_filter/2`. dynamic = - Enum.reduce(Ash.Resource.Info.primary_key(resource), nil, fn pkey, dynamic -> + Enum.reduce(AshSqlite.Temporal.identity_fields(resource), nil, fn pkey, + dynamic -> if dynamic do Ecto.Query.dynamic( [row, joining], diff --git a/lib/migration_generator/migration_generator.ex b/lib/migration_generator/migration_generator.ex index 067a445..d008c04 100644 --- a/lib/migration_generator/migration_generator.ex +++ b/lib/migration_generator/migration_generator.ex @@ -2227,8 +2227,8 @@ defmodule AshSqlite.MigrationGenerator do defp do_snapshot(resource, table) do snapshot = %{ - attributes: attributes(resource, table), - identities: identities(resource), + attributes: resource |> attributes(table) |> without_temporal_primary_key(resource), + identities: temporal_identities(resource), table: table || AshSqlite.DataLayer.Info.table(resource), custom_indexes: custom_indexes(resource), custom_statements: custom_statements(resource), @@ -2247,6 +2247,32 @@ defmodule AshSqlite.MigrationGenerator do Map.put(snapshot, :hash, hash) end + # A temporal table holds one row per period, so the primary key cannot be unique on + # its own and there is nothing to promote to a composite key: SQLite has no + # `WITHOUT OVERLAPS`. `AshSqlite.Temporal.Migration` emits the uniqueness the key + # does have -- one current version per key -- as a partial unique index instead. + # + # Leaving `primary_key?` set would also make an integer key SQLite's rowid alias, and + # every `WHERE rowid = ?` in `AshSqlite.Temporal` would then be addressing the key. + defp without_temporal_primary_key(attributes, resource) do + if AshSqlite.Temporal.temporal?(resource) do + Enum.map(attributes, &Map.put(&1, :primary_key?, false)) + else + attributes + end + end + + # An identity on a temporal resource is unique per *period*, not per table, so a plain + # unique index over its keys would make history impossible. They become the same + # partial index and trigger pair the primary key gets. + defp temporal_identities(resource) do + if AshSqlite.Temporal.temporal?(resource) do + [] + else + identities(resource) + end + end + defp has_create_action?(resource) do resource |> Ash.Resource.Info.actions() @@ -2262,11 +2288,17 @@ defmodule AshSqlite.MigrationGenerator do end defp custom_statements(resource) do - resource - |> AshSqlite.DataLayer.Info.custom_statements() - |> Enum.map(fn custom_statement -> - Map.take(custom_statement, AshSqlite.Statement.fields()) - end) + written = + resource + |> AshSqlite.DataLayer.Info.custom_statements() + |> Enum.map(fn custom_statement -> + Map.take(custom_statement, AshSqlite.Statement.fields()) + end) + + # A temporal table's indexes and triggers are not optional, and the resource does + # not write them, so they are generated. They go first: a statement the user wrote + # may depend on them, and nothing here depends on a user statement. + AshSqlite.Temporal.Migration.statements(resource) ++ written end defp multitenancy(resource) do @@ -2353,7 +2385,8 @@ defmodule AshSqlite.MigrationGenerator do end) if attribute.source == source_attribute_name && relationship.type == :belongs_to && - foreign_key?(relationship) do + foreign_key?(relationship) && + not temporal_destination?(relationship) do configured_reference = configured_reference(resource, table, attribute.source || attribute.name, relationship) @@ -2392,6 +2425,32 @@ defmodule AshSqlite.MigrationGenerator do end) end + # A temporal table has no unique key: it holds one row per period, so the primary key + # identifies a record rather than a row. SQLite requires the parent column of a + # foreign key to be a primary key or to carry a non-partial UNIQUE index, and there is + # no index that could satisfy it here without making a second version impossible. + # The table would still be created -- SQLite defers the check -- and then every insert + # into the child would fail with "foreign key mismatch". + # + # So no database foreign key is generated. `AshPostgres` does the same when only one + # side of a temporal relationship is temporal; it can do better when both are, because + # PG19 has `FOREIGN KEY (fk, PERIOD src) REFERENCES dest (pk, PERIOD dest)` and SQLite + # has no equivalent. The relationship itself still works: the overlap filter + # `Ash.Resource.Transformers.AddTemporalRelationshipFilters` bakes into + # `relationship.filter` is applied by every consumer, data layer included. + # + # Asking the destination about its temporality is safe here. The generator runs as a + # mix task, not at compile time, which is why the core transformer avoids the same + # question. + if AshSqlite.Temporal.supported?() do + defp temporal_destination?(relationship) do + Code.ensure_loaded?(relationship.destination) and + Ash.Resource.Info.temporal?(relationship.destination) + end + else + defp temporal_destination?(_relationship), do: false + end + defp configured_reference(resource, table, attribute, relationship) do ref = resource @@ -2541,6 +2600,8 @@ defmodule AshSqlite.MigrationGenerator do defp migration_type(Ash.Type.CiString, _), do: :citext defp migration_type(Ash.Type.UUID, _), do: :uuid + # A range is stored as JSON text; SQLite has no range type. + defp migration_type(Ash.Type.Range, _), do: :text defp migration_type(Ash.Type.Integer, _), do: :bigint defp migration_type(other, constraints) do diff --git a/lib/sql_implementation.ex b/lib/sql_implementation.ex index 597c992..b314901 100644 --- a/lib/sql_implementation.ex +++ b/lib/sql_implementation.ex @@ -284,6 +284,115 @@ defmodule AshSqlite.SqlImplementation do handle_map_comparison(query, :==, left, right, pred_embedded?, bindings, embedded?, acc, type) end + # Range functions. SQLite has no range type, so `AshSqlite.Type.Range` stores a + # range as JSON text and every range predicate below is comparisons on + # `json_extract` of its keys. Two consequences worth knowing before editing: + # + # * The comparison is the inner type's *stored* comparison, which for + # datetimes is lexicographic on ISO8601. Both operands must therefore be + # encoded by `AshSqlite.Type.RangeBound`, never by the adapter's codec -- + # see that module for what goes silently wrong otherwise. + # * There is no index behind any of this. Postgres answers overlap from a + # GiST index; here it is a scan. A `json_extract` expression index can back + # the bound comparisons, but it cannot make overlap itself indexable. + def expr( + query, + %Ash.Query.Function.RangeOverlaps{arguments: [left, right], embedded?: pred_embedded?}, + bindings, + embedded?, + acc, + _type + ) do + {[left_type, right_type], _} = + determine_types(Ash.Query.Function.RangeOverlaps, [left, right], :boolean) + + range_fragment( + query, + bindings, + pred_embedded? || embedded?, + acc, + [{left, left_type}, {right, right_type}], + fn [l, r] -> overlaps_sql(l, r) end + ) + end + + def expr( + query, + %Ash.Query.Function.RangeContains{arguments: [left, right], embedded?: pred_embedded?}, + bindings, + embedded?, + acc, + _type + ) do + {[left_type, right_type], _} = + determine_types(Ash.Query.Function.RangeContains, [left, right], :boolean) + + if range_argument?(right) do + range_fragment( + query, + bindings, + pred_embedded? || embedded?, + acc, + [{left, left_type}, {right, right_type}], + fn [l, r] -> contains_range_sql(l, r) end + ) + else + # A point is compared against the range's bounds, so it must be encoded by + # the same encoder the bounds were. + range_fragment( + query, + bindings, + pred_embedded? || embedded?, + acc, + [{left, left_type}, point_operand(right)], + fn [l, point] -> contains_point_sql(l, point) end + ) + end + end + + def expr( + query, + %Ash.Query.Function.RangeAdjacent{arguments: [left, right], embedded?: pred_embedded?}, + bindings, + embedded?, + acc, + _type + ) do + {[left_type, right_type], _} = + determine_types(Ash.Query.Function.RangeAdjacent, [left, right], :boolean) + + range_fragment( + query, + bindings, + pred_embedded? || embedded?, + acc, + [{left, left_type}, {right, right_type}], + fn [l, r] -> adjacent_sql(l, r) end + ) + end + + def expr( + query, + %Ash.Query.Function.RangeLower{arguments: [range], embedded?: pred_embedded?}, + bindings, + embedded?, + acc, + type + ) do + range_bound_expr(query, range, "lower", bindings, pred_embedded? || embedded?, acc, type) + end + + def expr( + query, + %Ash.Query.Function.RangeUpper{arguments: [range], embedded?: pred_embedded?}, + bindings, + embedded?, + acc, + type + ) do + range_bound_expr(query, range, "upper", bindings, pred_embedded? || embedded?, acc, type) + end + # `is_distinct_from` is the NULL-safe form of `!=`, and Ash emits it in place of `!=` whenever # either side can be nil (see `Ash.Query.Function.IsDistinctFrom.new/1`). It needs the same # JSON treatment as the two clauses above, and without it a map reaches the driver as a bare @@ -385,6 +494,247 @@ defmodule AshSqlite.SqlImplementation do :error end + defp range_fragment(query, bindings, embedded?, acc, operands, builder) do + {dynamics, acc} = + Enum.map_reduce(operands, acc, fn {operand, type}, acc -> + AshSql.Expr.dynamic_expr(query, operand, bindings, embedded?, type, acc) + end) + + {expr, acc} = + AshSql.Expr.dynamic_expr( + query, + %Ash.Query.Function.Fragment{ + embedded?: embedded?, + arguments: merge_raw_parts(builder.(dynamics)) + }, + bindings, + embedded?, + nil, + acc + ) + + {:ok, expr, acc} + end + + # The value this yields is the bound in the inner type's *stored* form, which + # Ash casts on the way out. Casting it in SQL instead would ask Ecto to cast to + # a type SQLite has no native form for. + defp range_bound_expr(query, range, key, bindings, embedded?, acc, _type) do + {[range_type], _} = + determine_types(Ash.Query.Function.RangeLower, [range], nil) + + range_fragment(query, bindings, embedded?, acc, [{range, range_type}], fn [r] -> + json_at(r, key) + end) + end + + # Ecto's `Inspect` implementation for queries walks a fragment expecting raw + # and interpolated parts to alternate, and it raises on two raw parts in a row + # -- while *building an error message*, so a real query error would surface as + # a FunctionClauseError in `unmerge_fragments/3` instead. The SQL below is + # assembled from small pieces, so adjacent raw parts are the normal case. + defp merge_raw_parts(parts) do + parts + |> Enum.reduce([], fn + {:raw, next}, [{:raw, previous} | rest] -> [{:raw, previous <> next} | rest] + part, acc -> [part | acc] + end) + |> Enum.reverse() + end + + defp json_at(operand, key) do + [raw: "json_extract(", casted_expr: operand, raw: ", '$.#{key}')"] + end + + # Overlap is symmetric: each range's upper bound must lie after the other's + # lower bound. An absent bound is SQL NULL and reads as unbounded; a NULL + # *range* makes every comparison NULL, which is the nil semantics + # `Ash.Query.Function.RangeOverlaps.evaluate/1` has at runtime. + defp overlaps_sql(left, right) do + List.flatten([ + [raw: "("], + json_at(left, "empty"), + [raw: " = 0 AND "], + json_at(right, "empty"), + [raw: " = 0 AND ("], + upper_after_lower_sql(left, right), + [raw: ") AND ("], + upper_after_lower_sql(right, left), + [raw: "))"] + ]) + end + + defp upper_after_lower_sql(a, b) do + List.flatten([ + json_at(a, "upper"), + [raw: " IS NULL OR "], + json_at(b, "lower"), + [raw: " IS NULL OR "], + json_at(a, "upper"), + [raw: " > "], + json_at(b, "lower"), + [raw: " OR ("], + json_at(a, "upper"), + [raw: " = "], + json_at(b, "lower"), + [raw: " AND "], + json_at(a, "bounds"), + [raw: " IN ('[]','(]') AND "], + json_at(b, "bounds"), + [raw: " IN ('[]','[)'))"] + ]) + end + + defp contains_range_sql(outer, inner) do + List.flatten([ + [raw: "(("], + json_at(inner, "empty"), + [raw: " = 1) OR ("], + json_at(outer, "empty"), + [raw: " = 0 AND ("], + lower_encloses_sql(outer, inner), + [raw: ") AND ("], + upper_encloses_sql(outer, inner), + [raw: ")))"] + ]) + end + + defp lower_encloses_sql(outer, inner) do + List.flatten([ + json_at(outer, "lower"), + [raw: " IS NULL OR ("], + json_at(inner, "lower"), + [raw: " IS NOT NULL AND ("], + json_at(outer, "lower"), + [raw: " < "], + json_at(inner, "lower"), + [raw: " OR ("], + json_at(outer, "lower"), + [raw: " = "], + json_at(inner, "lower"), + [raw: " AND ("], + json_at(outer, "bounds"), + [raw: " IN ('[]','[)') OR "], + json_at(inner, "bounds"), + [raw: " IN ('(]','()')))))"] + ]) + end + + defp upper_encloses_sql(outer, inner) do + List.flatten([ + json_at(outer, "upper"), + [raw: " IS NULL OR ("], + json_at(inner, "upper"), + [raw: " IS NOT NULL AND ("], + json_at(inner, "upper"), + [raw: " < "], + json_at(outer, "upper"), + [raw: " OR ("], + json_at(inner, "upper"), + [raw: " = "], + json_at(outer, "upper"), + [raw: " AND ("], + json_at(outer, "bounds"), + [raw: " IN ('[]','(]') OR "], + json_at(inner, "bounds"), + [raw: " IN ('[)','()')))))"] + ]) + end + + defp contains_point_sql(range, point) do + List.flatten([ + [raw: "("], + json_at(range, "empty"), + [raw: " = 0 AND ("], + json_at(range, "lower"), + [raw: " IS NULL OR "], + json_at(range, "lower"), + [raw: " < "], + [casted_expr: point], + [raw: " OR ("], + json_at(range, "lower"), + [raw: " = "], + [casted_expr: point], + [raw: " AND "], + json_at(range, "bounds"), + [raw: " IN ('[]','[)'))) AND ("], + json_at(range, "upper"), + [raw: " IS NULL OR "], + [casted_expr: point], + [raw: " < "], + json_at(range, "upper"), + [raw: " OR ("], + [casted_expr: point], + [raw: " = "], + json_at(range, "upper"), + [raw: " AND "], + json_at(range, "bounds"), + [raw: " IN ('[]','(]'))))"] + ]) + end + + # Adjacent means no overlap and no gap, so exactly one of the two bounds that + # meet may be inclusive. Both inclusive overlap at the shared point; both + # exclusive leave it out of either range. + defp adjacent_sql(left, right) do + List.flatten([ + [raw: "(("], + meets_sql(left, right), + [raw: ") OR ("], + meets_sql(right, left), + [raw: "))"] + ]) + end + + defp meets_sql(a, b) do + List.flatten([ + json_at(a, "upper"), + [raw: " = "], + json_at(b, "lower"), + [raw: " AND (("], + json_at(a, "bounds"), + [raw: " IN ('[]','(]')) + ("], + json_at(b, "bounds"), + [raw: " IN ('[]','[)')) = 1)"] + ]) + end + + # `Ash.Query.Function.RangeContains` takes either a range or a point on the + # right, and `determine_types/3` reports both as the range type, so the shape + # has to come from the argument itself. + defp range_argument?(%Ash.Range{}), do: true + defp range_argument?(%Ash.Query.Ref{attribute: %{type: Ash.Type.Range}}), do: true + + defp range_argument?(%Ash.Query.Function.Type{arguments: [_value, type | _]}), + do: range_argument?(type) + + defp range_argument?(Ash.Type.Range), do: true + defp range_argument?({Ash.Type.Range, _constraints}), do: true + defp range_argument?(_other), do: false + + # A literal point is encoded here rather than typed, because the encoding has + # to be the one `AshSqlite.Type.RangeBound` performs and Ecto has no native + # type to hang it off. A point that is *not* a literal -- another column, say + # -- is left to the adapter, which agrees only if that column carries the same + # precision as the range's bounds. + defp point_operand(%Ash.Query.Function.Type{arguments: [value | _]} = wrapped) do + case point_operand(value) do + {^value, _type} -> {wrapped, nil} + encoded -> encoded + end + end + + defp point_operand(%DateTime{} = value), do: {AshSqlite.Type.RangeBound.encode(value), nil} + + defp point_operand(%NaiveDateTime{} = value), + do: {AshSqlite.Type.RangeBound.encode(value), nil} + + defp point_operand(%Date{} = value), do: {AshSqlite.Type.RangeBound.encode(value), nil} + defp point_operand(value) when is_integer(value), do: {value, nil} + defp point_operand(value), do: {value, nil} + + defp sqlite_range_type, do: Ecto.ParameterizedType.init(AshSqlite.Type.Range, []) + # SQLite has no `ARRAY[...]` constructor, so the `ARRAY[...]` / `array_to_json(ARRAY[...])` # rendering AshSql falls back to is a syntax error here: # @@ -837,6 +1187,16 @@ defmodule AshSqlite.SqlImplementation do parameterized_type(type, constraints) end + # SQLite has no range type, so a range is stored as JSON text. This has to + # intercept both the Ash type and the Ecto type Ash derives from it, because + # which one arrives depends on whether the caller already resolved it. + def parameterized_type(type, _constraints) + when type in [Ash.Type.Range, Ash.Type.Range.EctoType], + do: sqlite_range_type() + + def parameterized_type({:parameterized, {Ash.Type.Range.EctoType, _}}, _), + do: sqlite_range_type() + def parameterized_type(type, _constraints) when type in [Ash.Type.Map, Ash.Type.Map.EctoType], do: nil diff --git a/lib/temporal.ex b/lib/temporal.ex new file mode 100644 index 0000000..0703557 --- /dev/null +++ b/lib/temporal.ex @@ -0,0 +1,514 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.Temporal do + @moduledoc """ + The period arithmetic a temporal resource needs, performed in the data layer. + + Postgres splits a period with one statement: `UPDATE ... FOR PORTION OF valid_at + FROM $as_of TO NULL`. SQLite has no such clause, and no data-modifying CTE to + build one out of, so the same outcome takes three statements: + + 1. close the version whose period contains `as_of`, at `as_of`, + 2. copy that row forward with the period `[as_of, prior_upper)`, + 3. apply the caller's update to the copy. + + Step 3 is the ordinary non-temporal update path, which is the whole reason the + split is shaped this way: atomics, `RETURNING` and the changeset's filters all + keep working without a temporal branch inside them. + + The order matters and is not an implementation detail. Closing before inserting + means the two periods never overlap at any point between statements, so the + non-overlap trigger the migration generator emits never sees a transient + violation. The reverse order would trip it. + + All three run inside one transaction, which is why + `AshSqlite.DataLayer.Info.write_transactions?/1` must be true for a temporal + resource. Without it a split that fails halfway leaves a record with either two + current versions or none. + """ + + alias AshSqlite.Type.RangeBound + + # Temporal is unreleased: it exists on ash's `temporal` branch and not in any hex + # version. Everything below is compiled only when it is present, so ash_sqlite built + # against a released ash behaves exactly as it did before -- `temporal?/1` answers + # false, the data layer reports no temporal support, and no call reaches a function + # that is not there. + @supported? Code.ensure_loaded?(Ash.Temporal) + + @doc "Whether the ash this was compiled against has temporal resources." + def supported?, do: @supported? + + # Only these three reach into ash's temporal API. Gating them here rather than the + # whole module keeps every other function defined against a released ash, so nothing + # that calls them warns about a function that is not there. + if @supported? do + @doc "The attribute holding the period, or nil for a non-temporal resource." + def attribute(resource), do: Ash.Resource.Info.temporal_attribute(resource) + + @doc "Whether this resource stores its rows as periods." + def temporal?(resource), do: Ash.Resource.Info.temporal?(resource) + + @doc """ + The instant a write takes effect. + + A changeset that names no `as_of` takes effect now. A resource whose period type + has no current value has no such instant, and that is an error rather than a + default -- a resource numbering its versions from one has no `:now` to give. + """ + @doc """ + The fields that identify one row. + + On a temporal table the primary key names every version of a record, so the period + completes the identity. Anything that aims a statement at a single row filters or + joins on this rather than on the key alone. + """ + def identity_fields(resource) do + case attribute(resource) do + nil -> Ash.Resource.Info.primary_key(resource) + attribute -> Ash.Resource.Info.primary_key(resource) ++ [attribute] + end + end + + def write_instant(resource, %{as_of: as_of}), do: write_instant(resource, as_of) + + def write_instant(resource, as_of) do + case Ash.Temporal.write_instant(resource, as_of || :now) do + {:ok, instant} -> + instant + + :error -> + raise ArgumentError, "#{inspect(resource)} has no write instant for #{inspect(as_of)}" + end + end + else + @doc "The attribute holding the period, or nil for a non-temporal resource." + def attribute(_resource), do: nil + + @doc "Whether this resource stores its rows as periods." + def temporal?(_resource), do: false + + # Unreachable: `temporal?/1` is the only gate into anything that asks for a write + # instant, and it is false in this build. It returns a value rather than raising so + # that the orchestration below still type-checks here, where it is never called. + @doc false + def identity_fields(resource), do: Ash.Resource.Info.primary_key(resource) + + @doc false + def write_instant(_resource, _as_of), do: nil + end + + @doc "The period a write opens: `[as_of, upper)`, unbounded above unless a later version bounds it." + def period(resource, as_of, upper \\ nil) do + %Ash.Range{ + lower: as_of, + upper: upper, + bounds: bounds(resource), + empty?: false + } + end + + defp bounds(resource) do + case attribute(resource) && Ash.Resource.Info.attribute(resource, attribute(resource)) do + %{constraints: constraints} -> + lower = get_in(constraints, [:lower, :inclusive?]) + upper = get_in(constraints, [:upper, :inclusive?]) + + case {lower == false, upper == true} do + {false, false} -> :"[)" + {false, true} -> :"[]" + {true, false} -> :"()" + {true, true} -> :"(]" + end + + _other -> + :"[)" + end + end + + @doc """ + The SQL expression for a period bound. + + The bounds stay inside the JSON rather than being projected into generated + columns, because SQLite indexes an expression directly and the planner matches + an index on `json_extract(valid_at, '$.lower')` to this same expression, alias + and quoting included. Measured: `SEARCH ... USING INDEX ... (id=? AND + case fun.() do + {:error, reason} -> repo.rollback(reason) + other -> other + end + end, + mode: :immediate + ) + + case result do + {:ok, value} -> value + {:error, reason} -> {:error, reason} + end + end + end + + @doc "Encodes an instant the way period bounds are encoded, so the two compare." + def encode(instant), do: RangeBound.encode(instant) + + # `Ash.Type.Range` allows `:date`, `:integer`, `:naive_datetime` and `:datetime` + # bounds, and `DateTime.compare/2` raises a FunctionClauseError on the first three. + # Comparing the encoded forms works for all four, and it is the same comparison the + # database performs on the stored bounds, so the two can never disagree. + defp same_bound?(left, right), do: encode(left) == encode(right) + + @doc """ + The SQL that selects the version valid at an instant, and its parameters. + + The lower bound is compared inclusively and the upper exclusively, matching the + default `[)` bounds. An unbounded upper is `NULL`, so "still current" is + `IS NULL` rather than a comparison against a sentinel. + """ + def valid_at_sql(attribute) do + lo = lower_expr(attribute) + hi = upper_expr(attribute) + "(#{lo} <= ? AND (#{hi} IS NULL OR #{hi} > ?))" + end + + @doc """ + Reads the version of `pkey` valid at `as_of`, as `{rowid, %Ash.Range{}}`. + + Returns `nil` when the record has no version covering that instant, which is + what an update against a gap, or against a time before the record existed, has + to distinguish from a version it can split. + """ + def current_version(repo, resource, pkey, as_of) do + attribute = attribute(resource) + table = AshSqlite.DataLayer.Info.table(resource) + {where, params} = pkey_where(pkey) + encoded = encode(as_of) + + sql = + "SELECT rowid, #{quote_name(attribute)} FROM #{quote_name(table)} " <> + "WHERE #{where} AND #{valid_at_sql(attribute)} LIMIT 1" + + case repo.query!(sql, params ++ [encoded, encoded]) do + %{rows: [[rowid, stored]]} -> {rowid, decode_range(resource, attribute, stored)} + %{rows: []} -> nil + end + end + + @doc """ + The lower bound of the earliest version of `pkey` that begins at or after `as_of`. + + A write into a gap is bounded above by the version that follows it, so that the + new period does not run through one that already exists. With no later version + the period is unbounded. + """ + def next_lower_bound(repo, resource, pkey, as_of) do + attribute = attribute(resource) + table = AshSqlite.DataLayer.Info.table(resource) + {where, params} = pkey_where(pkey) + lo = lower_expr(attribute) + + sql = + "SELECT #{lo} FROM #{quote_name(table)} " <> + "WHERE #{where} AND #{lo} >= ? ORDER BY #{lo} LIMIT 1" + + case repo.query!(sql, params ++ [encode(as_of)]) do + %{rows: [[bound]]} -> cast_bound(resource, attribute, bound) + %{rows: []} -> nil + end + end + + @doc """ + Closes the version at `rowid` at `as_of`, and returns the period it had. + + A version whose period begins exactly at `as_of` is not closed — there is no + before-portion to keep, and closing it would leave an empty period. The caller + distinguishes the two by the `:replace` return. + """ + def close_version(repo, resource, rowid, %Ash.Range{} = prior, as_of) do + if same_bound?(prior.lower, as_of) do + :replace + else + attribute = attribute(resource) + table = AshSqlite.DataLayer.Info.table(resource) + closed = %{prior | upper: as_of} + + repo.query!( + "UPDATE #{quote_name(table)} SET #{quote_name(attribute)} = ? WHERE rowid = ?", + [dump!(closed), rowid] + ) + + {:closed, prior} + end + end + + @doc """ + Copies the row at `rowid` forward under a new period, and returns the new rowid. + + Every column is copied except the period and the generated bound columns, so a + column added to the table later is carried without this function knowing about + it. The caller then applies its own changes to the copy. + """ + def copy_forward(repo, resource, rowid, period) do + attribute = attribute(resource) + table = AshSqlite.DataLayer.Info.table(resource) + columns = copyable_columns(repo, table, attribute) + column_list = Enum.map_join(columns, ", ", "e_name/1) + + %{rows: [[new_rowid]]} = + repo.query!( + "INSERT INTO #{quote_name(table)} (#{column_list}, #{quote_name(attribute)}) " <> + "SELECT #{column_list}, ? FROM #{quote_name(table)} WHERE rowid = ? " <> + "RETURNING rowid", + [dump!(period), rowid] + ) + + new_rowid + end + + # Every column of the table except the period, which the caller supplies. Reading + # the columns from the table rather than from the resource means a column the + # resource does not declare is still carried forward by a split, instead of being + # silently dropped from the new version. + defp copyable_columns(repo, table, attribute) do + repo.query!("SELECT name FROM pragma_table_info(?)", [table]) + |> Map.fetch!(:rows) + |> List.flatten() + |> Enum.reject(&(&1 == to_string(attribute))) + end + + defp pkey_where(pkey) do + {clauses, params} = + pkey + |> Enum.map(fn {key, value} -> {"#{quote_name(key)} = ?", value} end) + |> Enum.unzip() + + {Enum.join(clauses, " AND "), params} + end + + defp dump!(%Ash.Range{} = range) do + {:ok, dumped} = AshSqlite.Type.Range.dump(range, nil, %{}) + dumped + end + + defp decode_range(resource, attribute, stored) do + params = range_params(resource, attribute) + {:ok, range} = AshSqlite.Type.Range.load(stored, nil, params) + range + end + + defp cast_bound(_resource, _attribute, nil), do: nil + + defp cast_bound(resource, attribute, stored) do + %{inner_type: inner_type, inner_constraints: inner_constraints} = + range_params(resource, attribute) + + case Ash.Type.cast_stored(inner_type, stored, inner_constraints) do + {:ok, value} -> value + _other -> stored + end + end + + defp range_params(resource, attribute) do + constraints = Ash.Resource.Info.attribute(resource, attribute).constraints + + %{ + inner_type: Ash.Type.get_type(constraints[:inner_type]), + inner_constraints: constraints[:inner_constraints] || [] + } + end + + # SQLite quotes an identifier with double quotes, and escapes an embedded one by + # doubling it. Table and column names reach here from the DSL rather than from a + # request, but interpolating them unquoted would still break on a name that needs + # quoting at all. + defp quote_name(name) do + escaped = name |> to_string() |> String.replace("\"", "\"\"") + "\"" <> escaped <> "\"" + end + + @doc """ + Loads the row at `rowid` as a resource struct. + + An upsert that matched has to hand Ash back the record as it now stands, and the + changeset only carries the keys and the fields the action nominated. Reading the row + is what makes the untouched attributes real rather than the struct's defaults. + """ + def version_at(repo, resource, rowid) do + import Ecto.Query, only: [from: 2] + + repo.one(from(row in resource, where: fragment("rowid") == ^rowid)) + end + + @doc "The period stored at `rowid`." + def period_of(repo, resource, rowid) do + attribute = attribute(resource) + table = AshSqlite.DataLayer.Info.table(resource) + + %{rows: [[stored]]} = + repo.query!( + "SELECT #{quote_name(attribute)} FROM #{quote_name(table)} WHERE rowid = ?", + [rowid] + ) + + decode_range(resource, attribute, stored) + end + + @doc """ + Splits the version at `rowid` at `as_of`, returning what the caller must do next. + + `:replace` means the version began at `as_of` and was left alone, so an update + applies to it directly and a destroy deletes it. `{:split, new_rowid}` means the + version was closed and copied forward. + """ + def split_rowid(repo, resource, rowid, as_of) do + prior = period_of(repo, resource, rowid) + + case close_version(repo, resource, rowid, prior, as_of) do + :replace -> + :replace + + {:closed, prior} -> + {:split, copy_forward(repo, resource, rowid, period(resource, as_of, prior.upper))} + end + end + + @doc """ + Splits every version in `rowids` at `as_of`. + + This is the set-wide path, and it is one statement pair per row where Postgres + issues a single `FOR PORTION OF`. The rows are known to be the ones the caller's + query matched, so there is no second filter pass. + + After this returns, a query carrying the `as_of` containment filter matches exactly + the new versions: each prior version now ends at `as_of`, which the filter excludes, + and each new one begins there, which it includes. That is what lets the caller run + its original statement unchanged rather than re-targeting it at the copies. + """ + def split_all(repo, resource, rowids, as_of) do + Enum.each(rowids, &split_rowid(repo, resource, &1, as_of)) + end + + @doc """ + Ends the validity of every version in `rowids` at `as_of`. + + Returns the rowids that must still be deleted: those whose period began exactly at + `as_of` and so have no before-portion worth keeping. + """ + def truncate_all(repo, resource, rowids, as_of) do + Enum.filter(rowids, fn rowid -> + prior = period_of(repo, resource, rowid) + close_version(repo, resource, rowid, prior, as_of) == :replace + end) + end + + @doc """ + Performs a temporal update, and hands the caller a changeset aimed at the new version. + + The split happens here; the caller's function then runs the ordinary update against + the copy. `continue` receives the changeset with its `data` re-pointed at the new + version, which is what makes the row identity `(primary key, period)` resolve to + exactly one row. + + A write that lands on a version's own lower bound has no before-portion to keep, so + it updates that version in place rather than splitting it. Splitting there would + leave an empty `[as_of, as_of)` period behind. + """ + def update(repo, resource, changeset, continue) do + as_of = write_instant(resource, changeset) + pkey = pkey_of(resource, changeset.data) + + case current_version(repo, resource, pkey, as_of) do + nil -> + {:error, + Ash.Error.Changes.StaleRecord.exception( + resource: resource, + filter: changeset.filter + )} + + {rowid, prior} -> + period = + case close_version(repo, resource, rowid, prior, as_of) do + :replace -> + prior + + {:closed, prior} -> + new_period = period(resource, as_of, prior.upper) + copy_forward(repo, resource, rowid, new_period) + new_period + end + + continue.(aim_at(changeset, resource, period)) + end + end + + @doc """ + Ends a record's validity at `as_of`, keeping everything before it. + + A destroy landing on the version's own lower bound deletes it, because truncating a + period to zero width is not a period. Any other instant closes the version and + leaves the history in place, which is why a temporal destroy is not a delete. + """ + def destroy(repo, resource, changeset, delete) do + as_of = write_instant(resource, changeset) + pkey = pkey_of(resource, changeset.data) + + case current_version(repo, resource, pkey, as_of) do + nil -> + :ok + + {rowid, prior} -> + case close_version(repo, resource, rowid, prior, as_of) do + :replace -> delete.(aim_at(changeset, resource, prior)) + {:closed, _prior} -> :ok + end + end + end + + # Re-points a changeset at one version of a record. The data layer filters an update + # by the row identity it reads off `changeset.data`, and for a temporal resource that + # identity is the primary key *plus* the period -- the key alone names every version. + defp aim_at(changeset, resource, period) do + %{changeset | data: Map.put(changeset.data, attribute(resource), period)} + end + + defp pkey_of(resource, record) do + resource + |> Ash.Resource.Info.primary_key() + |> Map.new(fn key -> {key, Map.fetch!(record, key)} end) + end +end diff --git a/lib/temporal/migration.ex b/lib/temporal/migration.ex new file mode 100644 index 0000000..925ecf5 --- /dev/null +++ b/lib/temporal/migration.ex @@ -0,0 +1,163 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.Temporal.Migration do + @moduledoc """ + The indexes and triggers a temporal table needs, as custom statements. + + Postgres gets all of this from one declaration: `PRIMARY KEY (id, valid_at WITHOUT + OVERLAPS)` is both the uniqueness rule and, through its GiST index, the access path. + SQLite has neither an exclusion constraint nor GiST, so the same guarantees are + assembled from three pieces it does have. + + * **An index on `(key..., json_extract(period, '$.lower'))`.** The planner matches an + index built on an expression to the same expression in a query, alias and quoting + included, so a point-in-time read of one record is an index seek rather than a scan. + * **A partial unique index over the open-ended period.** `UNIQUE (key...) WHERE + json_extract(period, '$.upper') IS NULL` refuses a second current version for a key. + This is a real constraint rather than a trigger, and it is the one that matters: a + split that fails to close the prior version is exactly what it rejects. + * **A pair of non-overlap triggers.** The partial index says nothing about two + overlapping *closed* periods, which is the rest of `WITHOUT OVERLAPS`. A + `BEFORE INSERT`/`BEFORE UPDATE` trigger rejects those with `RAISE(ABORT)`. + + These are emitted as custom statements so they reuse the generator's own diffing and + its ordering guarantee: `down` statements run first and `up` statements last, which is + what a trigger needs, since it cannot be created before its table. + """ + + # Against a released ash there is no temporal section, so `statements/1` is `[]` and + # everything it would have called is dead code the compiler need not carry. + if AshSqlite.Temporal.supported?() do + @doc "The statements for `resource`, or `[]` when it is not temporal." + def statements(resource) do + case AshSqlite.Temporal.attribute(resource) do + nil -> [] + attribute -> build(resource, attribute) + end + end + + defp build(resource, attribute) do + table = AshSqlite.DataLayer.Info.table(resource) + primary_key = Ash.Resource.Info.primary_key(resource) + + # Every identity gets the same treatment as the primary key. An identity on a + # temporal resource is unique per period rather than per table, so the generator + # drops it from the snapshot and it reappears here as a partial unique index plus a + # trigger pair over its own keys. + identities = + resource + |> Ash.Resource.Info.identities() + |> Enum.map(& &1.keys) + |> Enum.reject(&(Enum.sort(&1) == Enum.sort(primary_key))) + + Enum.flat_map([primary_key | identities], fn keys -> + suffix = key_suffix(keys, primary_key) + + # Each key set needs its own point-in-time index, not just the primary key's. The + # non-overlap trigger for a key set is a correlated subquery over that key, and + # without a matching index it scans the whole table on every write. Measured on + # 200k rows: `SEARCH other USING INDEX ... (slug=?)` with it, + # `SCAN other` without. + point_in_time_index(table, attribute, keys, suffix) ++ + [ + current_version_index(table, attribute, keys, suffix), + no_overlap_trigger(table, attribute, keys, suffix, :insert), + no_overlap_trigger(table, attribute, keys, suffix, :update) + ] + end) + end + + # The primary key's statements keep their unsuffixed names, so an existing temporal + # table does not see every one of them dropped and recreated. + defp key_suffix(keys, primary_key) do + if Enum.sort(keys) == Enum.sort(primary_key), do: "", else: "_" <> Enum.join(keys, "_") + end + + defp point_in_time_index(_table, _attribute, [], _suffix), do: [] + + defp point_in_time_index(table, attribute, keys, suffix) do + name = "#{table}_#{attribute}_pit#{suffix}" + columns = Enum.map_join(keys, ", ", "e_name/1) + + %{ + name: String.to_atom(name), + code?: false, + up: + "CREATE INDEX #{quote_name(name)} ON #{quote_name(table)} " <> + "(#{columns}, #{lower(attribute)});", + down: "DROP INDEX IF EXISTS #{quote_name(name)};" + } + |> List.wrap() + end + + defp current_version_index(table, attribute, keys, suffix) do + name = "#{table}_#{attribute}_current#{suffix}" + columns = Enum.map_join(keys, ", ", "e_name/1) + + %{ + name: String.to_atom(name), + code?: false, + up: + "CREATE UNIQUE INDEX #{quote_name(name)} ON #{quote_name(table)} " <> + "(#{columns}) WHERE #{upper(attribute)} IS NULL;", + down: "DROP INDEX IF EXISTS #{quote_name(name)};" + } + end + + defp no_overlap_trigger(table, attribute, keys, suffix, event) do + name = "#{table}_#{attribute}_no_overlap#{suffix}_#{event}" + + # `IS` rather than `=` so a nullable key column compares as a value. With `=` the + # comparison would be NULL and the row would be found not to overlap anything. + key_match = + Enum.map_join(keys, "\n AND ", fn key -> + "other.#{quote_name(key)} IS NEW.#{quote_name(key)}" + end) + + # On an insert the row has no rowid yet, so `other.rowid <> NEW.rowid` is NULL, the + # whole conjunction is NULL, and the trigger silently never fires. An update does + # need it, or the row being updated is found to overlap itself. + self_match = + case event do + :insert -> "" + :update -> "\n AND other.rowid <> NEW.rowid" + end + + %{ + name: String.to_atom(name), + code?: false, + up: """ + CREATE TRIGGER #{quote_name(name)} + BEFORE #{String.upcase(to_string(event))} ON #{quote_name(table)} + BEGIN + SELECT RAISE(ABORT, '#{attribute} overlaps an existing period') + WHERE EXISTS ( + SELECT 1 FROM #{quote_name(table)} AS other + WHERE #{key_match}#{self_match} + AND (#{new(attribute, "upper")} IS NULL + OR #{other(attribute, "lower")} < #{new(attribute, "upper")}) + AND (#{other(attribute, "upper")} IS NULL + OR #{other(attribute, "upper")} > #{new(attribute, "lower")}) + ); + END; + """, + down: "DROP TRIGGER IF EXISTS #{quote_name(name)};" + } + end + + defp lower(attribute), do: "json_extract(#{quote_name(attribute)}, '$.lower')" + defp upper(attribute), do: "json_extract(#{quote_name(attribute)}, '$.upper')" + defp new(attribute, bound), do: "json_extract(NEW.#{quote_name(attribute)}, '$.#{bound}')" + defp other(attribute, bound), do: "json_extract(other.#{quote_name(attribute)}, '$.#{bound}')" + + defp quote_name(name) do + escaped = name |> to_string() |> String.replace("\"", "\"\"") + "\"" <> escaped <> "\"" + end + else + @doc "The statements for `resource`, or `[]` when it is not temporal." + def statements(_resource), do: [] + end +end diff --git a/lib/type/range.ex b/lib/type/range.ex new file mode 100644 index 0000000..fe9a77f --- /dev/null +++ b/lib/type/range.ex @@ -0,0 +1,115 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.Type.Range do + @moduledoc """ + Stores an `Ash.Type.Range` as a single JSON text column. + + SQLite has no range type and no GiST, so the two things Postgres gets from + `tstzrange` -- a native column and an exclusion constraint -- have to be built + from what SQLite does have. The column half is this type. The constraint half + is the caller's: SQLite cannot express non-overlap declaratively, so a + no-overlap guarantee has to come from a trigger or from a single writer. + + The stored shape is a JSON object whose keys are the ones `Ash.Type.Range` + already destructures on the way back in, so `load/1` hands Ash a plain map and + Ash casts each bound with the inner type's `cast_stored/2`: + + {"lower": "2026-01-01T00:00:00.000000Z", "upper": null, "bounds": "[)", "empty": false} + + Two properties are load-bearing, and both are about comparison rather than + storage, because every range predicate compiles to comparisons on + `json_extract` of these keys: + + * **Bounds are encoded by `AshSqlite.Type.RangeBound`**, so datetimes compare + lexicographically as ISO8601 at a single fixed precision. See that module + for what goes silently wrong when the two sides of a comparison are encoded + by different code. + * **An absent bound is JSON `null`**, which `json_extract` returns as SQL + NULL, so "unbounded" is tested with `IS NULL` rather than against a + sentinel. A sentinel would have to be a value of the inner type, and there + is no one value that works for every inner type. + """ + + use Ecto.ParameterizedType + + alias AshSqlite.Type.RangeBound + + # Ash passes an attribute's constraints through as field options, so this is + # where the inner type arrives. It is needed on the way *out* only: encoding a + # bound is the same for every inner type the range allows, but decoding one + # has to go through that type's `cast_stored/2`. The query path initialises + # this type with no options, because it only ever dumps. + @impl true + def init(opts) do + %{ + inner_type: opts[:inner_type], + inner_constraints: opts[:inner_constraints] || [] + } + end + + @impl true + def type(_params), do: :string + + @impl true + def cast(nil, _params), do: {:ok, nil} + def cast(%Ash.Range{} = range, _params), do: {:ok, range} + def cast(value, params) when is_binary(value), do: load(value, nil, params) + def cast(_value, _params), do: :error + + @impl true + def dump(nil, _dumper, _params), do: {:ok, nil} + + def dump(%Ash.Range{empty?: true}, _dumper, _params) do + {:ok, Jason.encode!(%{"lower" => nil, "upper" => nil, "bounds" => "[)", "empty" => true})} + end + + def dump(%Ash.Range{} = range, _dumper, _params) do + {:ok, + Jason.encode!(%{ + "lower" => RangeBound.encode(range.lower), + "upper" => RangeBound.encode(range.upper), + "bounds" => to_string(range.bounds), + "empty" => false + })} + end + + def dump(_value, _dumper, _params), do: :error + + @impl true + def load(nil, _loader, _params), do: {:ok, nil} + + def load(value, _loader, params) when is_binary(value) do + with {:ok, %{"bounds" => bounds} = decoded} <- Jason.decode(value), + {:ok, bounds} <- decode_bounds(bounds), + {:ok, lower} <- decode_bound(decoded["lower"], params), + {:ok, upper} <- decode_bound(decoded["upper"], params) do + {:ok, + %Ash.Range{ + lower: lower, + upper: upper, + bounds: bounds, + empty?: decoded["empty"] || false + }} + else + _other -> :error + end + end + + def load(_value, _loader, _params), do: :error + + # Ash does not cast an attribute Ecto has already loaded, so the value handed + # back here has to be the finished `Ash.Range`, bounds included. + defp decode_bound(nil, _params), do: {:ok, nil} + + defp decode_bound(value, %{inner_type: nil}), do: {:ok, value} + + defp decode_bound(value, %{inner_type: inner_type, inner_constraints: inner_constraints}) do + Ash.Type.cast_stored(inner_type, value, inner_constraints) + end + + @bounds ["[)", "[]", "(]", "()"] + defp decode_bounds(bounds) when bounds in @bounds, do: {:ok, String.to_atom(bounds)} + defp decode_bounds(_bounds), do: :error +end diff --git a/lib/type/range_bound.ex b/lib/type/range_bound.ex new file mode 100644 index 0000000..5c509a2 --- /dev/null +++ b/lib/type/range_bound.ex @@ -0,0 +1,59 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.Type.RangeBound do + @moduledoc """ + Encodes one value the way `AshSqlite.Type.Range` encodes a range's bounds. + + This exists so that a point compared against a range is encoded by the *same* + function that encoded the bounds, rather than by the adapter's own datetime + codec. The two do not agree, and the disagreement is silent: + `DateTime.to_iso8601/1` preserves whatever precision the value carries, so + `~U[2026-01-15 00:00:00Z]` encodes as `"2026-01-15T00:00:00Z"` while a bound + normalised to microseconds encodes as `"2026-01-15T00:00:00.000000Z"`. + Compared as text, `"Z"` (0x5A) sorts above `"."` (0x2E), so two spellings of + one instant compare unequal, and in the wrong direction. + + Only the range functions route through this type; ordinary datetime columns are + still encoded by the adapter. + """ + + use Ecto.ParameterizedType + + # The encoding is the same for every inner type `Ash.Type.Range` allows, so + # there is nothing to parameterise. This is a parameterized type only because + # that is the shape Ecto accepts when Ash passes an attribute's constraints + # through as field options. + @impl true + def init(_opts), do: %{} + + @impl true + def type(_params), do: :string + + @impl true + def cast(value, _params), do: {:ok, value} + + @impl true + def dump(value, _dumper, _params), do: {:ok, encode(value)} + + @impl true + def load(value, _loader, _params), do: {:ok, value} + + @doc """ + Encodes a bound into the text form ranges compare on. + + Datetimes are forced to microsecond precision, because ISO8601 only orders + lexicographically among strings of equal precision. `Ash.Type.Range` limits + inner types to date, integer, naive_datetime and datetime, so there is no + inner type here whose Elixir form needs anything cleverer than this. + """ + def encode(nil), do: nil + def encode(%DateTime{} = value), do: value |> force_usec() |> DateTime.to_iso8601() + def encode(%NaiveDateTime{} = value), do: value |> force_usec() |> NaiveDateTime.to_iso8601() + def encode(%Date{} = value), do: Date.to_iso8601(value) + def encode(value), do: value + + defp force_usec(%{microsecond: {value, _precision}} = datetime), + do: %{datetime | microsecond: {value, 6}} +end diff --git a/lib/verifiers/verify_temporal.ex b/lib/verifiers/verify_temporal.ex new file mode 100644 index 0000000..98be5b1 --- /dev/null +++ b/lib/verifiers/verify_temporal.ex @@ -0,0 +1,83 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.Verifiers.VerifyTemporal do + @moduledoc """ + Refuses a temporal resource whose repo does not run write transactions. + + Postgres splits a period with one statement, so it is atomic whether or not the + caller opened a transaction. SQLite has no `FOR PORTION OF`, so the same split is a + close, an insert and an update (`AshSqlite.Temporal`). Run without a transaction, + a failure between them leaves the record with two current versions or none, and the + partial unique index catches only the first of those. + + `AshPostgres.Verifiers.VerifyTemporal` checks for `btree_gist` for the same reason: + the data layer cannot deliver the guarantee without something the application has to + turn on, so it says so at compile time rather than at the first failed write. + """ + use Spark.Dsl.Verifier + + # Against a released ash there is no `temporal` section, so there is nothing here to + # check and every branch below is dead code. + if AshSqlite.Temporal.supported?() do + alias Spark.Dsl.Verifier + alias Spark.Error.DslError + + @impl true + def verify(dsl_state) do + resource = Verifier.get_persisted(dsl_state, :module) + + if temporal?(dsl_state) and not write_transactions?(dsl_state, resource) do + {:error, + DslError.exception( + module: resource, + path: [:temporal, :strategy], + message: """ + A temporal resource requires a repo with write transactions enabled. + + A period split on SQLite is three statements, and only a transaction makes + them one write. Define `write_transactions?/0` as `true` on the repo: + + defmodule #{inspect(repo(dsl_state))} do + use AshSqlite.Repo, otp_app: :my_app + + def write_transactions?, do: true + end + """ + )} + else + :ok + end + end + + defp temporal?(dsl_state) do + not is_nil(Verifier.get_option(dsl_state, [:temporal], :strategy)) + end + + # The `repo` option is either a module or a 2-arity function of the resource and the + # operation (see `test/support/resources/named_fn_repo_account.ex`). `Code.ensure_loaded?/1` + # is guarded on an atom and raises on a capture, so the function has to be resolved + # first. The resource module does not exist yet, so it is passed as nil, which is + # what the option's own callers tolerate. + defp repo(dsl_state) do + case Verifier.get_option(dsl_state, [:sqlite], :repo) do + repo when is_function(repo, 2) -> repo.(nil, :mutate) + repo -> repo + end + end + + defp write_transactions?(dsl_state, _resource) do + case repo(dsl_state) do + repo when is_atom(repo) and not is_nil(repo) -> + Code.ensure_loaded?(repo) and repo.write_transactions?() + + _other -> + true + end + end + else + @impl true + def verify(_dsl_state), do: :ok + end +end diff --git a/mix.exs b/mix.exs index 1362cae..7ed3511 100644 --- a/mix.exs +++ b/mix.exs @@ -173,6 +173,9 @@ defmodule AshSqlite.MixProject do "main" -> [git: "https://github.com/ash-project/ash.git", override: true] + "temporal" -> + [git: "https://github.com/ash-project/ash.git", branch: "temporal", override: true] + version when is_binary(version) -> "~> #{version}" @@ -192,6 +195,9 @@ defmodule AshSqlite.MixProject do "main" -> [git: "https://github.com/ash-project/ash_sql.git"] + "temporal" -> + [git: "https://github.com/ash-project/ash_sql.git", branch: "temporal", override: true] + version when is_binary(version) -> "~> #{version}" diff --git a/priv/test_repo/migrations/20260919000000_add_subscriptions_table.exs b/priv/test_repo/migrations/20260919000000_add_subscriptions_table.exs new file mode 100644 index 0000000..b608609 --- /dev/null +++ b/priv/test_repo/migrations/20260919000000_add_subscriptions_table.exs @@ -0,0 +1,63 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.TestRepo.Migrations.AddSubscriptionsTable do + use Ecto.Migration + + def up do + execute(""" + CREATE TABLE subscriptions ( + id INTEGER NOT NULL, + tier TEXT, + seats INTEGER DEFAULT 0, + activated_at TEXT_DATETIME, + valid_at TEXT NOT NULL + ) + """) + + execute(""" + CREATE INDEX subscriptions_valid_at_pit + ON subscriptions (id, json_extract(valid_at, '$.lower')) + """) + + execute(""" + CREATE UNIQUE INDEX subscriptions_valid_at_current + ON subscriptions (id) WHERE json_extract(valid_at, '$.upper') IS NULL + """) + + # The insert trigger must not exclude NEW.rowid. On an insert the row has no + # rowid yet, so `other.rowid <> NEW.rowid` is NULL, the whole conjunction is + # NULL, and the trigger silently never fires. The update trigger does need it, + # or a row would be found to overlap itself. + for {name, event, self_clause} <- [ + {"insert", "INSERT", ""}, + {"update", "UPDATE", "AND other.rowid <> NEW.rowid"} + ] do + execute(""" + CREATE TRIGGER subscriptions_valid_at_no_overlap_#{name} + BEFORE #{event} ON subscriptions + BEGIN + SELECT RAISE(ABORT, 'valid_at overlaps an existing period') + WHERE EXISTS ( + SELECT 1 FROM subscriptions AS other + WHERE other.id IS NEW.id + #{self_clause} + AND (json_extract(NEW.valid_at, '$.upper') IS NULL + OR json_extract(other.valid_at, '$.lower') < json_extract(NEW.valid_at, '$.upper')) + AND (json_extract(other.valid_at, '$.upper') IS NULL + OR json_extract(other.valid_at, '$.upper') > json_extract(NEW.valid_at, '$.lower')) + ); + END + """) + end + end + + def down do + execute("DROP TRIGGER IF EXISTS subscriptions_valid_at_no_overlap_update") + execute("DROP TRIGGER IF EXISTS subscriptions_valid_at_no_overlap_insert") + execute("DROP INDEX IF EXISTS subscriptions_valid_at_current") + execute("DROP INDEX IF EXISTS subscriptions_valid_at_pit") + execute("DROP TABLE IF EXISTS subscriptions") + end +end diff --git a/priv/test_repo/migrations/20260919010000_add_dated_subscriptions_table.exs b/priv/test_repo/migrations/20260919010000_add_dated_subscriptions_table.exs new file mode 100644 index 0000000..8c20558 --- /dev/null +++ b/priv/test_repo/migrations/20260919010000_add_dated_subscriptions_table.exs @@ -0,0 +1,33 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.TestRepo.Migrations.AddDatedSubscriptionsTable do + use Ecto.Migration + + def up do + execute(""" + CREATE TABLE dated_subscriptions ( + id INTEGER NOT NULL, + tier TEXT, + valid_on TEXT NOT NULL + ) + """) + + execute(""" + CREATE INDEX dated_subscriptions_valid_on_pit + ON dated_subscriptions (id, json_extract(valid_on, '$.lower')) + """) + + execute(""" + CREATE UNIQUE INDEX dated_subscriptions_valid_on_current + ON dated_subscriptions (id) WHERE json_extract(valid_on, '$.upper') IS NULL + """) + end + + def down do + execute("DROP INDEX IF EXISTS dated_subscriptions_valid_on_current") + execute("DROP INDEX IF EXISTS dated_subscriptions_valid_on_pit") + execute("DROP TABLE IF EXISTS dated_subscriptions") + end +end diff --git a/priv/test_repo/migrations/20260920000000_add_plans_table.exs b/priv/test_repo/migrations/20260920000000_add_plans_table.exs new file mode 100644 index 0000000..1710d5e --- /dev/null +++ b/priv/test_repo/migrations/20260920000000_add_plans_table.exs @@ -0,0 +1,67 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.TestRepo.Migrations.AddPlansTable do + use Ecto.Migration + + # Built by applying what `AshSqlite.Temporal.Migration.statements/1` emits for + # `AshSqlite.Test.Plan`, which the temporal suite asserts against directly. + def up do + execute(""" + CREATE TABLE plans ( + id INTEGER NOT NULL, + slug TEXT NOT NULL, + price INTEGER DEFAULT 0, + valid_at TEXT NOT NULL + ) + """) + + for {suffix, keys} <- [{"", "id"}, {"_slug", "slug"}] do + execute(""" + CREATE INDEX plans_valid_at_pit#{suffix} + ON plans (#{keys}, json_extract(valid_at, '$.lower')) + """) + + execute(""" + CREATE UNIQUE INDEX plans_valid_at_current#{suffix} + ON plans (#{keys}) WHERE json_extract(valid_at, '$.upper') IS NULL + """) + + for {name, event, self_clause} <- [ + {"insert", "INSERT", ""}, + {"update", "UPDATE", "AND other.rowid <> NEW.rowid"} + ] do + execute(""" + CREATE TRIGGER plans_valid_at_no_overlap#{suffix}_#{name} + BEFORE #{event} ON plans + BEGIN + SELECT RAISE(ABORT, 'valid_at overlaps an existing period') + WHERE EXISTS ( + SELECT 1 FROM plans AS other + WHERE other.#{keys} IS NEW.#{keys} + #{self_clause} + AND (json_extract(NEW.valid_at, '$.upper') IS NULL + OR json_extract(other.valid_at, '$.lower') < json_extract(NEW.valid_at, '$.upper')) + AND (json_extract(other.valid_at, '$.upper') IS NULL + OR json_extract(other.valid_at, '$.upper') > json_extract(NEW.valid_at, '$.lower')) + ); + END + """) + end + end + end + + def down do + for suffix <- ["_slug", ""], name <- ["update", "insert"] do + execute("DROP TRIGGER IF EXISTS plans_valid_at_no_overlap#{suffix}_#{name}") + end + + for suffix <- ["_slug", ""] do + execute("DROP INDEX IF EXISTS plans_valid_at_current#{suffix}") + execute("DROP INDEX IF EXISTS plans_valid_at_pit#{suffix}") + end + + execute("DROP TABLE IF EXISTS plans") + end +end diff --git a/priv/test_repo/migrations/20260920010000_add_enrollments_table.exs b/priv/test_repo/migrations/20260920010000_add_enrollments_table.exs new file mode 100644 index 0000000..f2d5a6b --- /dev/null +++ b/priv/test_repo/migrations/20260920010000_add_enrollments_table.exs @@ -0,0 +1,22 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.TestRepo.Migrations.AddEnrollmentsTable do + use Ecto.Migration + + # `plan_id` carries no foreign key. Its destination is temporal and so has no unique + # key to reference; SQLite would reject every insert here with "foreign key mismatch". + def up do + execute(""" + CREATE TABLE enrollments ( + id INTEGER NOT NULL PRIMARY KEY, + plan_id INTEGER + ) + """) + end + + def down do + execute("DROP TABLE IF EXISTS enrollments") + end +end diff --git a/test/range_test.exs b/test/range_test.exs new file mode 100644 index 0000000..f94231d --- /dev/null +++ b/test/range_test.exs @@ -0,0 +1,250 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +defmodule AshSqlite.RangeTest do + @moduledoc """ + `Ash.Type.Range` on SQLite, checked against `Ash.Range` as the oracle. + + Every range predicate here is compiled to comparisons on `json_extract`, and + the thing that can quietly go wrong is not whether the SQL runs but whether it + answers the same question `Ash.Range` answers in Elixir. So each test asserts + agreement with `Ash.Range.intersects?/2`, `contains?/2` or `adjacent?/2` over a + set of pairs chosen to include the cases where inclusivity decides the answer: + ranges that merely touch, and ranges that touch with every combination of + bounds. + """ + use AshSqlite.RepoCase, async: false + + import Ash.Expr, only: [expr: 1] + + require Ash.Query + + alias AshSqlite.TestRepo + + defmodule Reservation do + @moduledoc false + use Ash.Resource, domain: nil, data_layer: AshSqlite.DataLayer + + sqlite do + table("range_reservations") + repo(AshSqlite.TestRepo) + end + + actions do + default_accept(:*) + defaults([:create, :read, :update, :destroy]) + end + + attributes do + uuid_primary_key(:id) + attribute(:label, :string, public?: true) + attribute(:window, Ash.Type.Range, constraints: [inner_type: :datetime], public?: true) + end + end + + defmodule Domain do + @moduledoc false + use Ash.Domain, validate_config_inclusion?: false + resources(do: resource(Reservation)) + end + + setup do + TestRepo.query!("DROP TABLE IF EXISTS range_reservations") + + TestRepo.query!(""" + CREATE TABLE range_reservations ( + id TEXT PRIMARY KEY, + label TEXT, + window TEXT + ) + """) + + :ok + end + + defp day(n), do: DateTime.new!(Date.new!(2026, 1, n), ~T[00:00:00], "Etc/UTC") + + defp range(lower, upper, bounds \\ :"[)") do + %Ash.Range{lower: lower, upper: upper, bounds: bounds} + end + + # The cases where inclusivity is the whole answer: [1,2) vs [2,3) do not + # overlap, but [1,2] vs [2,3) do, and the SQL has to distinguish them. + defp cases do + [ + {"jan1_jan2", range(day(1), day(2))}, + {"jan1_jan2_closed", range(day(1), day(2), :"[]")}, + {"jan2_jan3", range(day(2), day(3))}, + {"jan2_jan3_open_lower", range(day(2), day(3), :"(]")}, + {"jan1_jan5", range(day(1), day(5))}, + {"jan3_jan4", range(day(3), day(4))}, + {"unbounded_lower", range(nil, day(2))}, + {"unbounded_upper", range(day(4), nil)}, + {"unbounded_both", range(nil, nil)} + ] + end + + defp seed! do + for {label, window} <- cases() do + Reservation + |> Ash.Changeset.for_create(:create, %{label: label, window: window}, domain: Domain) + |> Ash.create!() + end + end + + defp labels_matching(filter) do + Reservation + |> Ash.Query.do_filter(filter) + |> Ash.read!(domain: Domain) + |> Enum.map(& &1.label) + |> Enum.sort() + end + + defp expected(oracle) do + cases() + |> Enum.filter(fn {_label, window} -> oracle.(window) end) + |> Enum.map(fn {label, _window} -> label end) + |> Enum.sort() + end + + describe "round trip" do + test "a range survives a write and a read" do + window = range(day(1), day(2), :"[]") + + created = + Reservation + |> Ash.Changeset.for_create(:create, %{label: "a", window: window}, domain: Domain) + |> Ash.create!() + + assert [read] = Ash.read!(Reservation, domain: Domain) + assert read.id == created.id + assert read.window == window + end + + test "an unbounded side round trips as nil, not as a sentinel" do + Reservation + |> Ash.Changeset.for_create(:create, %{label: "a", window: range(nil, nil)}, domain: Domain) + |> Ash.create!() + + assert [%{window: window}] = Ash.read!(Reservation, domain: Domain) + assert window.lower == nil + assert window.upper == nil + end + + test "bounds are stored at a single precision" do + # A range written at second precision and one written at microsecond + # precision must be stored identically, or they compare wrongly as text. + second = DateTime.new!(Date.new!(2026, 1, 1), ~T[00:00:00], "Etc/UTC") + usec = %{second | microsecond: {0, 6}} + + for {label, lower} <- [{"second", second}, {"usec", usec}] do + Reservation + |> Ash.Changeset.for_create(:create, %{label: label, window: range(lower, day(2))}, + domain: Domain + ) + |> Ash.create!() + end + + stored = + TestRepo.query!( + "SELECT json_extract(window, '$.lower') FROM range_reservations ORDER BY label" + ).rows + |> List.flatten() + |> Enum.uniq() + + assert stored == ["2026-01-01T00:00:00.000000Z"] + end + end + + describe "range_overlaps/2" do + test "agrees with Ash.Range.intersects?/2" do + seed!() + + for {label, probe} <- cases() do + assert labels_matching(expr(range_overlaps(window, ^probe))) == + expected(&Ash.Range.intersects?(&1, probe)), + "range_overlaps disagreed with Ash.Range.intersects?/2 for #{label}" + end + end + + test "a null range matches nothing" do + Reservation + |> Ash.Changeset.for_create(:create, %{label: "no_window", window: nil}, domain: Domain) + |> Ash.create!() + + assert labels_matching(expr(range_overlaps(window, ^range(day(1), day(9))))) == [] + end + end + + describe "range_contains/2" do + test "range in range agrees with Ash.Range.contains?/2" do + seed!() + + for {label, probe} <- cases() do + assert labels_matching(expr(range_contains(window, ^probe))) == + expected(&Ash.Range.contains?(&1, probe)), + "range_contains disagreed with Ash.Range.contains?/2 for #{label}" + end + end + + test "point in range agrees with Ash.Range.contains?/2" do + seed!() + + for point <- [day(1), day(2), day(3), day(4), day(5)] do + assert labels_matching(expr(range_contains(window, ^point))) == + expected(&Ash.Range.contains?(&1, point)), + "range_contains disagreed with Ash.Range.contains?/2 for point #{point}" + end + end + + test "a point at second precision is not treated as a different instant" do + # The adapter would encode this as "...00:00:00Z" and the bound as + # "...00:00:00.000000Z", which compare unequal as text. + Reservation + |> Ash.Changeset.for_create( + :create, + %{label: "a", window: range(%{day(1) | microsecond: {0, 6}}, day(5))}, + domain: Domain + ) + |> Ash.create!() + + point = DateTime.new!(Date.new!(2026, 1, 1), ~T[00:00:00], "Etc/UTC") + + assert labels_matching(expr(range_contains(window, ^point))) == ["a"] + end + end + + describe "range_adjacent/2" do + test "agrees with Ash.Range.adjacent?/2" do + seed!() + + for {label, probe} <- cases() do + assert labels_matching(expr(range_adjacent(window, ^probe))) == + expected(&Ash.Range.adjacent?(&1, probe)), + "range_adjacent disagreed with Ash.Range.adjacent?/2 for #{label}" + end + end + end + + describe "range_lower/1 and range_upper/1" do + test "select a range's bounds" do + Reservation + |> Ash.Changeset.for_create(:create, %{label: "a", window: range(day(1), day(2))}, + domain: Domain + ) + |> Ash.create!() + + assert [%{lower: lower, upper: upper}] = + Reservation + |> Ash.Query.select([:id]) + |> Ash.Query.calculate(:lower, :datetime, expr(range_lower(window))) + |> Ash.Query.calculate(:upper, :datetime, expr(range_upper(window))) + |> Ash.read!(domain: Domain) + |> Enum.map(&%{lower: &1.calculations.lower, upper: &1.calculations.upper}) + + assert DateTime.compare(lower, day(1)) == :eq + assert DateTime.compare(upper, day(2)) == :eq + end + end +end diff --git a/test/support/domain.ex b/test/support/domain.ex index 1f8abf4..7975a1a 100644 --- a/test/support/domain.ex +++ b/test/support/domain.ex @@ -22,6 +22,14 @@ defmodule AshSqlite.Test.Domain do resource(AshSqlite.Test.Organization) resource(AshSqlite.Test.Manager) resource(AshSqlite.Test.Device) + + # The temporal DSL is unreleased; the resource only exists when ash has it. + if Code.ensure_loaded?(Ash.Temporal) do + resource(AshSqlite.Test.Subscription) + resource(AshSqlite.Test.DatedSubscription) + resource(AshSqlite.Test.Plan) + resource(AshSqlite.Test.Enrollment) + end end authorization do diff --git a/test/support/resources/dated_subscription.ex b/test/support/resources/dated_subscription.ex new file mode 100644 index 0000000..4235eff --- /dev/null +++ b/test/support/resources/dated_subscription.ex @@ -0,0 +1,53 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +# The temporal DSL is unreleased. Skip this entirely when ash does not have it. +if Code.ensure_loaded?(Ash.Temporal) do + defmodule AshSqlite.Test.DatedSubscription do + @moduledoc """ + A temporal resource whose period is a `:date` rather than a `:datetime`. + + `Ash.Type.Range` allows `:date`, `:integer`, `:naive_datetime` and `:datetime` + bounds, and only `:datetime` has a `DateTime.compare/2`. This resource is here so + the split is exercised on a bound type that does not. + """ + use Ash.Resource, + domain: AshSqlite.Test.Domain, + data_layer: AshSqlite.DataLayer + + sqlite do + table("dated_subscriptions") + repo(AshSqlite.TransactionTestRepo) + end + + temporal do + strategy(:context) + attribute(:valid_on) + end + + attributes do + attribute(:id, :integer, primary_key?: true, allow_nil?: false, public?: true) + attribute(:tier, :string, public?: true) + + attribute(:valid_on, Ash.Type.Range, + allow_nil?: false, + constraints: [ + inner_type: :date, + lower: [inclusive?: true], + upper: [inclusive?: false] + ], + public?: true + ) + end + + actions do + defaults([:read, :destroy, create: [:id, :tier]]) + + update :change_tier do + require_atomic?(true) + accept([:tier]) + end + end + end +end diff --git a/test/support/resources/enrollment.ex b/test/support/resources/enrollment.ex new file mode 100644 index 0000000..6b4c902 --- /dev/null +++ b/test/support/resources/enrollment.ex @@ -0,0 +1,50 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +# The temporal DSL is unreleased. Skip this entirely when ash does not have it. +if Code.ensure_loaded?(Ash.Temporal) do + defmodule AshSqlite.Test.Enrollment do + @moduledoc """ + A non-temporal resource pointing at a temporal one. + + `temporal_keys {nil, :valid_at}` is how a relationship declares that only its + destination keeps periods. `Ash.Resource.Transformers.AddTemporalRelationshipFilters` + bakes its `range_overlaps(parent(source), destination)` filter only when *both* sides + are given, so this shape needs no `parent/1` and therefore no lateral join, which + SQLite has no equivalent of. + + No database foreign key is generated for the relationship. A temporal table has no + unique key for one to point at, and SQLite rejects every insert into the child with + "foreign key mismatch" if one is declared anyway. + """ + use Ash.Resource, + domain: AshSqlite.Test.Domain, + data_layer: AshSqlite.DataLayer + + sqlite do + table("enrollments") + repo(AshSqlite.TransactionTestRepo) + end + + attributes do + attribute(:id, :integer, primary_key?: true, allow_nil?: false, public?: true) + attribute(:plan_id, :integer, public?: true) + end + + relationships do + belongs_to :plan, AshSqlite.Test.Plan do + source_attribute(:plan_id) + destination_attribute(:id) + define_attribute?(false) + attribute_type(:integer) + temporal_keys({nil, :valid_at}) + public?(true) + end + end + + actions do + defaults([:read, :destroy, create: [:id, :plan_id]]) + end + end +end diff --git a/test/support/resources/plan.ex b/test/support/resources/plan.ex new file mode 100644 index 0000000..f661ab8 --- /dev/null +++ b/test/support/resources/plan.ex @@ -0,0 +1,60 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +# The temporal DSL is unreleased. Skip this entirely when ash does not have it. +if Code.ensure_loaded?(Ash.Temporal) do + defmodule AshSqlite.Test.Plan do + @moduledoc """ + A temporal resource with an identity that is not its primary key. + + Postgres writes such an identity as `UNIQUE (slug, valid_at WITHOUT OVERLAPS)`, so + the slug is unique *at every instant* rather than unique in the table. This resource + exists to check that the partial index and trigger pair `AshSqlite.Temporal.Migration` + generates per identity reach the same guarantee. + """ + use Ash.Resource, + domain: AshSqlite.Test.Domain, + data_layer: AshSqlite.DataLayer + + sqlite do + table("plans") + repo(AshSqlite.TransactionTestRepo) + end + + temporal do + strategy(:context) + attribute(:valid_at) + end + + attributes do + attribute(:id, :integer, primary_key?: true, allow_nil?: false, public?: true) + attribute(:slug, :string, allow_nil?: false, public?: true) + attribute(:price, :integer, default: 0, public?: true) + + attribute(:valid_at, Ash.Type.Range, + allow_nil?: false, + constraints: [ + inner_type: :datetime, + inner_constraints: [precision: :microsecond], + lower: [inclusive?: true], + upper: [inclusive?: false] + ], + public?: true + ) + end + + identities do + identity(:unique_slug, [:slug]) + end + + actions do + defaults([:read, :destroy, create: [:id, :slug, :price]]) + + update :change_price do + require_atomic?(true) + accept([:price]) + end + end + end +end diff --git a/test/support/resources/subscription.ex b/test/support/resources/subscription.ex new file mode 100644 index 0000000..947a29d --- /dev/null +++ b/test/support/resources/subscription.ex @@ -0,0 +1,76 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +# The temporal DSL is unreleased. Skip these entirely when ash does not have it, +# rather than failing to compile against a released version. +if Code.ensure_loaded?(Ash.Temporal) do + defmodule AshSqlite.Test.Subscription do + @moduledoc false + use Ash.Resource, + domain: AshSqlite.Test.Domain, + data_layer: AshSqlite.DataLayer + + sqlite do + table("subscriptions") + repo(AshSqlite.TransactionTestRepo) + end + + temporal do + strategy(:context) + attribute(:valid_at) + end + + attributes do + attribute(:id, :integer, primary_key?: true, allow_nil?: false, public?: true) + attribute(:tier, :string, public?: true) + attribute(:seats, :integer, default: 0, public?: true) + attribute(:activated_at, :utc_datetime_usec, public?: true) + + attribute(:valid_at, Ash.Type.Range, + allow_nil?: false, + constraints: [ + inner_type: :datetime, + inner_constraints: [precision: :microsecond], + lower: [inclusive?: true], + upper: [inclusive?: false] + ], + public?: true + ) + end + + actions do + defaults([:read, :destroy, create: [:id, :tier, :seats, :activated_at]]) + + create :upsert_tier do + accept([:id, :tier, :seats]) + upsert?(true) + upsert_identity(:id) + upsert_fields([:tier, :seats]) + end + + update :change_tier do + require_atomic?(true) + accept([:tier]) + end + + update :change_tier_nonatomic do + require_atomic?(false) + accept([:tier]) + end + + update :add_seat do + require_atomic?(true) + change(atomic_update(:seats, expr(seats + 1))) + end + + destroy :expire do + require_atomic?(true) + end + end + + identities do + identity(:id, [:id]) + end + end +end diff --git a/test/temporal_test.exs b/test/temporal_test.exs new file mode 100644 index 0000000..f92cedf --- /dev/null +++ b/test/temporal_test.exs @@ -0,0 +1,1045 @@ +# SPDX-FileCopyrightText: 2023 ash_sqlite contributors +# +# SPDX-License-Identifier: MIT + +# The temporal DSL is unreleased. Skip these entirely when ash does not have it, +# rather than failing to compile against a released version. +if Code.ensure_loaded?(Ash.Temporal) do + defmodule AshSqlite.TemporalTest do + @moduledoc """ + The behaviour contract for a temporal resource on SQLite. + + SQLite has no `FOR PORTION OF`, so every period split is a read-modify-write the + data layer performs inside a transaction. These tests assert the *outcome* of that + split, which is the same outcome Postgres reaches with one statement, and separately + assert the two constraints SQLite can enforce underneath it. + """ + use AshSqlite.RepoCase, async: false + + alias AshSqlite.Test.Subscription + + require Ash.Query + require Ash.Expr + + import Ash.Expr + + setup do + :ok = Ecto.Adapters.SQL.Sandbox.checkout(AshSqlite.TransactionTestRepo) + Ecto.Adapters.SQL.Sandbox.mode(AshSqlite.TransactionTestRepo, {:shared, self()}) + :ok + end + + @jan ~U[2026-01-01 00:00:00.000000Z] + @feb ~U[2026-02-01 00:00:00.000000Z] + @mar ~U[2026-03-01 00:00:00.000000Z] + @apr ~U[2026-04-01 00:00:00.000000Z] + + defp create!(id, tier, as_of, opts \\ []) do + Subscription + |> Ash.Changeset.for_create(:create, Keyword.merge([id: id, tier: tier], opts)) + |> Ash.Changeset.set_context(%{}) + |> Ash.Changeset.as_of(as_of) + |> Ash.create!() + end + + # Every version of a record, oldest first. This reads the table directly: a temporal + # read is always pinned to an instant, so there is no Ash query that returns history. + defp periods(id) do + id + |> raw_rows() + |> Enum.map(fn [lower, upper, tier] -> {tier, lower, upper} end) + end + + defp raw_seats(id) do + {:ok, %{rows: rows}} = + AshSqlite.TransactionTestRepo.query( + "select json_extract(valid_at,'$.lower'), seats from subscriptions " <> + "where id = ? order by json_extract(valid_at,'$.lower')", + [id] + ) + + Enum.map(rows, fn [lower, seats] -> {lower, seats} end) + end + + defp iso(nil), do: nil + defp iso(%DateTime{} = value), do: DateTime.to_iso8601(value) + + # Every version of a record, ignoring the default "as of now" filter. A temporal + # read never returns history, so the raw table is the only way to assert a split. + defp raw_rows(id) do + {:ok, %{rows: rows}} = + AshSqlite.TransactionTestRepo.query( + "select json_extract(valid_at,'$.lower'), json_extract(valid_at,'$.upper'), tier " <> + "from subscriptions where id = ? order by json_extract(valid_at,'$.lower')", + [id] + ) + + rows + end + + describe "capability" do + test "the data layer reports temporal support" do + assert Ash.DataLayer.can?(AshSqlite.DataLayer, Subscription, :temporal) + end + + test "the resource is temporal and names its period attribute" do + assert Ash.Resource.Info.temporal?(Subscription) + assert Ash.Resource.Info.temporal_attribute(Subscription) == :valid_at + end + end + + test "a temporal resource on a repo without write transactions is refused" do + # Spark runs verifiers from `@after_verify`, so the error surfaces from the + # parallel checker rather than from `defmodule` itself. Calling the verifier is + # what actually asserts the rule, rather than asserting on compilation order. + assert {:error, %Spark.Error.DslError{} = error} = + AshSqlite.Verifiers.VerifyTemporal.verify( + dsl_with_repo(AshSqlite.Test.Subscription, AshSqlite.TestRepo) + ) + + assert Exception.message(error) =~ "requires a repo with write transactions" + end + + test "a temporal resource on a repo with write transactions is accepted" do + assert :ok = + AshSqlite.Verifiers.VerifyTemporal.verify( + AshSqlite.Test.Subscription.spark_dsl_config() + ) + end + + test "a non-temporal resource is accepted on any repo" do + assert :ok = + AshSqlite.Verifiers.VerifyTemporal.verify(AshSqlite.Test.Post.spark_dsl_config()) + end + + defp dsl_with_repo(resource, repo) do + resource + |> then(& &1.spark_dsl_config()) + |> Spark.Dsl.Transformer.set_option([:sqlite], :repo, repo) + end + + describe "create" do + test "stamps the period from as_of, unbounded above" do + record = create!(1, "free", @jan) + + assert record.valid_at.lower == @jan + assert is_nil(record.valid_at.upper) + end + + test "a create without as_of takes effect now" do + before = DateTime.utc_now() + + record = + Subscription + |> Ash.Changeset.for_create(:create, %{id: 2, tier: "free"}) + |> Ash.create!() + + assert DateTime.compare(record.valid_at.lower, before) in [:gt, :eq] + assert is_nil(record.valid_at.upper) + end + + test "the period attribute is not accepted as input" do + assert_raise Ash.Error.Invalid, ~r/valid_at/, fn -> + Subscription + |> Ash.Changeset.for_create(:create, %{id: 3, tier: "free", valid_at: %{}}) + |> Ash.create!() + end + end + end + + describe "reads default to the current version" do + setup do + create!(10, "free", @jan) + :ok + end + + test "a read with no as_of returns the version valid now" do + assert [%{tier: "free"}] = Ash.read!(Subscription) + end + + test "a read returns one row per key, not the whole history" do + Subscription + |> Ash.get!(10) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.Changeset.as_of(@feb) + |> Ash.update!() + + assert length(Ash.read!(Subscription)) == 1 + end + end + + describe "as_of reads" do + setup do + create!(20, "free", @jan) + + Subscription + |> Ash.get!(20) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + :ok + end + + test "returns the version valid at that instant" do + assert [%{tier: "free"}] = + Subscription |> Ash.Query.as_of(@feb) |> Ash.read!() + + assert [%{tier: "pro"}] = + Subscription |> Ash.Query.as_of(@apr) |> Ash.read!() + end + + test "returns nothing before the first period begins" do + assert [] = + Subscription + |> Ash.Query.as_of(~U[2025-12-01 00:00:00.000000Z]) + |> Ash.read!() + end + + test "the lower bound is inclusive and the upper bound is exclusive" do + assert [%{tier: "free"}] = Subscription |> Ash.Query.as_of(@jan) |> Ash.read!() + assert [%{tier: "pro"}] = Subscription |> Ash.Query.as_of(@mar) |> Ash.read!() + end + + test "Ash.get respects as_of" do + assert %{tier: "free"} = Ash.get!(Subscription, 20, as_of: @feb) + end + end + + describe "update splits the period at as_of" do + test "an atomic update closes the prior version and opens a new one" do + create!(30, "free", @jan) + + Subscription + |> Ash.get!(30) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + assert periods(30) == [{"free", iso(@jan), iso(@mar)}, {"pro", iso(@mar), nil}] + end + + test "a non-atomic update splits the same way" do + create!(31, "free", @jan) + + Subscription + |> Ash.get!(31) + |> Ash.Changeset.for_update(:change_tier_nonatomic, %{tier: "pro"}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + assert periods(31) == [{"free", iso(@jan), iso(@mar)}, {"pro", iso(@mar), nil}] + end + + test "an atomic arithmetic update applies to the new slice only" do + create!(32, "free", @jan, seats: 3) + + Subscription + |> Ash.get!(32) + |> Ash.Changeset.for_update(:add_seat) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + seats = + 32 + |> raw_seats() + |> Enum.map(&elem(&1, 1)) + + assert seats == [3, 4] + end + + test "two successive updates leave three contiguous periods" do + create!(33, "free", @jan) + + for {tier, at} <- [{"pro", @feb}, {"max", @mar}] do + Subscription + |> Ash.get!(33) + |> Ash.Changeset.for_update(:change_tier, %{tier: tier}) + |> Ash.Changeset.as_of(at) + |> Ash.update!() + end + + assert periods(33) == [ + {"free", iso(@jan), iso(@feb)}, + {"pro", iso(@feb), iso(@mar)}, + {"max", iso(@mar), nil} + ] + end + + test "an update at the period's own lower bound replaces it rather than leaving an empty period" do + create!(34, "free", @jan) + + Subscription + |> Ash.get!(34) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.Changeset.as_of(@jan) + |> Ash.update!() + + assert periods(34) == [{"pro", iso(@jan), nil}] + end + end + + describe "destroy truncates validity" do + test "a destroy ends the current period at as_of and keeps the history" do + create!(40, "free", @jan) + + Subscription + |> Ash.get!(40) + |> Ash.Changeset.for_destroy(:expire) + |> Ash.Changeset.as_of(@mar) + |> Ash.destroy!() + + assert periods(40) == [{"free", iso(@jan), iso(@mar)}] + assert [] = Subscription |> Ash.Query.as_of(@apr) |> Ash.read!() + assert [%{tier: "free"}] = Subscription |> Ash.Query.as_of(@feb) |> Ash.read!() + end + + test "destroying at the period's lower bound removes the row" do + create!(41, "free", @jan) + + Subscription + |> Ash.get!(41) + |> Ash.Changeset.for_destroy(:expire) + |> Ash.Changeset.as_of(@jan) + |> Ash.destroy!() + + assert periods(41) == [] + end + end + + describe "upsert" do + test "a miss inserts a new period" do + Subscription + |> Ash.Changeset.for_create(:upsert_tier, %{id: 50, tier: "free"}) + |> Ash.Changeset.as_of(@jan) + |> Ash.create!() + + assert periods(50) == [{"free", iso(@jan), nil}] + end + + test "a match splits the period valid at as_of" do + Subscription + |> Ash.Changeset.for_create(:upsert_tier, %{id: 51, tier: "free"}) + |> Ash.Changeset.as_of(@jan) + |> Ash.create!() + + Subscription + |> Ash.Changeset.for_create(:upsert_tier, %{id: 51, tier: "pro"}) + |> Ash.Changeset.as_of(@mar) + |> Ash.create!() + + assert periods(51) == [{"free", iso(@jan), iso(@mar)}, {"pro", iso(@mar), nil}] + end + end + + describe "bulk actions" do + test "bulk_create stamps each record's period from as_of" do + Ash.bulk_create!( + [%{id: 60, tier: "free"}, %{id: 61, tier: "pro"}], + Subscription, + :create, + as_of: @jan, + return_errors?: true + ) + + assert periods(60) == [{"free", iso(@jan), nil}] + assert periods(61) == [{"pro", iso(@jan), nil}] + end + + test "a bulk update splits every matched record at as_of" do + create!(62, "free", @jan) + create!(63, "free", @jan) + + Subscription + |> Ash.Query.filter(tier == "free") + |> Ash.bulk_update!(:change_tier, %{tier: "pro"}, + as_of: @mar, + strategy: [:stream, :atomic, :atomic_batches], + return_errors?: true + ) + + assert periods(62) == [{"free", iso(@jan), iso(@mar)}, {"pro", iso(@mar), nil}] + assert periods(63) == [{"free", iso(@jan), iso(@mar)}, {"pro", iso(@mar), nil}] + end + + test "a bulk destroy truncates every matched record at as_of" do + create!(64, "free", @jan) + create!(65, "free", @jan) + + Subscription + |> Ash.Query.filter(tier == "free") + |> Ash.bulk_destroy!(:expire, %{}, + as_of: @mar, + strategy: [:stream, :atomic, :atomic_batches], + return_errors?: true + ) + + assert periods(64) == [{"free", iso(@jan), iso(@mar)}] + assert periods(65) == [{"free", iso(@jan), iso(@mar)}] + end + end + + describe "the database refuses a broken timeline" do + test "a second open-ended period for one key is rejected by the partial unique index" do + create!(70, "free", @jan) + + assert {:error, _} = + AshSqlite.TransactionTestRepo.query( + "insert into subscriptions (id, tier, seats, valid_at) values (?, ?, ?, ?)", + [ + 70, + "pro", + 0, + ~s|{"lower":"2026-03-01T00:00:00.000000Z","upper":null,"bounds":"[)","empty":false}| + ] + ) + end + + test "an overlapping closed period is rejected by the non-overlap trigger" do + create!(71, "free", @jan) + + Subscription + |> Ash.get!(71) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + assert {:error, _} = + AshSqlite.TransactionTestRepo.query( + "insert into subscriptions (id, tier, seats, valid_at) values (?, ?, ?, ?)", + [ + 71, + "mid", + 0, + ~s|{"lower":"2026-01-15T00:00:00.000000Z","upper":"2026-02-15T00:00:00.000000Z","bounds":"[)","empty":false}| + ] + ) + end + + test "an adjacent, non-overlapping period is accepted" do + create!(72, "free", @jan) + + Subscription + |> Ash.get!(72) + |> Ash.Changeset.for_destroy(:expire) + |> Ash.Changeset.as_of(@feb) + |> Ash.destroy!() + + assert {:ok, _} = + AshSqlite.TransactionTestRepo.query( + "insert into subscriptions (id, tier, seats, valid_at) values (?, ?, ?, ?)", + [ + 72, + "pro", + 0, + ~s|{"lower":"2026-02-01T00:00:00.000000Z","upper":null,"bounds":"[)","empty":false}| + ] + ) + end + + test "a failed split leaves no partial write" do + create!(73, "free", @jan) + + # An update whose insert half violates the trigger must roll the close back too, + # or the record is left with no current version at all. + assert periods(73) == [{"free", iso(@jan), nil}] + end + end + + describe "now() anchors to as_of" do + test "a filter on now() is evaluated at the read's as_of" do + create!(80, "free", @jan, activated_at: @feb) + + assert [] = + Subscription + |> Ash.Query.as_of(@jan) + |> Ash.Query.filter(activated_at < now()) + |> Ash.read!() + + assert [%{id: 80}] = + Subscription + |> Ash.Query.as_of(@mar) + |> Ash.Query.filter(activated_at < now()) + |> Ash.read!() + end + end + + describe "range expressions over the period" do + test "range_overlaps filters by overlap with a literal range" do + create!(90, "free", @jan) + + overlapping = %Ash.Range{lower: @jan, upper: @feb, bounds: :"[)"} + + disjoint = %Ash.Range{ + lower: ~U[2025-01-01 00:00:00.000000Z], + upper: ~U[2025-06-01 00:00:00.000000Z], + bounds: :"[)" + } + + assert [%{id: 90}] = + Subscription + |> Ash.Query.as_of(@jan) + |> Ash.Query.filter(id == 90 and range_overlaps(valid_at, ^overlapping)) + |> Ash.read!() + + assert [] = + Subscription + |> Ash.Query.as_of(@jan) + |> Ash.Query.filter(id == 90 and range_overlaps(valid_at, ^disjoint)) + |> Ash.read!() + end + + test "range_lower selects the period's lower bound" do + create!(91, "free", @jan) + + assert [@jan] = + Subscription + |> Ash.Query.filter(id == 91) + |> Ash.Query.calculate(:lo, :utc_datetime_usec, expr(range_lower(valid_at))) + |> Ash.read!() + |> Enum.map(& &1.calculations.lo) + end + end + + describe "the split is one transaction" do + test "the close and the insert are both visible or neither is" do + create!(100, "free", @jan) + + assert length(raw_rows(100)) == 1 + + Subscription + |> Ash.get!(100) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + assert length(raw_rows(100)) == 2 + end + end + + describe "the migration generator emits what a temporal table needs" do + @describetag :tmp_dir + + setup %{tmp_dir: tmp_dir} do + %{ + snapshot_path: Path.join(tmp_dir, "snapshots"), + migration_path: Path.join(tmp_dir, "migrations") + } + end + + defp generated_migration(snapshot_path, migration_path) do + AshSqlite.MigrationGenerator.generate(AshSqlite.Test.Domain, + snapshot_path: snapshot_path, + migration_path: migration_path, + quiet: true, + format: false, + auto_name: true + ) + + migration_path + |> Path.join("**/*_migrate_resources*.exs") + |> Path.wildcard() + |> Enum.map_join("\n", &File.read!/1) + end + + test "a point-in-time index, a current-version constraint and both triggers", %{ + snapshot_path: snapshot_path, + migration_path: migration_path + } do + sql = generated_migration(snapshot_path, migration_path) + + assert sql =~ + ~s|CREATE INDEX "subscriptions_valid_at_pit" ON "subscriptions" ("id", json_extract("valid_at", '$.lower'))| + + assert sql =~ + ~s|CREATE UNIQUE INDEX "subscriptions_valid_at_current" ON "subscriptions" ("id") WHERE json_extract("valid_at", '$.upper') IS NULL| + + assert sql =~ ~s|CREATE TRIGGER "subscriptions_valid_at_no_overlap_insert"| + assert sql =~ ~s|CREATE TRIGGER "subscriptions_valid_at_no_overlap_update"| + end + + test "the insert trigger does not exclude NEW.rowid and the update trigger does", %{ + snapshot_path: snapshot_path, + migration_path: migration_path + } do + sql = generated_migration(snapshot_path, migration_path) + + [_, insert_trigger, update_trigger] = + String.split( + sql, + ~r/CREATE TRIGGER "subscriptions_valid_at_no_overlap_(insert|update)"/ + ) + + refute insert_trigger =~ "NEW.rowid" + assert update_trigger =~ ~s|other.rowid <> NEW.rowid| + end + + test "the table has no primary key and no plain unique index", %{ + snapshot_path: snapshot_path, + migration_path: migration_path + } do + sql = generated_migration(snapshot_path, migration_path) + + # A temporal table holds one row per period. A PRIMARY KEY or a plain unique + # index over the key makes a second version impossible, and on an integer key a + # PRIMARY KEY also makes it SQLite's rowid alias, which every `WHERE rowid = ?` + # in AshSqlite.Temporal would then be addressing. + assert String.contains?(sql, "create table(:subscriptions, primary_key: false)") + refute String.contains?(sql, "subscriptions_id_index") + + subscriptions_block = + sql + |> String.split("create table(:subscriptions") + |> Enum.at(1) + |> String.split("\nend\n") + |> Enum.at(0) + + refute String.contains?(subscriptions_block, "primary_key: true") + end + + test "a non-temporal resource keeps its primary key", %{ + snapshot_path: snapshot_path, + migration_path: migration_path + } do + sql = generated_migration(snapshot_path, migration_path) + + assert sql =~ "create table(:posts, primary_key: false)" + assert sql =~ "primary_key: true" + end + + test "applying the generated statements leaves a table that holds two periods" do + {:ok, db} = Exqlite.Sqlite3.open(":memory:") + + :ok = + Exqlite.Sqlite3.execute(db, """ + CREATE TABLE subscriptions ( + id INTEGER NOT NULL, tier TEXT, seats INTEGER, activated_at TEXT, valid_at TEXT NOT NULL + ) + """) + + for %{up: up} <- AshSqlite.Temporal.Migration.statements(Subscription) do + :ok = Exqlite.Sqlite3.execute(db, up) + end + + insert = fn lower, upper -> + period = Jason.encode!(%{lower: lower, upper: upper, bounds: "[)", empty: false}) + + Exqlite.Sqlite3.execute( + db, + "INSERT INTO subscriptions (id, tier, seats, valid_at) VALUES " <> + "(1, 'free', 0, '" <> period <> "')" + ) + end + + assert :ok = insert.("2026-01-01", "2026-03-01") + assert :ok = insert.("2026-03-01", nil) + assert {:error, _} = insert.("2026-02-01", "2026-04-01") + assert {:error, _} = insert.("2026-05-01", nil) + end + + test "a non-temporal resource gets none of it" do + assert AshSqlite.Temporal.Migration.statements(AshSqlite.Test.Post) == [] + end + end + + describe "period bound types other than datetime" do + # `Ash.Temporal.raw_instant/2` accepts only a `DateTime` or `:now`, so a `:date` + # resource cannot be handed an explicit `Date` as_of. It still reaches a `Date` + # bound through `:now`, because `Ash.Temporal.now_for/1` returns `Date.utc_today/0` + # for a `:date` inner type. That is the path these cover. + setup do + {:ok, _} = AshSqlite.TransactionTestRepo.query("DELETE FROM dated_subscriptions", []) + :ok + end + + test "closing a :date period does not compare it as a DateTime" do + # Seeded directly so the prior period begins before today and the write splits it. + yesterday = Date.add(Date.utc_today(), -1) + + {:ok, _} = + AshSqlite.TransactionTestRepo.query( + "insert into dated_subscriptions (id, tier, valid_on) values (?, ?, ?)", + [1, "free", ~s|{"lower":"#{yesterday}","upper":null,"bounds":"[)","empty":false}|] + ) + + AshSqlite.Test.DatedSubscription + |> Ash.get!(1) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.update!() + + {:ok, %{rows: rows}} = + AshSqlite.TransactionTestRepo.query( + "select json_extract(valid_on,'$.lower'), json_extract(valid_on,'$.upper'), tier " <> + "from dated_subscriptions where id = 1 order by json_extract(valid_on,'$.lower')", + [] + ) + + today = to_string(Date.utc_today()) + + assert rows == [ + [to_string(yesterday), today, "free"], + [today, nil, "pro"] + ] + end + + test "a write landing on a :date period's own lower bound replaces it" do + AshSqlite.Test.DatedSubscription + |> Ash.Changeset.for_create(:create, %{id: 2, tier: "free"}) + |> Ash.create!() + + AshSqlite.Test.DatedSubscription + |> Ash.get!(2) + |> Ash.Changeset.for_update(:change_tier, %{tier: "pro"}) + |> Ash.update!() + + {:ok, %{rows: rows}} = + AshSqlite.TransactionTestRepo.query( + "select json_extract(valid_on,'$.lower'), json_extract(valid_on,'$.upper'), tier " <> + "from dated_subscriptions where id = 2", + [] + ) + + assert rows == [[to_string(Date.utc_today()), nil, "pro"]] + end + end + + describe "a limited bulk update" do + test "splits only the rows it updates" do + for id <- 200..204, do: create!(id, "free", @jan) + + Subscription + |> Ash.Query.filter(tier == "free" and id >= 200 and id <= 204) + |> Ash.Query.limit(2) + |> Ash.Query.sort(id: :asc) + |> Ash.bulk_update!(:change_tier, %{tier: "pro"}, + as_of: @mar, + strategy: [:stream, :atomic, :atomic_batches], + return_errors?: true + ) + + split = Enum.count(200..204, fn id -> length(periods(id)) == 2 end) + + assert split == 2, + "expected exactly the two limited rows to be split, got #{split}: " <> + inspect(Enum.map(200..204, &{&1, periods(&1)})) + end + end + + describe "the set-wide update targets only the versions it split" do + test "a limited bulk update leaves earlier versions of the matched record alone" do + create!(300, "free", @jan) + + Subscription + |> Ash.get!(300) + |> Ash.Changeset.for_update(:change_tier, %{tier: "basic"}) + |> Ash.Changeset.as_of(@feb) + |> Ash.update!() + + create!(301, "free", @jan) + + # A limit forces `bulk_updatable_query/6` to rewrite the statement as a join + # against a subquery. Joined on the primary key alone that reaches every version + # of the record, including the closed one. + Subscription + |> Ash.Query.filter(id in [300, 301]) + |> Ash.Query.sort(id: :asc) + |> Ash.Query.limit(1) + |> Ash.bulk_update!(:change_tier, %{tier: "pro"}, + as_of: @mar, + strategy: [:atomic], + return_errors?: true + ) + + assert periods(300) == [ + {"free", iso(@jan), iso(@feb)}, + {"basic", iso(@feb), iso(@mar)}, + {"pro", iso(@mar), nil} + ] + + assert periods(301) == [{"free", iso(@jan), nil}] + end + end + + describe "a query pinned to one instant and a write at another" do + test "the write takes effect at the instant the query is pinned to" do + create!(310, "free", @jan) + + Subscription + |> Ash.Query.filter(id == 310) + |> Ash.Query.as_of(@feb) + |> Ash.bulk_update!(:change_tier, %{tier: "pro"}, + as_of: @mar, + strategy: [:atomic], + return_errors?: true + ) + + # The update must land somewhere rather than silently matching nothing. + assert periods(310) == [{"free", iso(@jan), iso(@feb)}, {"pro", iso(@feb), nil}] + end + end + + describe "bulk upsert" do + test "Ash.bulk_create with upsert? splits a match and inserts a miss" do + create!(320, "free", @jan) + + Ash.bulk_create!( + [%{id: 320, tier: "pro"}, %{id: 321, tier: "new"}], + Subscription, + :upsert_tier, + as_of: @mar, + return_errors?: true + ) + + assert periods(320) == [{"free", iso(@jan), iso(@mar)}, {"pro", iso(@mar), nil}] + assert periods(321) == [{"new", iso(@mar), nil}] + end + end + + describe "the record an upsert returns" do + test "carries the attributes the upsert did not name" do + create!(330, "free", @jan, seats: 9, activated_at: @feb) + + record = + Subscription + |> Ash.Changeset.for_create(:upsert_tier, %{id: 330, tier: "pro"}) + |> Ash.Changeset.as_of(@mar) + |> Ash.create!() + + assert record.tier == "pro" + assert record.activated_at == @feb + end + end + + describe "identity is unique at every instant, not unique in the table" do + alias AshSqlite.Test.Plan + + setup do + {:ok, _} = AshSqlite.TransactionTestRepo.query("DELETE FROM plans", []) + :ok + end + + defp plan!(id, slug, as_of) do + Plan + |> Ash.Changeset.for_create(:create, %{id: id, slug: slug}) + |> Ash.Changeset.as_of(as_of) + |> Ash.create!() + end + + test "two records cannot hold the same identity at the same instant" do + plan!(1, "pro", @jan) + + assert_raise Ash.Error.Unknown, ~r/overlaps an existing period/, fn -> + plan!(2, "pro", @feb) + end + end + + test "two records can hold the same identity at instants that do not overlap" do + plan!(1, "pro", @jan) + + Plan + |> Ash.get!(1) + |> Ash.Changeset.for_destroy(:destroy) + |> Ash.Changeset.as_of(@feb) + |> Ash.destroy!() + + # The slug is free from February onwards, so another record may take it. + assert %{id: 2} = plan!(2, "pro", @mar) + end + + test "a record's own split does not collide with its own identity" do + plan!(3, "solo", @jan) + + Plan + |> Ash.get!(3) + |> Ash.Changeset.for_update(:change_price, %{price: 10}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + {:ok, %{rows: rows}} = + AshSqlite.TransactionTestRepo.query( + "select json_extract(valid_at,'$.lower'), json_extract(valid_at,'$.upper'), price " <> + "from plans where id = 3 order by json_extract(valid_at,'$.lower')", + [] + ) + + assert rows == [ + [iso(@jan), iso(@mar), 0], + [iso(@mar), nil, 10] + ] + end + + test "the primary key still identifies the record, and the period is not part of it" do + plan!(4, "keyed", @jan) + + assert Ash.Resource.Info.primary_key(Plan) == [:id] + + assert %{primary_key?: false, generated?: true} = + Ash.Resource.Info.attribute(Plan, :valid_at) + + # One id, many rows: `Ash.get!` resolves it to the version valid at the instant. + Plan + |> Ash.get!(4) + |> Ash.Changeset.for_update(:change_price, %{price: 99}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + assert %{price: 0} = Ash.get!(Plan, 4, as_of: @feb) + assert %{price: 99} = Ash.get!(Plan, 4, as_of: @apr) + end + + test "the generator emits a partial index and a trigger pair per identity" do + names = Enum.map(AshSqlite.Temporal.Migration.statements(Plan), & &1.name) + + assert :plans_valid_at_current in names + assert :plans_valid_at_current_slug in names + assert :plans_valid_at_no_overlap_slug_insert in names + assert :plans_valid_at_no_overlap_slug_update in names + + # The trigger for an identity is a correlated subquery over that identity's + # keys, so it needs its own index or every write scans the table. + assert :plans_valid_at_pit_slug in names + end + + test "the trigger for an identity seeks rather than scans" do + {:ok, db} = Exqlite.Sqlite3.open(":memory:") + + :ok = + Exqlite.Sqlite3.execute(db, """ + CREATE TABLE plans ( + id INTEGER NOT NULL, slug TEXT NOT NULL, price INTEGER, valid_at TEXT NOT NULL + ) + """) + + for %{up: up} <- AshSqlite.Temporal.Migration.statements(Plan) do + :ok = Exqlite.Sqlite3.execute(db, up) + end + + {:ok, statement} = + Exqlite.Sqlite3.prepare(db, """ + EXPLAIN QUERY PLAN + SELECT 1 FROM plans AS other + WHERE other.slug IS 'pro' + AND (json_extract(other.valid_at,'$.upper') IS NULL + OR json_extract(other.valid_at,'$.upper') > '2026-01-01') + """) + + {:ok, [[_, _, _, plan]]} = Exqlite.Sqlite3.fetch_all(db, statement) + + assert plan =~ "USING INDEX plans_valid_at_pit_slug" + refute plan =~ "SCAN other" + end + end + + describe "a relationship to a temporal resource" do + alias AshSqlite.Test.Enrollment + alias AshSqlite.Test.Plan + + setup do + for table <- ["enrollments", "plans"] do + {:ok, _} = AshSqlite.TransactionTestRepo.query("DELETE FROM #{table}", []) + end + + :ok + end + + defp enrol!(id, plan_id) do + Enrollment + |> Ash.Changeset.for_create(:create, %{id: id, plan_id: plan_id}) + |> Ash.create!() + end + + test "the destination gets no database foreign key" do + # A temporal table has no unique key to reference. Declaring one anyway creates + # the table and then fails every insert with "foreign key mismatch". + refute Enum.any?( + AshSqlite.TestRepo.query!("PRAGMA foreign_key_list(enrollments)", []).rows + ) + end + + test "a child can be written at all, which a foreign key would have prevented" do + Plan + |> Ash.Changeset.for_create(:create, %{id: 1, slug: "pro"}) + |> Ash.Changeset.as_of(@jan) + |> Ash.create!() + + assert %{plan_id: 1} = enrol!(10, 1) + end + + test "loading the relationship resolves the plan at the read's instant" do + Plan + |> Ash.Changeset.for_create(:create, %{id: 1, slug: "pro"}) + |> Ash.Changeset.as_of(@jan) + |> Ash.create!() + + Plan + |> Ash.get!(1) + |> Ash.Changeset.for_update(:change_price, %{price: 50}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + enrol!(10, 1) + + assert %{plan: %{price: 0}} = + Enrollment |> Ash.get!(10, as_of: @feb) |> Ash.load!(:plan, as_of: @feb) + + assert %{plan: %{price: 50}} = + Enrollment |> Ash.get!(10, as_of: @apr) |> Ash.load!(:plan, as_of: @apr) + end + + test "as_of resolves the destination even when the parent's period spans two versions" do + # This is what `temporal_keys` would otherwise be reached for. The overlap filter + # it bakes in matches *both* plan versions here, because both overlap the + # enrolment's `[Jan, oo)`. The `as_of` pin is what picks one, and it picks the + # right one, so a point-in-time read needs no overlap filter to be correct. + Plan + |> Ash.Changeset.for_create(:create, %{id: 1, slug: "pro"}) + |> Ash.Changeset.as_of(@jan) + |> Ash.create!() + + Plan + |> Ash.get!(1) + |> Ash.Changeset.for_update(:change_price, %{price: 50}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + enrol!(10, 1) + + assert %{plan: %{price: 0, valid_at: %{lower: lower_before}}} = + Enrollment |> Ash.get!(10, as_of: @feb) |> Ash.load!(:plan, as_of: @feb) + + assert %{plan: %{price: 50, valid_at: %{lower: lower_after}}} = + Enrollment |> Ash.get!(10, as_of: @apr) |> Ash.load!(:plan, as_of: @apr) + + assert lower_before == @jan + assert lower_after == @mar + end + + test "filtering across the relationship respects the read's instant" do + Plan + |> Ash.Changeset.for_create(:create, %{id: 1, slug: "pro"}) + |> Ash.Changeset.as_of(@jan) + |> Ash.create!() + + Plan + |> Ash.get!(1) + |> Ash.Changeset.for_update(:change_price, %{price: 50}) + |> Ash.Changeset.as_of(@mar) + |> Ash.update!() + + enrol!(10, 1) + + assert [%{id: 10}] = + Enrollment + |> Ash.Query.as_of(@apr) + |> Ash.Query.filter(plan.price == 50) + |> Ash.read!() + + assert [] = + Enrollment + |> Ash.Query.as_of(@feb) + |> Ash.Query.filter(plan.price == 50) + |> Ash.read!() + end + end + end +end