From a0a9bf30584969298b0c0cd523bbc3c18209578a Mon Sep 17 00:00:00 2001 From: Conor Sinclair Date: Sun, 20 Sep 2026 18:44:04 +0200 Subject: [PATCH 1/3] chore: allow building against ash's unreleased temporal branches Temporal lives on the `temporal` branches of ash and ash_sql and is in no released version, so ASH_VERSION and ASH_SQL_VERSION gain a "temporal" case alongside the existing "main" and "local" ones. ASH_VERSION=temporal ASH_SQL_VERSION=temporal mix test Without it the feature compiles out entirely and the suite runs as before. --- mix.exs | 6 ++++++ 1 file changed, 6 insertions(+) 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}" From 9e5bad6e3e0fa87c56ae8b955f2284ee5ef40d68 Mon Sep 17 00:00:00 2001 From: Conor Sinclair Date: Sun, 20 Sep 2026 18:43:25 +0200 Subject: [PATCH 2/3] feat: store and query an Ash.Type.Range on SQLite SQLite has no range type, so a range attribute is stored as a single JSON text column and every range predicate compiles to comparisons on json_extract of its keys. AshSqlite.Type.RangeBound encodes both sides of every comparison at one fixed ISO8601 precision. The comparison a range predicate performs is the inner type's stored comparison rather than the adapter's, and the two do not agree: DateTime.to_iso8601/1 keeps whatever precision the value carries, so one instant has two spellings that compare unequal and in the wrong direction. An absent bound is JSON null, so an unbounded side is tested with IS NULL rather than against a sentinel of the inner type. AshSqlite.SqlImplementation renders range_overlaps, range_contains over a range and over a point, range_adjacent, range_lower and range_upper. attribute_ecto_type/2 covers the write path, which the expression seam does not reach. The migration generator maps Ash.Type.Range to :text. Each test asserts agreement with the runtime Ash.Range function it mirrors. --- lib/data_layer.ex | 9 + .../migration_generator.ex | 2 + lib/sql_implementation.ex | 360 ++++++++++++++++++ lib/type/range.ex | 115 ++++++ lib/type/range_bound.ex | 59 +++ test/range_test.exs | 250 ++++++++++++ 6 files changed, 795 insertions(+) create mode 100644 lib/type/range.ex create mode 100644 lib/type/range_bound.ex create mode 100644 test/range_test.exs diff --git a/lib/data_layer.ex b/lib/data_layer.ex index ec44534..c59eee0 100644 --- a/lib/data_layer.ex +++ b/lib/data_layer.ex @@ -440,6 +440,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 diff --git a/lib/migration_generator/migration_generator.ex b/lib/migration_generator/migration_generator.ex index 067a445..6e55e56 100644 --- a/lib/migration_generator/migration_generator.ex +++ b/lib/migration_generator/migration_generator.ex @@ -2541,6 +2541,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/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/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 From 55262c91675d6f954328533d4b53e20814699273 Mon Sep 17 00:00:00 2001 From: Conor Sinclair Date: Sun, 20 Sep 2026 18:43:45 +0200 Subject: [PATCH 3/3] feat: support temporal resources A temporal resource stores one row per validity period. Postgres splits a period with one statement, `UPDATE ... FOR PORTION OF valid_at FROM $as_of TO NULL`, over a `PRIMARY KEY (id, valid_at WITHOUT OVERLAPS)` exclusion constraint. SQLite has none of that, and no data-modifying CTE to build it out of, so AshSqlite.Temporal reaches the same outcome in three statements: close the version whose period contains as_of, copy that row forward under [as_of, prior_upper), then apply the caller's update to the copy. Step three is the ordinary non-temporal update path, which is why atomics, RETURNING and the changeset's filters need no temporal branch inside them. Closing before inserting means the two periods never overlap at any point between statements, so the non-overlap trigger sees no transient violation. The reverse order trips both the trigger and the partial unique index. The transaction is opened here rather than by Ash. Ash wraps an atomic update only when the resource has after-batch hooks or the data layer asks for prefer_transaction_for_atomic_updates?, and this one answers false because a non-temporal atomic update is a single statement. Without it a failure in the third statement leaves the history rewritten with no current version, and two concurrent writers interleave and lose a write. AshSqlite.Verifiers.VerifyTemporal refuses a temporal resource whose repo leaves write transactions off. The row identity on a temporal table is the primary key plus the period, so pkey_filter/2 carries the period and so does the join bulk_updatable_query/6 builds when a query needs one. The key alone names every version, and joining on it rewrites closed history. set_as_of/3 renders the containment test against the same json_extract expression the index is built on, so a keyed point-in-time read is an index seek rather than a scan. The upsert resolves its own match. SQLite will not use a partial unique index as an ON CONFLICT target and the only unique index on a temporal table is the partial one; Postgres cannot use ON CONFLICT against its exclusion constraint either. AshSqlite.Temporal.Migration generates what SQLite will not give from one declaration: an index on each key set plus the period's lower bound, a partial unique index refusing a second current version per key set, and a non-overlap trigger pair. Every identity gets the same treatment, because an identity on a temporal resource is unique per period rather than per table. The insert trigger deliberately omits the NEW.rowid exclusion the update trigger carries: an inserted row has no rowid yet, so the comparison is NULL and the trigger would silently never fire. The generator also stops emitting three things for a temporal resource. A PRIMARY KEY, because it makes a second version impossible and on an integer key makes the column SQLite's rowid alias. Identities as plain unique indexes, for the same reason. And a foreign key to a temporal destination: there is no unique key to point at, and SQLite creates the table anyway and then fails every insert into the child with "foreign key mismatch". --- lib/data_layer.ex | 328 +++++- .../migration_generator.ex | 75 +- lib/temporal.ex | 514 ++++++++ lib/temporal/migration.ex | 163 +++ lib/verifiers/verify_temporal.ex | 83 ++ ...20260919000000_add_subscriptions_table.exs | 63 + ...19010000_add_dated_subscriptions_table.exs | 33 + .../20260920000000_add_plans_table.exs | 67 ++ .../20260920010000_add_enrollments_table.exs | 22 + test/support/domain.ex | 8 + test/support/resources/dated_subscription.ex | 53 + test/support/resources/enrollment.ex | 50 + test/support/resources/plan.ex | 60 + test/support/resources/subscription.ex | 76 ++ test/temporal_test.exs | 1045 +++++++++++++++++ 15 files changed, 2625 insertions(+), 15 deletions(-) create mode 100644 lib/temporal.ex create mode 100644 lib/temporal/migration.ex create mode 100644 lib/verifiers/verify_temporal.ex create mode 100644 priv/test_repo/migrations/20260919000000_add_subscriptions_table.exs create mode 100644 priv/test_repo/migrations/20260919010000_add_dated_subscriptions_table.exs create mode 100644 priv/test_repo/migrations/20260920000000_add_plans_table.exs create mode 100644 priv/test_repo/migrations/20260920010000_add_enrollments_table.exs create mode 100644 test/support/resources/dated_subscription.ex create mode 100644 test/support/resources/enrollment.ex create mode 100644 test/support/resources/plan.ex create mode 100644 test/support/resources/subscription.ex create mode 100644 test/temporal_test.exs diff --git a/lib/data_layer.ex b/lib/data_layer.ex index c59eee0..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,9 +443,9 @@ 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 + # 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 @@ -458,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 @@ -637,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)) @@ -914,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 @@ -1354,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? = @@ -1502,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 = @@ -1547,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 = @@ -1634,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{} -> @@ -1759,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{} -> @@ -1939,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 6e55e56..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 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/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/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/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