{
    "posts": {
        "created": [
            {
                "id": "d1633195-156f-4f9d-9ccf-7740203b080e",
                "_status": "created",
                "_changed": "",
                "title": "Phoenix",
                "content": "Phoenix is a web framework for Elixir",
                "likes": 200,
                "created_at": 1588400731806,
                "updated_at": 1588400731806
            }
        ],
        "updated": [
            {
                "id": "2d7c6a82-eb04-47b1-be52-6f8f6cf806ff",
                "_status": "updated",
                "_changed": "updated_at,title,content,likes",
                "title": "Elixir",
                "content": "Elixir is amazing",
                "likes": 100,
                "created_at": 1588389279195,
                "updated_at": 1588400691047
            }
        ],
        "deleted": [
            "2b130e52-079d-4b31-9f42-ce257cf546f0"
        ]
    },
    "comments": {
        "created": [],
        "updated": [
            {
                "id": "1e945c88-baf2-4db7-aa39-286b6865b3fb",
                "comment": "That's good!"
            }
        ],
        "deleted": []
    }
}

$ mix archive.uninstall phx_new
$ mix archive.install hex phx_new 1.5.1

$ mix phx.new blog_app

$ docker run --name blog-db -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=blog_app_dev -d -p 5432:5432 postgres:12.2

$ cd blog_app
$ mix ecto.create

$ mix ecto.gen.migration create_version_seq

# priv/repo/migrations/xxx_create_version_seq.exs
defmodule BlogApp.Repo.Migrations.CreateVersionSeq do
  use Ecto.Migration

  def change do
    execute "CREATE SEQUENCE version_seq"
  end
end

$ mix ecto.migrate

$ mix phx.gen.schema Blog.Post posts title:string content:string likes:integer push_id:integer created_at:utc_datetime_usec updated_at:utc_datetime_usec created_at_server:utc_datetime_usec updated_at_server:utc_datetime_usec deleted_at_server:utc_datetime_usec version:integer version_created:integer --binary-id

# priv/repo/migrations/xxx_create_posts.exs
defmodule BlogApp.Repo.Migrations.CreatePosts do
  use Ecto.Migration

  def change do
    create table(:posts, primary_key: false) do
      add :id, :binary_id, primary_key: true
      add :title, :string
      add :content, :string
      add :likes, :integer
      add :created_at, :utc_datetime_usec
      add :updated_at, :utc_datetime_usec
      add :created_at_server, :utc_datetime_usec
      add :updated_at_server, :utc_datetime_usec
      add :deleted_at_server, :utc_datetime_usec
      add :push_id, :integer
-     add :version, :integer
+     add :version, :bigint, default: fragment("nextval('version_seq')")
-     add :version_created, :integer
+     add :version_created, :bigint, default: fragment("nextval('version_seq')")

-     timestamps()
    end
  end
end

# lib/blog_app/blog/post.ex
defmodule BlogApp.Blog.Post do
  use Ecto.Schema
  import Ecto.Changeset

  @primary_key {:id, :binary_id, autogenerate: true}
  @foreign_key_type :binary_id
+ @derive {Jason.Encoder, only: [:id, :title, :content, :likes]}
  schema "posts" do
    field :title, :string
    field :content, :string
    field :likes, :integer
    field :created_at, :utc_datetime_usec
    field :updated_at, :utc_datetime_usec
    field :created_at_server, :utc_datetime_usec
    field :updated_at_server, :utc_datetime_usec
    field :deleted_at_server, :utc_datetime_usec
    field :push_id, :integer
    field :version, :integer
    field :version_created, :integer

-   timestamps()
  end
  # ...
end

$ mix ecto.migrate

# lib/blog_app_web/router.ex
defmodule BlogAppWeb.Router do
    # ...
    scope "/api", BlogAppWeb do
      pipe_through :api

      post "/sync/push", SyncController, :push
      get "/sync/pull", SyncController, :pull
    end
    # ...
end

# lib/blog_app_web/controllers/sync_controller.ex
defmodule BlogAppWeb.SyncController do
  use BlogAppWeb, :controller
  alias BlogApp.Sync

  def push(
        %Plug.Conn{
          body_params: req_body,
          query_params: %{"lastPulledVersion" => last_pulled_version}
        } = conn,
        _params
      ) do

    resp =  Sync.push(req_body, String.to_integer(last_pulled_version))
    json(conn, resp)
  end

  def pull(
        %Plug.Conn{
          query_params: %{"lastPulledVersion" => last_pulled_version}
        } = conn,
        _params
      ) do

    resp = Sync.pull(String.to_integer(last_pulled_version))
    json(conn, resp)
  end
end

# lib/blog_app/sync.ex
defmodule BlogApp.Sync do
  alias BlogApp.{Repo, Blog}

  def push(changes, last_pulled_version) do
    push_id = Enum.random(1..1_000_000_000)

    Ecto.Multi.new()
    |> Blog.record_posts(changes["posts"], last_pulled_version, push_id)
    |> Repo.transaction()

    pull(last_pulled_version, push_id)
  end
  # ...
end

