implement data update.

lib/farmbot/bot_state/transport/gen_mqtt/client.ex

@@ -69,7 +69,27 @@ defmodule Farmbot.BotState.Transport.GenMQTT.Client do
 
   def on_publish(["bot", _bot, "sync"], msg, state) do
     sync_cmd = msg |> Poison.decode!()
-    Logger.warn("Not implemented yet: #{inspect sync_cmd}")
+    repo = Farmbot.Repo.other_repo()
+    mod = Module.concat(["Farmbot", "Repo", sync_cmd["kind"]])
+    if Code.ensure_loaded?(mod) do
+      Logger.warn "Updating #{sync_cmd["kind"]} => #{sync_cmd["body"]["id"]}"
+      obj = sync_cmd["body"] |> Poison.encode! |> Poison.decode!(as: struct(mod))
+
+      # require IEx; IEx.pry
+      # We need to check if this object exists in the database.
+      case repo.get(mod, obj.id) do
+        # If it does not, just return the newly created object.
+        nil -> obj
+
+        # if there is an existing record, copy the ecto  meta from the old
+        # record. This allows `insert_or_update` to work properly.
+        existing -> %{obj | __meta__: existing.__meta__}
+      end
+      |> mod.changeset()
+      |> repo.insert_or_update!()
+    else
+      Logger.warn "Unknown module: #{mod} #{inspect sync_cmd["body"]}"
+    end
     {:ok, state}
   end
 

lib/farmbot/http/http.ex

@@ -72,6 +72,12 @@ defmodule Farmbot.HTTP do
     request!(:post, url, body, headers, opts)
   end
 
+  def put(url, body, headers \\ [], opts \\ [])
+
+  def put(url, body, headers, opts) do
+    request(:put, url, body, headers, opts)
+  end
+
   @doc "Download a file to the filesystem."
   def download_file(url, path, progress_callback \\ nil, payload \\ "", headers \\ [])
 

lib/farmbot/repo/device.ex

@@ -0,0 +1,21 @@
+defmodule Farmbot.Repo.Device do
+  @moduledoc """
+  """
+
+  use Ecto.Schema
+  import Ecto.Changeset
+
+  schema "devices" do
+    field(:name, :string)
+  end
+
+  use Farmbot.Repo.Syncable
+  @required_fields [:id, :name]
+
+  def changeset(device, params \\ %{}) do
+    device
+    |> cast(params, @required_fields)
+    |> validate_required(@required_fields)
+    |> unique_constraint(:id)
+  end
+end

lib/farmbot/repo/repo.ex

@@ -3,28 +3,48 @@ defmodule Farmbot.Repo do
 
   use GenServer
 
-  def repo(repo_worker) do
-    GenServer.call(repo_worker, :repo)
+  def current_repo do
+    GenServer.call(__MODULE__, :current_repo)
   end
 
-  def flip(repo_worker) do
-    GenServer.call(repo_worker, :flip)
+  def other_repo do
+    GenServer.call(__MODULE__, :other_repo)
   end
 
-  def start_link(repos, opts \\ []) do
-    GenServer.start_link(__MODULE__, repos, opts)
+  def flip() do
+    GenServer.call(__MODULE__, :flip)
+  end
+
+  def register_sync_cmd(sync_cmd) do
+    GenServer.call(__MODULE__, {:register_sync_cmd, sync_cmd})
+  end
+
+  def start_link(repos) do
+    GenServer.start_link(__MODULE__, repos, [name: __MODULE__])
   end
 
   def init([_repo_a, _repo_b] = repos) do
-    {:ok, repos}
+    {:ok, %{repos: repos, sync_cmds: []}}
   end
 
-  def handle_call(:repo, _, [repo, _] = repos) do
-    {:reply, repo, repos}
+  def handle_call(:current_repo, _, %{repos: [repo_a, _]} = state) do
+    {:reply, repo_a, state}
   end
 
-  def handle_call(:flip, _, [repo_a, repo_b]) do
-    {:reply, repo_b, [repo_b, repo_a]}
+  def handle_call(:other_repo, _, %{repos: [_, repo_b]} = state) do
+    {:reply, repo_b, state}
+  end
+
+  def handle_call(:flip, _, %{repos: [repo_a, repo_b]} = state) do
+    Enum.reverse(state.sync_cmds) |> Enum.map(fn(sync_cmd) ->
+      mod = Module.concat(["Farmbot", "Repo", sync_cmd["kind"]])
+      mod.changeset(struct(mod), sync_cmd["body"]) |> repo_a.insert_or_update!()
+    end)
+    {:reply, repo_b, %{state | repos: [repo_b, repo_a]}}
+  end
+
+  def handle_call({:register_sync_cmd, sync_cmd}, _from, state) do
+    {:reply, :ok, %{state | sync_cmds: [sync_cmd | state.sync_cmds]}}
   end
 
   @doc false

lib/farmbot/repo/supervisor.ex

@@ -14,7 +14,7 @@ defmodule Farmbot.Repo.Supervisor do
     |> Enum.map(fn repo ->
          supervisor(repo, [])
        end)
-    |> Kernel.++([worker(Farmbot.Repo, [@repos, [name: Farmbot.Repo]])])
+    |> Kernel.++([worker(Farmbot.Repo, [@repos])])
     |> supervise(strategy: :one_for_one)
   end
 end

priv/config_storage/migrations/20171025231226_add_devices_table.exs

@@ -0,0 +1,7 @@
+defmodule Farmbot.System.ConfigStorage.Migrations.AddDevicesTable do
+  use Ecto.Migration
+
+  def change do
+
+  end
+end

priv/repo/migrations/20171025231225_add_devices_table.exs

@@ -0,0 +1,12 @@
+defmodule Farmbot.Repo.A.Migrations.AddDevicesTable do
+  use Ecto.Migration
+
+  def change do
+    create table("devices", primary_key: false) do
+      add(:id, :integer)
+      add(:name, :string)
+    end
+
+    create(unique_index("devices", [:id]))
+  end
+end