exws_channels

defmodule ExWsChannels do
	@sharder Application.compile_env(:exws_channels, :sharder, ExWsChannels)

	def join(channel, {pid, socket}, meta \\ nil) do
		channel = sharded(channel, pid)
		user_data = {pid, channel}
		if match?([], :ets.match(:exws_channels_users, user_data)) do
			:ets.insert(:exws_channels_users, user_data)
			:ets.insert(:exws_channels, {channel, pid, socket, meta})
		end
	end

	def leave(channel, {pid, _socket}) do
		channel = sharded(channel, pid)
		:ets.match_delete(:exws_channels_users, {pid, channel})
		:ets.match_delete(:exws_channels, {channel, pid, :_, :_})
	end

	def leave({pid, _socket}) do
		Enum.each(:ets.lookup(:exws_channels_users, pid), fn {_pid, channel} ->
			:ets.match_delete(:exws_channels, {channel, pid, :_, :_})
		end)
		:ets.delete(:exws_channels_users, pid)
	end

	def member?(channel, {pid, _socket}) do
		channel = sharded(channel, pid)
		# ets.member doesn't work like I think it should for duplicate_bags??
		:ets.match(:exws_channels_users, {pid, channel}) == [[]]
	end

	def broadcast(channel, message, opts \\ []) do
		shards = @sharder.shards(channel) - 1
		Enum.each(0..shards, fn shard ->
			:poolboy.transaction(:exws_channels_writer, fn pid ->
				GenServer.cast(pid, {:broadcast, {channel, shard}, message})
			end, opts[:timeout] || 30_000)
		end)
	end

	def channels({pid, _socket}) do
		channels = :ets.lookup(:exws_channels_users, pid)
		Enum.reduce(channels, [], fn
			{_pid, {channel, _shard}}, acc -> [channel | acc]
			{_pid, channel}, acc -> [channel | acc]
		end)
	end

	def reduce(channel, acc, fun) do
		shards = @sharder.shards(channel) - 1
		Enum.reduce(0..shards, acc, fn shard, acc ->
			users = :ets.lookup(:exws_channels, {channel, shard})
			Enum.reduce(users, acc, fn {_channel, pid, socket, meta}, acc ->
				fun.({pid, socket, meta}, acc)
			end)
		end)
	end

	# Channels can be sharded. Sharding allows concurrent writers to
	# broadcast to users. When sharded, the internal channel name (
	# which is the key in our ETS, becomes {channel, H(pid) % SHARD_COUNT}).
	defp sharded(channel, pid) do
		case @sharder.shards(channel) do
			1 -> {channel, 0} # can avoid the phash2 here
			shards -> {channel, rem(:erlang.phash2(pid), shards)}
		end
	end

	# default implementation
	@doc false
	def shards(_), do: 1
end