# lib/blog_app/blog.ex
defmodule BlogApp.Blog do
  import Ecto.Query
  alias Ecto.Multi
  alias BlogApp.Repo
  alias BlogApp.Blog.Post

  def record_posts(%Multi{} = multi, post_changes, last_pulled_version, push_id) do
    multi
    |> Multi.run(:check_conflict_posts, fn _, _changes ->
      case check_conflict_version_posts(post_changes, last_pulled_version) do
        :no_conflict -> {:ok, :no_conflict}
        :conflict -> {:error, :conflict}
      end
    end)
    |> record_created_posts(post_changes["created"] |> set_push_id(push_id))
    |> record_updated_posts(post_changes["updated"] |> set_push_id(push_id))
    |> record_deleted_posts(post_changes["deleted"], push_id)
  end
  # ...
  defp set_push_id(posts, push_id) do
    posts
    |> Enum.map(fn post -> post |> Map.put("push_id", push_id) end)
  end
  # ...
end

# lib/blog_app/blog.ex
defmodule BlogApp.Blog do
  # ...
  def check_conflict_version_posts(post_changes, last_pulled_version) do
    ids =
      Enum.concat(post_changes["created"], post_changes["updated"])
      |> Enum.map(fn post -> post["id"] end)
      |> Enum.concat(post_changes["deleted"])

    count =
      Post
      |> select([p], count(p.version))
      |> where([p], p.id in ^ids)
      |> where([p], p.version > ^last_pulled_version or p.version_created > ^last_pulled_version)
      |> Repo.one()

    case count do
      0 -> :no_conflict
      _ -> :conflict
    end
  end
  # ...
end

# lib/blog_app/blog.ex
defmodule BlogApp.Blog do
  # ...
  def record_created_posts(%Multi{} = multi, created_changes),
    do: upsert_posts(multi, :create_posts, created_changes)

  def record_updated_posts(%Multi{} = multi, updated_changes),
    do: upsert_posts(multi, :update_posts, updated_changes)

  def upsert_posts(%Multi{} = multi, _name, changes) when is_nil(changes),
    do: multi

  def upsert_posts(%Multi{} = multi, name, changes) do
    now = DateTime.utc_now()

    posts =
      changes
      |> Enum.map(fn row ->
        row
        |> Map.put("created_at", row["created_at"] * 1000 |> DateTime.from_unix!(:microsecond))
        |> Map.put("updated_at", row["updated_at"] * 1000 |> DateTime.from_unix!(:microsecond))
        |> Map.put("created_at_server", now)
        |> Map.put("updated_at_server", now)
        |> Map.take(["id", "title", "content", "likes", "created_at", "updated_at", "created_at_server", "updated_at_server", "push_id"])
        |> key_to_atom()
      end)

    Multi.insert_all(multi, name, Post, posts,
      conflict_target: :id,
      on_conflict: {:replace_all_except, [:id, :version_created, :created_at_server, :deleted_at_server]},
      returning: true
    )
  end

  def key_to_atom(map) do
    Enum.reduce(map, %{}, fn
      {key, value}, acc when is_atom(key) -> Map.put(acc, key, value)
      {key, value}, acc when is_binary(key) -> Map.put(acc, String.to_existing_atom(key), value)
    end)
  end
  # ...
end

# lib/blog_app/blog.ex
defmodule BlogApp.Blog do
  # ...
  def record_deleted_posts(%Multi{} = multi, deleted_ids, _push_id) when is_nil(deleted_ids),
    do: multi

  def record_deleted_posts(%Multi{} = multi, deleted_ids, push_id) do
    now = DateTime.utc_now()

    posts =
      deleted_ids
      |> Enum.map(fn id ->
        %{id: id, deleted_at_server: now, push_id: push_id}
      end)

    Multi.insert_all(multi, :delete_posts, Post, posts,
      conflict_target: :id,
      on_conflict: {:replace, [:deleted_at_server, :version, :push_id]},
      returning: true
    )
  end
  # ...
end

# lib/blog_app/sync.ex
defmodule BlogApp.Sync do
  # ...
  def pull(last_pulled_version, push_id \\ nil) do
    %{latest_version: latest_version_posts, changes: posts_changes} =
      Blog.list_posts_changes(last_pulled_version, push_id)

    latest_version =
      [last_pulled_version, latest_version_posts]
      |> Enum.max()

    %{
      "latestVersion" => latest_version,
      "changes" => %{
        "posts" => posts_changes
      }
    }
  end
end

# lib/blog_app/blog.ex
defmodule BlogApp.Blog do
  # ...
  def list_posts_changes(last_pulled_version, push_id) do
    posts_latest =
      Post
      |> where([p], p.version_created > ^last_pulled_version or p.version > ^last_pulled_version)
      |> Repo.all()

    posts_changes =
      posts_latest
      |> Enum.reject(fn post -> is_just_pushed(post, push_id) end)
      |> Enum.group_by(fn post ->
        cond do
          post.version_created > last_pulled_version and is_nil(post.deleted_at_server) -> :created
          post.created_at_server != post.updated_at_server and is_nil(post.deleted_at_server) -> :updated
          not is_nil(post.deleted_at_server) -> :deleted
        end
      end)
      |> Map.update(:created, [], fn posts -> posts end)
      |> Map.update(:updated, [], fn posts -> posts end)
      |> Map.update(:deleted, [], fn posts -> posts |> Enum.map(fn post -> post.id end) end)

    latest_version = find_latest_version(posts_latest)

    %{latest_version: latest_version, changes: posts_changes}
  end

  defp find_latest_version(posts) do
    posts
    |> Enum.flat_map(fn post -> [post.version, post.version_created] end)
    |> Enum.max(fn -> 0 end)
  end
  # ...
  defp is_just_pushed(_post, push_id) when is_nil(push_id), do: false
  defp is_just_pushed(post, push_id), do: post.push_id == push_id
end

$ mix phx.server