defmodule Ash.Notifier.PubSub do
require Logger
@publish %Spark.Dsl.Entity{
name: :publish,
target: Ash.Notifier.PubSub.Publication,
describe: "Configure a given action to publish its results over a given topic.",
examples: [
"publish :create, \"created\"",
"""
publish :assign, "assigned"
"""
],
schema: Ash.Notifier.PubSub.Publication.schema(),
args: [:action, :topic]
}
@publish_all %Spark.Dsl.Entity{
name: :publish_all,
target: Ash.Notifier.PubSub.Publication,
describe: """
Works the same as `publish`, except that it takes a type and publishes all actions of that type.
""",
examples: [
"publish_all :create, \"created\""
],
schema: Ash.Notifier.PubSub.Publication.publish_all_schema(),
args: [:type, :topic]
}
@pub_sub %Spark.Dsl.Section{
name: :pub_sub,
describe: """
A section for configuring how resource actions are published over pubsub
""",
examples: [
"""
pub_sub do
module MyEndpoint
prefix "post"
publish :destroy, ["destroyed", :id]
publish :update, ["updated", :name], event: "name_change"
publish_all :create, "created"
end
"""
],
entities: [
@publish,
@publish_all
],
no_depend_modules: [:module],
schema: [
module: [
type: :atom,
doc: "The module to call `broadcast/3` on e.g module.broadcast(topic, event, message).",
required: true
],
prefix: [
type: :string,
doc:
"A prefix for all pubsub messages, e.g `users`. A message with `created` would be published as `users:created`"
],
delimiter: [
type: :string,
doc: "A delimiter for building topics. Default is a colon (:)"
],
filter: [
type: {:fun, 1},
doc:
"A filter for notifications. Receives a notification, and ignores it if the function returns a falsy value. Both this and filters on specific publications must return a truthy value for a notification to be emitted."
],
transform: [
type: {:fun, 1},
doc:
"A transformer for notifications. Specific transformers on each publication *override* this option"
],
broadcast_type: [
type: {:one_of, [:notification, :phoenix_broadcast, :broadcast]},
default: :notification,
doc: """
What shape the event payloads will be in. See
"""
],
name: [
type: :atom,
doc: "A named pub sub to pass as the first argument to broadcast."
]
]
}
@sections [@pub_sub]
@moduledoc """
A builtin notifier to help you publish events over any kind of pub-sub tooling.
This is plug and play with `Phoenix.PubSub`, but could be used with any pubsub system.
You configure a module that defines a `broadcast/3` function, and then add some "publications"
which configure under what conditions an event should be sent and what the topic should be.
## Example
```elixir
defmodule MyApp.User do
use Ash.Resource,
# ...
notifiers: [Ash.Notifier.PubSub]
# ...
pub_sub do
module MyAppWeb.Endpoint
prefix "user"
publish :update, ["updated", :_pkey]
end
end
```
## Debugging PubSub
It can be quite frustrating when setting up pub_sub when everything appears to be set up properly, but
you aren't receiving events. This usually means some kind of mismatch between the event names produced
by the resource/config of your publications, and you can use the following flag to display debug
information about all pub sub events.
```elixir
config :ash, :pub_sub, debug?: true
```
## Topic Templates
Often you want to include some piece of data in the thing being changed, like the `:id` attribute. This
is done by providing a list as the topic, and using atoms which will be replaced by their corresponding
values. They will ultimately be joined with `:`.
For example:
```elixir
prefix "user"
publish :create, ["created", :user_id]
```
This might publish a message to "user:created:1" for example.
For updates, if the field in the template is being changed, a message is sent
to *both* values. So if you change `user 1` to `user 2`, the same message would
be published to `user:updated:1` and `user:updated:2`. If there are multiple
attributes in the template, and they are all being changed, a message is sent for
every combination of substitutions.
## Important
If the previous value was `nil` or the field was not selected on the data passed into the action, then a
notification is not sent for the previous value.
If the new value is `nil` then a notification is not sent for the new value.
## Template parts
Templates may contain lists, in which case all combinations of values in the list will be used. Add
`nil` to the list if you want to produce a pattern where that entry is omitted.
The atom `:_tenant` may be used. If the changeset has a tenant set on it, that
value will be used, otherwise that combination of values is ignored.
The atom `:_pkey` may be used. It will be a stringified, concatenation of the primary key fields,
or just the primary key if there is only one primary key field.
The atom `nil` may be used. It only makes sense to use it in the context of a list of alternatives,
and adds a pattern where that part is skipped.
```elixir
publish :updated, [[:team_id, :_tenant], "updated", [:id, nil]]
```
Would produce the following messages, given a `team_id` of 1, a `tenant` of `org_1`, and an `id` of `50`:
```elixir
"1:updated:50"
"1:updated"
"org_1:updated:50"
"org_1:updated"
```
## Custom Delimiters
It's possible to change the default delimiter used when generating topics. This is useful when working with message brokers
like RabbitMQ, which rely on a different set of delimiters for routing.
```elixir
pub_sub do
delimiter "."
end
```
## Named Pubsub modules
If you are using a phoenix `Endpoint` module for pubsub then this is unnecessary. If you want to use a custom pub sub started
with something like `{Phoenix.PubSub, name: MyName}`, then you can provide `MyName` to here.
## Broadcast Types
Configured with `broadcast_type`.
- `:notification` just sends the notification
- `:phoenix_broadcast` sends a `%Phoenix.Socket.Broadcast{}` (see above)
- `:broadcast` sends `%{topic: (topic), event: (event), payload: (notification)}`
"""
use Spark.Dsl.Extension,
sections: @sections,
verifiers: [Ash.Notifier.PubSub.Verifiers.VerifyActionNames]
use Ash.Notifier
alias Ash.Notifier.PubSub.Info
@doc false
def notify(%Ash.Notifier.Notification{resource: resource} = notification) do
filter = Ash.Notifier.PubSub.Info.filter(resource)
resource
|> Ash.Notifier.PubSub.Info.publications()
|> Enum.filter(
&(matches?(&1, notification.action) &&
(is_nil(&1.filter) || &1.filter.(notification)) &&
(is_nil(filter) || filter.(notification)))
)
|> Enum.each(&publish_notification(&1, notification))
end
@doc false
def requires_original_data?(resource, action) do
resource
|> Ash.Notifier.PubSub.Info.publications()
|> Enum.filter(&(&1.previous_values? && matches?(&1, action)))
|> Enum.flat_map(fn publish ->
publish.topic
|> List.flatten()
|> Enum.filter(&is_atom/1)
end)
|> Enum.any?(&Ash.Resource.Info.attribute(resource, &1))
end
defp publish_notification(publish, notification) do
debug? = Application.get_env(:ash, :pub_sub)[:debug?] || false
event = publish.event || to_string(notification.action.name)
prefix = Ash.Notifier.PubSub.Info.prefix(notification.resource) || ""
delimiter = Info.delimiter(notification.resource)
topics =
publish.topic
|> fill_template(notification, delimiter, publish.previous_values?)
|> Enum.map(fn topic ->
case {prefix, topic} do
{"", ""} -> ""
{prefix, ""} -> prefix
{"", topic} -> topic
{prefix, topic} -> "#{prefix}#{delimiter}#{topic}"
end
end)
if debug? do
Logger.debug("""
Broadcasting to topics #{inspect(topics)} via #{inspect(Ash.Notifier.PubSub.Info.module(notification.resource))}.broadcast
Notification:
#{inspect(notification)}
""")
end
transform = publish.transform || Ash.Notifier.PubSub.Info.transform(notification.resource)
value =
if transform do
transform.(notification)
else
notification
end
Enum.each(topics, fn topic ->
args =
case Ash.Notifier.PubSub.Info.name(notification.resource) do
nil ->
[topic, event, to_payload(topic, event, notification, value)]
pub_sub ->
payload = to_payload(topic, event, notification, value)
[pub_sub, topic, payload]
end
args =
case publish.dispatcher do
nil ->
args
dispatcher ->
args ++ dispatcher
end
apply(Ash.Notifier.PubSub.Info.module(notification.resource), :broadcast, args)
end)
end
def to_payload(topic, event, notification, value) do
case Ash.Notifier.PubSub.Info.broadcast_type(notification.resource) do
:phoenix_broadcast ->
%{
__struct__: Phoenix.Socket.Broadcast,
topic: topic,
event: event,
payload: value
}
:broadcast ->
%{
topic: topic,
event: event,
payload: value
}
:notification ->
value
end
end
defp fill_template(topic, _notification, _delimiter, _previous_values?) when is_binary(topic),
do: [topic]
defp fill_template(topic, notification, delimiter, previous_values?) do
topic
|> all_combinations_of_values(notification, notification.action.type, previous_values?)
|> Enum.map(&List.flatten/1)
|> Enum.map(&Enum.join(&1, delimiter))
|> Enum.uniq()
end
defp all_combinations_of_values(
items,
notification,
action_type,
_previous_values?,
trail \\ []
)
defp all_combinations_of_values([], _, _, _previous_values?, trail), do: [Enum.reverse(trail)]
defp all_combinations_of_values(
[nil | rest],
notification,
action_type,
previous_values?,
trail
) do
all_combinations_of_values(rest, notification, action_type, previous_values?, trail)
end
defp all_combinations_of_values(
[item | rest],
notification,
action_type,
previous_values?,
trail
)
when is_binary(item) do
all_combinations_of_values(rest, notification, action_type, previous_values?, [item | trail])
end
defp all_combinations_of_values(
[:_tenant | rest],
notification,
action_type,
previous_values?,
trail
) do
if notification.changeset.to_tenant do
all_combinations_of_values(rest, notification, action_type, previous_values?, [
notification.changeset.to_tenant | trail
])
else
[]
end
end
defp all_combinations_of_values([:_pkey | rest], notification, type, true, trail)
when type in [:update, :destroy] do
pkey = Ash.Resource.Info.primary_key(notification.changeset.resource)
pkey_value_before_change = Enum.map_join(pkey, "-", &Map.get(notification.changeset.data, &1))
pkey_value_after_change = Enum.map_join(pkey, "-", &Map.get(notification.data, &1))
[pkey_value_before_change, pkey_value_after_change]
|> Enum.uniq()
|> Enum.flat_map(fn possible_value ->
all_combinations_of_values(rest, notification, type, true, [
possible_value | trail
])
end)
end
defp all_combinations_of_values([:_pkey | rest], notification, type, _, trail)
when type in [:update, :destroy] do
pkey = Ash.Resource.Info.primary_key(notification.changeset.resource)
pkey_value_after_change = Enum.map_join(pkey, "-", &Map.get(notification.data, &1))
all_combinations_of_values(rest, notification, type, false, [pkey_value_after_change | trail])
end
defp all_combinations_of_values(
[:_pkey | rest],
notification,
action_type,
previous_values?,
trail
) do
pkey = Ash.Resource.Info.primary_key(notification.changeset.resource)
all_combinations_of_values(rest, notification, action_type, previous_values?, [
Enum.map_join(pkey, "-", &Map.get(notification.data, &1)) | trail
])
end
defp all_combinations_of_values([item | rest], notification, type, true, trail)
when is_atom(item) and type in [:update, :destroy] do
value_before_change = Map.get(notification.changeset.data, item)
value_after_change = Map.get(notification.data, item)
[value_before_change, value_after_change]
|> Enum.filter(&publishable_value?(&1, notification))
|> Enum.uniq()
|> Enum.flat_map(fn possible_value ->
all_combinations_of_values(rest, notification, type, true, [possible_value | trail])
end)
end
defp all_combinations_of_values([item | rest], notification, type, _, trail)
when is_atom(item) and type in [:update, :destroy] do
value_after_change = Map.get(notification.data, item)
if publishable_value?(value_after_change, notification) do
all_combinations_of_values(rest, notification, type, false, [value_after_change | trail])
else
[]
end
end
defp all_combinations_of_values(
[item | rest],
notification,
action_type,
previous_values?,
trail
)
when is_atom(item) do
value = Map.get(notification.data, item)
if publishable_value?(value, notification) do
all_combinations_of_values(rest, notification, action_type, previous_values?, [
value | trail
])
else
[]
end
end
defp all_combinations_of_values(
[item | rest],
notification,
action_type,
previous_values?,
trail
)
when is_list(item) do
Enum.flat_map(item, fn possible_value ->
all_combinations_of_values(
[possible_value | rest],
notification,
action_type,
previous_values?,
trail
)
end)
end
defp publishable_value?(nil, _notification), do: false
defp publishable_value?(
%Ash.NotLoaded{field: field},
%{topic: topic, resource: resource}
) do
Logger.warning(
"Not publishing notification `#{inspect(topic)}` for #{inspect(resource)} because `#{field}` is not loaded"
)
false
end
defp publishable_value?(%Ash.NotLoaded{field: field}, notification) do
Logger.warning(
"Not publishing notification for #{inspect(notification.resource)} because `#{field}` is not loaded"
)
false
end
defp publishable_value?(
%Ash.ForbiddenField{field: field},
%{topic: topic, resource: resource}
) do
Logger.warning(
"Not publishing notification `#{inspect(topic)}` for #{inspect(resource)} because `#{field}` is an `%Ash.ForbiddenField{}`"
)
false
end
defp publishable_value?(%Ash.ForbiddenField{field: field}, notification) do
Logger.warning(
"Not publishing notification for #{inspect(notification.resource)} because `#{field}` is an `%Ash.ForbiddenField{}`"
)
false
end
defp publishable_value?(_, _), do: true
defp matches?(%{action: action}, %{name: action}), do: true
defp matches?(%{type: type, except: except}, %{type: type, name: action}) do
action not in except
end
defp matches?(_, _), do: false
end
