defmodule Statestores.Adapters.PostgresSnapshotAdapter do @moduledoc """ Implements the behavior defined in `Statestores.Adapters.SnapshotBehaviour` for Postgres databases. """ use Statestores.Adapters.SnapshotBehaviour use Ecto.Repo, otp_app: :spawn_statestores, adapter: Ecto.Adapters.Postgres alias Statestores.Schemas.{Snapshot, HistoricalSnapshot} def get_by_key(id), do: get_by(Snapshot, id: id) def get_by_key_and_revision(id, revision) do query = """ SELECT * FROM historical_snapshots WHERE actor_id = #{id} AND revision = #{revision} ORDER BY inserted_at DESC, updated_at DESC """ %Postgrex.Result{rows: rows} = Ecto.Adapters.SQL.query!(Statestores.Adapters.PostgresSnapshotAdapter, query) List.first(rows) |> to_snapshot(:historical) |> case do nil -> get_by_key(id) response -> response end end def get_all_snapshots_by_key(id) do query = "SELECT * FROM historical_snapshots WHERE actor_id = #{id}" %Postgrex.Result{rows: rows} = Ecto.Adapters.SQL.query!(Statestores.Adapters.PostgresSnapshotAdapter, query) rows |> Enum.map(&to_snapshot(&1, :historical)) end def get_snapshots_by_interval(id, time_start, time_end) do query = """ SELECT * FROM historical_snapshots WHERE actor_id = #{id} AND valid_from <= '#{time_end}' AND valid_to >= '#{time_start}' ORDER BY inserted_at ASC, updated_at ASC """ %Postgrex.Result{rows: rows} = Ecto.Adapters.SQL.query!(Statestores.Adapters.PostgresSnapshotAdapter, query) rows |> Enum.map(&to_snapshot(:historical, &1)) end def save(%Snapshot{} = actual_snapshot) do %Snapshot{} |> Snapshot.changeset(%{ id: actual_snapshot.id, actor: actual_snapshot.actor, system: actual_snapshot.system, status: actual_snapshot.status, node: actual_snapshot.node, revision: actual_snapshot.revision || 0, tags: actual_snapshot.tags, data_type: actual_snapshot.data_type, data: actual_snapshot.data }) |> insert_or_update( on_conflict: [ set: [ actor: actual_snapshot.actor, system: actual_snapshot.system, status: actual_snapshot.status, node: actual_snapshot.node, revision: (actual_snapshot.revision || 0) + 1, tags: actual_snapshot.tags, data_type: actual_snapshot.data_type, data: actual_snapshot.data, updated_at: DateTime.utc_now() ] ], conflict_target: [:id] ) end def default_port, do: "5432" defp to_snapshot(row, :current) do data = Statestores.Vault.decrypt!(Enum.at(row, 8)) %Snapshot{ id: Enum.at(row, 0), actor: Enum.at(row, 1), system: Enum.at(row, 2), status: Enum.at(row, 3), node: Enum.at(row, 4), revision: Enum.at(row, 5), tags: Enum.at(row, 6), data_type: Enum.at(row, 7), data: data, inserted_at: Enum.at(row, 9), updated_at: Enum.at(row, 10) } end defp to_snapshot(row, :historical) do data = Statestores.Vault.decrypt!(Enum.at(row, 9)) %Snapshot{ id: Enum.at(row, 1), actor: Enum.at(row, 2), system: Enum.at(row, 3), status: Enum.at(row, 4), node: Enum.at(row, 5), revision: Enum.at(row, 6), tags: Enum.at(row, 7), data_type: Enum.at(row, 8), data: data, inserted_at: Enum.at(row, 12), updated_at: Enum.at(row, 13) } end end