Bids schema and creating

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

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

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

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