diff --git a/lib/neon_vertigo_service/message_processor.ex b/lib/neon_vertigo_service/message_processor.ex index 929d4f6..687c2bd 100644 --- a/lib/neon_vertigo_service/message_processor.ex +++ b/lib/neon_vertigo_service/message_processor.ex @@ -1,6 +1,8 @@ defmodule NeonVertigoService.MessageProcessor do alias NeonVertigoService.Repo + import Ecto.Query, only: [from: 2] + def process_message(mqtt_pid, {"auctions", "create"}, payload, properties) do changeset = Auction.changeset(%Auction{}, payload) reply = case Repo.insert(changeset) do @@ -13,6 +15,28 @@ defmodule NeonVertigoService.MessageProcessor do reply(reply, mqtt_pid, properties) end + def process_message(mqtt_pid, {"auctions", auction_gid, "bid"}, payload, properties) do + auction = Repo.get(Auction, auction_gid) + if is_nil(auction) do + {:not_found, auction_gid} + else + changeset = Bid.changeset(%Bid{}, Map.put(payload, "auction_gid", auction.gid)) + reply = case Repo.insert(changeset) do + {:ok, bid} -> + bid_data = %{ + bid: bid.bid, + bidder_gid: bid.bidder_gid, + timestamp: bid.bid_timestamp + } + send_mqtt(bid_data, mqtt_pid, "notify/new_bid/#{auction.gid}") + %{ok: true} + {:error, changeset} -> %{ok: false} + end + + reply(reply, mqtt_pid, properties) + end + end + def process_message(_, _, _, _) do {:ok, :unknown_message} end diff --git a/lib/neon_vertigo_service/mqtt/server.ex b/lib/neon_vertigo_service/mqtt/server.ex index 85d4238..ef19fa1 100644 --- a/lib/neon_vertigo_service/mqtt/server.ex +++ b/lib/neon_vertigo_service/mqtt/server.ex @@ -25,7 +25,7 @@ defmodule NeonVertigoService.Mqtt.Server do ) {:ok, _mqtt_props} = :emqtt.connect(mqtt_pid) - :emqtt.subscribe(mqtt_pid, %{}, [{"$share/neon-vertigo-service/auctions/+", [{:qos, 1}]}]) + :emqtt.subscribe(mqtt_pid, %{}, [{"$share/neon-vertigo-service/auctions/#", [{:qos, 1}]}]) {:noreply, %{state | mqtt_pid: mqtt_pid}} end diff --git a/lib/schemas/auction.ex b/lib/schemas/auction.ex index c07e19b..787093f 100644 --- a/lib/schemas/auction.ex +++ b/lib/schemas/auction.ex @@ -2,6 +2,7 @@ defmodule Auction do use Ecto.Schema import Ecto.Changeset + @primary_key false schema "auctions" do field :gid, :string, primary_key: true field :author_gid, :string diff --git a/lib/schemas/bid.ex b/lib/schemas/bid.ex new file mode 100644 index 0000000..fd5553e --- /dev/null +++ b/lib/schemas/bid.ex @@ -0,0 +1,52 @@ +defmodule Bid do + use Ecto.Schema + import Ecto.Changeset + alias NeonVertigoService.Repo + import Ecto.Query, only: [from: 2] + + @primary_key false + schema "bids" do + belongs_to :auction, Auction, foreign_key: :auction_gid, references: :gid, type: :string, primary_key: true + field :bid_timestamp, :utc_datetime_usec, primary_key: true + field :bidder_gid, :string, primary_key: true + field :bid, :decimal + end + + def changeset(bid, params \\ %{}) do + bid + |> cast(params, [:auction_gid, :bidder_gid, :bid]) + |> validate_required([:auction_gid, :bidder_gid, :bid]) + |> validate_number(:bid, greater_than_or_equal_to: 0) + |> validate_bid_value_max + |> validate_bid_value_more_than_auction_starting_price + |> put_change(:bid_timestamp, DateTime.utc_now) + |> foreign_key_constraint(:auction_gid) + end + + defp validate_bid_value_max(changeset) do + auction_gid = get_field(changeset, :auction_gid) + new_bid = get_field(changeset, :bid) + existing_bid = Repo.one( + from b in Bid, where: b.auction_gid == ^auction_gid, + order_by: [desc: :bid_timestamp], + limit: 1, + select: b.bid + ) + + if is_nil(existing_bid) or Decimal.compare(existing_bid, new_bid) == :lt do + changeset + else + add_error(changeset, :bid, "smaller than last bid") + end + end + + defp validate_bid_value_more_than_auction_starting_price(changeset) do + auction_gid = get_field(changeset, :auction_gid) + auction = Repo.one!(from a in Auction, where: a.gid == ^auction_gid, select: a) + if Decimal.compare(auction.starting_price, get_field(changeset, :bid)) == :lt do + changeset + else + add_error(changeset, :bid, "smaller than auction's starting price") + end + end +end diff --git a/priv/repo/migrations/20260716122932_add_bids_table.exs b/priv/repo/migrations/20260716122932_add_bids_table.exs new file mode 100644 index 0000000..6116449 --- /dev/null +++ b/priv/repo/migrations/20260716122932_add_bids_table.exs @@ -0,0 +1,12 @@ +defmodule NeonVertigoService.Repo.Migrations.AddBidsTable do + use Ecto.Migration + + def change do + create table("bids", primary_key: false) do + add :auction_gid, :string, primary_key: true + add :bid_timestamp, :timestamp, cluster_key: true + add :bidder_gid, :string, cluster_key: true + add :bid, :decimal, null: false + end + end +end