diff --git a/compose.yml b/compose.yml index df2a7fb..dd81f7e 100644 --- a/compose.yml +++ b/compose.yml @@ -1,35 +1,13 @@ +name: neon-vertigo-service + services: - srv: - build: - context: . - target: dev - develop: - watch: - - path: ./ - include: "mix.*" - target: "/app" - action: sync+restart - initial_sync: true + db: + image: "cassandra:5" + ports: + - "7000:7000" + - "9042:9042" volumes: - - "app_build:/app/_build" - - "app_deps:/app_deps" - - "./config:/app/config" - - "./lib:/app/lib" - - "./test:/app/test" - environment: - MQTT_HOST: activemq-artemis - MQTT_USERNAME: "${MQTT_USERNAME}" - MQTT_PASSWORD: "${MQTT_PASSWORD}" - networks: - - default - - nv - - -networks: - nv: - name: neon-vertigo - external: true + - cassandra:/var/lib/cassandra volumes: - app_build: - app_deps: + cassandra: \ No newline at end of file diff --git a/config/config.exs b/config/config.exs index e69de29..9d3da89 100644 --- a/config/config.exs +++ b/config/config.exs @@ -0,0 +1,5 @@ +import Config + +config :neon_vertigo_service, ecto_repos: [NeonVertigoService.Repo] + +import_config "#{Mix.env()}.exs" \ No newline at end of file diff --git a/config/dev.exs b/config/dev.exs new file mode 100644 index 0000000..0d47fc6 --- /dev/null +++ b/config/dev.exs @@ -0,0 +1,12 @@ +import Config + +config :neon_vertigo_service, NeonVertigoService.Repo, [ + migration_primary_key: [name: :id, type: :uuid], # Overrides the default type `bigserial` used for version attribute in schema migration + nodes: ["127.0.0.1"], # List of database connection endpoints + keyspace: "nvservice", # Name of your keyspace + sync_connect: 5000, # Waiting time in milliseconds for the database connection + log: :info, + stacktrace: true, + show_sensitive_data_on_connection_error: true, + pool_size: 10 +] \ No newline at end of file diff --git a/lib/neon_vertigo_service.ex b/lib/neon_vertigo_service.ex index c1c9a3f..02ae5ec 100644 --- a/lib/neon_vertigo_service.ex +++ b/lib/neon_vertigo_service.ex @@ -8,6 +8,7 @@ defmodule NeonVertigoService do @impl true def start(_start_type, _start_args) do children = [ + NeonVertigoService.Repo, NeonVertigoService.Mqtt.Supervisor, ] diff --git a/lib/neon_vertigo_service/message_processor.ex b/lib/neon_vertigo_service/message_processor.ex index b600578..83a739e 100644 --- a/lib/neon_vertigo_service/message_processor.ex +++ b/lib/neon_vertigo_service/message_processor.ex @@ -1,20 +1,14 @@ defmodule NeonVertigoService.MessageProcessor do - @ets_auctions NeonVertigoService.Auctions - @ets_bids NeonVertigoService.Bids - - def init_tables() do - @ets_auctions = :ets.new(@ets_auctions, [:named_table]) - @ets_bids = :ets.new(@ets_bids, [:named_table]) - :ok - end - def process_message(["auctions", "create"], payload, _properties) do - gid = payload["gid"] - data = Map.take(payload, ["author_gid", "min_bet"]) |> Enum.map(fn {k, v} -> {String.to_atom(k), v} end) |> Map.new - {min_bet, _} = Decimal.parse(data[:min_bet]) - data = Map.put(data, :min_bet, min_bet) - :ets.insert_new(@ets_auctions, {gid, data}) - :noreply + changeset = Auction.changeset(%Auction{}, payload) + reply = case NeonVertigoService.Repo.insert(changeset) do + {:ok, auction} -> + %{ok: true, gid: auction.gid} + {:error, changeset} -> + %{ok: false, gid: changeset |> Ecto.Changeset.get_field(:gid)} + end + + {:reply, reply} end def process_message(_, _, _) do diff --git a/lib/neon_vertigo_service/mqtt/server.ex b/lib/neon_vertigo_service/mqtt/server.ex index ea3224c..b119167 100644 --- a/lib/neon_vertigo_service/mqtt/server.ex +++ b/lib/neon_vertigo_service/mqtt/server.ex @@ -11,8 +11,6 @@ defmodule NeonVertigoService.Mqtt.Server do # Callbacks def handle_continue(:post_init, {}) do - :ok = NeonVertigoService.MessageProcessor.init_tables() - mqtt_config = Application.fetch_env!(:neon_vertigo_service, :mqtt_params) {:ok, mqtt_pid} = Supervisor.start_child( NeonVertigoService.Mqtt.Supervisor, @@ -34,10 +32,14 @@ defmodule NeonVertigoService.Mqtt.Server do IO.inspect(msg) %{payload: payload_str, topic: topic_str, properties: properties} = msg topic = String.split(topic_str, "/") + response_topic = properties[:"Response-Topic"] payload = JSON.decode!(payload_str) case NeonVertigoService.MessageProcessor.process_message(topic, payload, properties) do - {:reply, topic, data} -> :emqtt.publish(mqtt_pid, topic, data) + {:reply, data} -> + if is_binary(response_topic) && String.length(response_topic) > 0 do + :emqtt.publish(mqtt_pid, response_topic, JSON.encode!(data), retain: true) + end :noreply -> nil end diff --git a/lib/neon_vertigo_service/repo.ex b/lib/neon_vertigo_service/repo.ex new file mode 100644 index 0000000..f214897 --- /dev/null +++ b/lib/neon_vertigo_service/repo.ex @@ -0,0 +1,3 @@ +defmodule NeonVertigoService.Repo do + use Ecto.Repo, otp_app: :neon_vertigo_service, adapter: Exandra +end diff --git a/lib/schemas/auction.ex b/lib/schemas/auction.ex new file mode 100644 index 0000000..c07e19b --- /dev/null +++ b/lib/schemas/auction.ex @@ -0,0 +1,18 @@ +defmodule Auction do + use Ecto.Schema + import Ecto.Changeset + + schema "auctions" do + field :gid, :string, primary_key: true + field :author_gid, :string + field :starting_price, :decimal + end + + def changeset(auction, params \\ %{}) do + auction + |> cast(params, [:gid, :author_gid, :starting_price]) + |> validate_required([:gid, :author_gid, :starting_price]) + |> validate_number(:starting_price, greater_than_or_equal_to: 0) + |> unique_constraint([:gid]) + end +end diff --git a/mix.exs b/mix.exs index 07055e8..2c083f8 100644 --- a/mix.exs +++ b/mix.exs @@ -24,6 +24,8 @@ defmodule NeonVertigoService.MixProject do [ {:emqtt, "~> 1.14"}, {:decimal, "~> 2.3"}, + {:ecto, "~> 3.10"}, + {:exandra, "~> 1.1"}, # {:dep_from_hexpm, "~> 0.3.0"}, # {:dep_from_git, git: "https://github.com/elixir-lang/my_dep.git", tag: "0.1.0"} ] diff --git a/mix.lock b/mix.lock index f039df3..2ca23ef 100644 --- a/mix.lock +++ b/mix.lock @@ -1,9 +1,17 @@ %{ "cowlib": {:hex, :cowlib, "2.13.0", "db8f7505d8332d98ef50a3ef34b34c1afddec7506e4ee4dd4a3a266285d282ca", [:make, :rebar3], [], "hexpm", "e1e1284dc3fc030a64b1ad0d8382ae7e99da46c3246b815318a4b848873800a4"}, + "db_connection": {:hex, :db_connection, "2.10.2", "ae391e803a5adff104da913c2fc1c0c14a37f8b10001dcef568796e1fb7bf95c", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "510b14482330f1af6490a2fa0efd8d4f1435d1529b165647df22ac0f2df0fa93"}, "decimal": {:hex, :decimal, "2.3.0", "3ad6255aa77b4a3c4f818171b12d237500e63525c2fd056699967a3e7ea20f62", [:mix], [], "hexpm", "a4d66355cb29cb47c3cf30e71329e58361cfcb37c34235ef3bf1d7bf3773aeac"}, + "ecto": {:hex, :ecto, "3.13.6", "352135b474f91d1ab99a1b502171d207e9db60421c9e3d0ecab4c7ab96b24d14", [:mix], [{:decimal, "~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "8afa059bc16cd2c94739ec0a11e3e5df69d828125119109bef35f20a21a76af2"}, + "ecto_sql": {:hex, :ecto_sql, "3.13.5", "2f8282b2ad97bf0f0d3217ea0a6fff320ead9e2f8770f810141189d182dc304e", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "aa36751f4e6a2b56ae79efb0e088042e010ff4935fc8684e74c23b1f49e25fdc"}, "emqtt": {:hex, :emqtt, "1.14.7", "7ace5a62337fe92c8596bc2872cbcb3430959ff90718c2004bdd691bce61614e", [:rebar3], [{:cowlib, "2.13.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:getopt, "1.0.3", [hex: :getopt, repo: "hexpm", optional: false]}, {:gun, "2.1.0", [hex: :gun, repo: "hexpm", optional: false]}, {:quicer, "0.2.15", [hex: :quicer, repo: "hexpm", optional: false]}], "hexpm", "9eb060ce4abe5092064114819dc72955bc657e020d494660ecc593772c4125db"}, + "exandra": {:hex, :exandra, "1.1.0", "e0e73a4099867ec1fdfe551390176ea5941126fe88f4cbf939f6143420640b64", [:mix], [{:decimal, "~> 1.6 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:ecto, "~> 3.13", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, "~> 3.13", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:xandra, "~> 0.19.4", [hex: :xandra, repo: "hexpm", optional: false]}], "hexpm", "0d4e670f5bc85c429ee1c96cc1c3f85b9772b9ba52574657793560640928669e"}, "getopt": {:hex, :getopt, "1.0.3", "4f3320c1f6f26b2bec0f6c6446b943eb927a1e6428ea279a1c6c534906ee79f1", [:rebar3], [], "hexpm", "7e01de90ac540f21494ff72792b1e3162d399966ebbfc674b4ce52cb8f49324f"}, "gun": {:hex, :gun, "2.1.0", "b4e4cbbf3026d21981c447e9e7ca856766046eff693720ba43114d7f5de36e87", [:make, :rebar3], [{:cowlib, "2.13.0", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "52fc7fc246bfc3b00e01aea1c2854c70a366348574ab50c57dfe796d24a0101d"}, + "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"}, + "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"}, "quicer": {:hex, :quicer, "0.2.15", "775fcfa09c9ce5a4a00d23c1c8b1e81cd361ebc847e43bb766efac4373cdb1b8", [:rebar3], [{:snabbkaffe, "1.0.10", [hex: :snabbkaffe, repo: "hexpm", optional: false]}], "hexpm", "52236d9384a541341706bcae438e81b209fa21d2045d3df317f8ba5beda6a8e5"}, "snabbkaffe": {:hex, :snabbkaffe, "1.0.10", "9be2f54f61fc6862391b666b2b5b76c3fa53598e2989a17cef1b48cf347a8a63", [:rebar3], [], "hexpm", "70a98df36ae756908d55b5770891d443d63c903833e3e87d544036e13d4fac26"}, + "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"}, + "xandra": {:hex, :xandra, "0.19.4", "bc6e46db726ce252a21b1bb5a7a21502b61702242dccf0d1007201210064e08c", [:mix], [{:decimal, "~> 1.7 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}, {:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "3584949976f1146d3bef0f4f91a0fb2ce025f24be181827172ba0bc2571cb362"}, } diff --git a/priv/repo/migrations/20260715174940_add_auctions_table.exs b/priv/repo/migrations/20260715174940_add_auctions_table.exs new file mode 100644 index 0000000..1f1536d --- /dev/null +++ b/priv/repo/migrations/20260715174940_add_auctions_table.exs @@ -0,0 +1,11 @@ +defmodule NeonVertigoService.Repo.Migrations.AddAuctionsTable do + use Ecto.Migration + + def change do + create table("auctions", primary_key: false) do + add :gid, :string, primary_key: true + add :author_gid, :string, null: false + add :starting_price, :decimal, null: false + end + end +end