defmodule SApp.Chat do @moduledoc """ Shard application for a replicated chat room with full history. Chat rooms are globally identified by their channel name. A chat room manifest is of the form: %SApp.Chat.Manifest{channel: channel_name} Future improvements: - message signing - storage of the chatroom messages to disk - use a DHT to find peers that are interested in this channel - epidemic broadcast (carefull not to be too costly, maybe by limiting the number of peers we talk to) - partial synchronization only == data distributed over peers """ use GenServer require Logger alias SData.MerkleSearchTree, as: MST defmodule Manifest do defstruct [:channel] end defimpl Shard.Manifest, for: Manifest do def start(m) do DynamicSupervisor.start_child(Shard.DynamicSupervisor, {SApp.Chat, m}) end end @doc """ Start a process that connects to a given channel """ def start_link(manifest) do GenServer.start_link(__MODULE__, manifest) end @doc """ Initialize channel process. """ def init(manifest) do %Manifest{channel: channel} = manifest id = SData.term_hash manifest case Shard.Manager.register(id, manifest, self()) do :ok -> Shard.Manager.dispatch_to(id, nil, self()) {:ok, page_store} = SApp.PageStore.start_link(id, :page_store) root = Shard.Manager.load_state id root = cond do root == nil -> nil GenServer.call(page_store, {:have_rec, root}) -> root true -> Logger.warn "Not all pages for saved root were saved, restarting from an empty state!" nil end mst = %MST{store: %SApp.PageStore{pid: page_store}, cmp: &msg_cmp/2, root: root} group = %SNet.PubShardGroup{id: id} SNet.Group.init_lookup(group, self()) {:ok, %{channel: channel, id: id, group: group, manifest: manifest, page_store: page_store, mst: mst, subs: MapSet.new, } } :redundant -> exit(:redundant) end end def find_proc(chan) do manifest = %Manifest{channel: chan} id = SData.term_hash manifest Shard.Manager.find_proc id end @doc """ Implementation of the :manifest call that returns the chat room's manifest """ def handle_call(:manifest, _from, state) do {:reply, state.manifest, state} end def handle_call({:read_history, top_bound, num}, _from, state) do ret = MST.last(state.mst, top_bound, num) {:reply, ret, state} end @doc """ Implementation of the :chat_send handler. This is the main handler that is used to send a message to the chat room. Puts the message in the store and syncs with all connected peers. """ def handle_cast({:chat_send, pk, msg}, state) do msgbin = SData.term_bin {(System.os_time :seconds), msg} {:ok, sign} = Shard.Keys.sign_detached(pk, msgbin) msgitem = {pk, msgbin, sign} prev_root = state.mst.root mst = MST.insert(state.mst, msgitem) state = %{state | mst: mst} Shard.Manager.save_state(state.id, mst.root) for pid <- state.subs do if Process.alive?(pid) do send(pid, {:chat_send, state.channel, msgitem}) end end notif = {:append, prev_root, msgitem, mst.root} SNet.Group.broadcast(state.group, notif) {:noreply, state} end @doc """ Implementation of the :interested handler, this is called when a peer we are connected to asks to recieve data for this channel. """ def handle_cast({:interested, conn_pid, auth}, state) do if SNet.Group.in_group?(state.group, conn_pid, auth) do Shard.Manager.send_pid(conn_pid, {state.id, nil, {:root, state.mst.root}}) end {:noreply, state} end def handle_cast({:subscribe, pid}, state) do Process.monitor(pid) new_subs = MapSet.put(state.subs, pid) {:noreply, %{ state | subs: new_subs }} end @doc """ Implementation of the :msg handler, which is the main handler for messages comming from other peers concerning this chat room. Messages are: - `{:get, start}`: get some messages starting at a given Merkle hash - `{:info, start, list, rest}`: put some messages and informs of the Merkle hash of the store of older messages. """ def handle_cast({:msg, conn_pid, auth, _shard_id, nil, msg}, state) do if not SNet.Group.in_group?(state.group, conn_pid, auth) do # Ignore message {:noreply, state} else state = case msg do {:get_manifest} -> Shard.Manager.send_pid(conn_pid, {state.id, nil, {:manifest, state.manifest}}) state {:append, prev_root, msgitem, new_root} -> # Append message: one single mesage has arrived if new_root == state.mst.root do # We already have the message, do nothing state else # Try adding the message {pk, bin, sign} = msgitem if Shard.Keys.verify(pk, bin, sign) == :ok do if prev_root == state.mst.root do # Only one new message, insert it directly mst2 = MST.insert(state.mst, msgitem) if mst2.root == new_root do state = %{state | mst: mst2} GenServer.cast(state.page_store, {:set_roots, [mst2.root]}) Shard.Manager.save_state(state.id, mst2.root) msg_callback(state, msgitem) state else Logger.warn("Invalid new root after inserting same message item!") state end else # Not a simple one-insertion transition, look at the whole tree init_merge(state, new_root, conn_pid) end else Logger.warn("Received message with invalid signature") state end end {:root, new_root} -> if new_root == state.mst.root do # already up to date, ignore state else init_merge(state, new_root, conn_pid) end x -> Logger.info("Unhandled message: #{inspect x}") state end {:noreply, state} end end defp init_merge(state, new_root, source_peer_pid) do if new_root == nil do state else # TODO: make the merge asynchronous Logger.info("Starting merge for #{inspect state.manifest}, merging root: #{new_root|>Base.encode16}") prev_last = for {x, true} <- MST.last(state.mst, nil, 100), into: MapSet.new, do: x mgmst = %{state.mst | root: new_root} mgmst = put_in(mgmst.store.prefer_ask, [source_peer_pid]) mst = MST.merge(state.mst, mgmst) new = for {x, true} <- MST.last(mst, nil, 100), not MapSet.member?(prev_last, x) do x end correct = for x <- new do {pk, bin, sign} = x Shard.Keys.verify(pk, bin, sign) end if Enum.all? correct do for x <- new do msg_callback(state, x) end GenServer.cast(state.page_store, {:set_roots, [mst.root]}) Shard.Manager.save_state(state.id, mst.root) %{state | mst: mst} else Logger.warn("Incorrect signatures somewhere while merging, dropping merged data") state end end end def handle_info({:DOWN, _ref, :process, pid, _reason}, state) do new_subs = MapSet.delete(state.subs, pid) {:noreply, %{ state | subs: new_subs }} end defp msg_callback(state, {pk, msgbin, sign}) do for pid <- state.subs do if Process.alive?(pid) do send(pid, {:chat_recv, state.channel, {pk, msgbin, sign}}) end end end def msg_cmp({pk1, msgbin1, _sign1}, {pk2, msgbin2, _sign2}) do {ts1, msg1} = SData.term_unbin msgbin1 {ts2, msg2} = SData.term_unbin msgbin2 cond do ts1 > ts2 -> :after ts1 < ts2 -> :before pk1 > pk2 -> :after pk1 < pk2 -> :before msg1 > msg2 -> :after msg1 < msg2 -> :before true -> :duplicate end end end